Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: update num-msgs archive metrics every minute and not only at the beginning #2287

Merged
merged 2 commits into from
Dec 14, 2023
Merged

fix: update num-msgs archive metrics every minute and not only at the beginning #2287

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8d5efa5
bug fix: update num-msgs archive metrics every minute and not only at…
Ivansete-status Dec 13, 2023
4a8c5b8
Merge branch 'master' into fix-report-archive-metrics
Ivansete-status Dec 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 0 additions & 8 deletions waku/node/waku_node.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -751,14 +751,6 @@ proc mountArchive*(node: WakuNode,
return err("error in mountArchive: " & wakuArchiveRes.error)

node.wakuArchive = wakuArchiveRes.get()

try:
let reportMetricRes = waitFor node.wakuArchive.reportStoredMessagesMetric()
if reportMetricRes.isErr():
return err("error in mountArchive: " & reportMetricRes.error)
except CatchableError:
return err("exception in mountArchive: " & getCurrentExceptionMsg())

asyncSpawn node.wakuArchive.start()
return ok()

Expand Down
23 changes: 16 additions & 7 deletions waku/waku_archive/archive.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ type
validator: MessageValidator
retentionPolicy: RetentionPolicy
retPolicyFut: Future[Result[void, string]] ## retention policy cancelable future
retMetricsRepFut: Future[Result[void, string]] ## metrics reporting cancelable future

proc new*(T: type WakuArchive,
driver: ArchiveDriver,
Expand Down Expand Up @@ -212,21 +213,29 @@ proc loopApplyRetentionPolicy*(w: WakuArchive):

return ok()

proc reportStoredMessagesMetric*(w: WakuArchive):
Future[Result[void, string]] {.async.} =
# Metrics reporting
const WakuArchiveDefaultMetricsReportInterval* = chronos.minutes(1)

proc loopReportStoredMessagesMetric*(w: WakuArchive):
Future[Result[void, string]] {.async.} =
if w.driver.isNil():
return err("driver is Nil in reportStoredMessagesMetric")
return err("driver is Nil in loopReportStoredMessagesMetric")

let resCount = await w.driver.getMessagesCount()
if resCount.isErr():
return err("failed to get messages count: " & resCount.error)
while true:
AlejandroCabeza marked this conversation as resolved.
Show resolved Hide resolved
let resCount = await w.driver.getMessagesCount()
if resCount.isErr():
return err("loopReportStoredMessagesMetric failed to get messages count: " & resCount.error)

waku_archive_messages.set(resCount.value, labelValues = ["stored"])
waku_archive_messages.set(resCount.value, labelValues = ["stored"])
await sleepAsync(WakuArchiveDefaultMetricsReportInterval)

return ok()

proc start*(self: WakuArchive) {.async.} =
## TODO: better control the Result in case of error. Now it is ignored
self.retPolicyFut = self.loopApplyRetentionPolicy()
self.retMetricsRepFut = self.loopReportStoredMessagesMetric()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder, how are these started? Don't they require the await keyword at some point?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good question!
This can be considered a "normal" proc that starts an infinite loop inside. This proc is marked with the {.async.} pragma, which in turn, makes the proc return a Future[void], implicitly. And this returned Future[void] needs to be either attended or discarded.

afaik, there are two ways to attend a "future": keep a reference of it or await it immediately. And in this case, we keep a reference that is used to cancel the future returned. In this case, the returned future is directly linked with the internal await sleepAsync(.. call, which in turn conforms a cancellation point.

Therefore, when we are cancelling that future, we are forcing the code to break and goto to end that proc call, and there is no need to perform an explicit break of the infinite loop. Notice that this "cancellation point" feature is given by our nim-chronos library.

I think this should be the standard way to proceed in such infinite periodic tasks.

Nevertheless, @arnetheduck can shed more light on this and provide a more accurate reasoning.

note: in case I didn't suggested before, this is a great explanation of how async works in Nim: https://peterme.net/asynchronous-programming-in-nim.html (I revisit it now and then ;P)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, if I understood correctly, this line self.retMetricsRepFut = self.loopReportStoredMessagesMetric() implicitly runs a waitFor?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, if I understood correctly, this line self.retMetricsRepFut = self.loopReportStoredMessagesMetric() implicitly runs a waitFor?

self.retMetricsRepFut = self.loopReportStoredMessagesMetric() <- This line should be understood as a way of starting an async and endless task.

waitFor is a mechanism by which the thread gets blocked until a certain Future is completed. waitFor should only be used in the upper layers of every app to prevent deadlocks, and by "upper layer" I mean the procs that are at the same level as the "main" module.

Looking at the waitFor implementation it can be seen that it simply waits synchronously (doesn't have the {.async.} pragma) for the future to complete while calling poll(), which is responsible for the dispatcher to progress and attend all the timers, network events, and async tasks: https://github.com/status-im/Nim/blob/3c9b68dc157804885b14a1984efc25e8b7cc861d/lib/pure/asyncdispatch.nim#L1958-L1963

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, okay then. The piece I'm missing is:
Why, if just running self.loopReportStoredMessagesMetric() starts the async task, does the compiler tell you you need to (discard) await it?
Or does it only start the async task when the rvalue of that is assigned to a Future variable?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Or does it only start the async task when the rvalue of that is assigned to a Future variable?

Good question! I think the answer is "yes". I think it only starts when the future is consumed either with await or assigned. Is just an assumption, I don't know 100% sure.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That would actually make sense with all other parts... I was making some assumptions from being used to Python's async 😅


proc stop*(self: WakuArchive) {.async.} =
self.retPolicyFut.cancel()
self.retMetricsRepFut.cancel()
Loading