chore: libwaku - better error handling and better waku thread destroy handling (#3167)

This commit is contained in:
Ivan FB 2024-11-08 14:59:02 +07:00 committed by GitHub
parent 3cb8ebdd8f
commit 294dd03c45
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 37 additions and 25 deletions

View File

@ -37,8 +37,7 @@ declarePublicGauge networkmonitor_peer_user_agents,
declarePublicHistogram networkmonitor_peer_ping, declarePublicHistogram networkmonitor_peer_ping,
"Histogram tracking ping durations for discovered peers", "Histogram tracking ping durations for discovered peers",
buckets = buckets = [10.0, 20.0, 50.0, 100.0, 200.0, 300.0, 500.0, 800.0, 1000.0, 2000.0, Inf]
[10.0, 20.0, 50.0, 100.0, 200.0, 300.0, 500.0, 800.0, 1000.0, 2000.0, Inf]
declarePublicGauge networkmonitor_peer_count, declarePublicGauge networkmonitor_peer_count,
"Number of discovered peers", labels = ["connected"] "Number of discovered peers", labels = ["connected"]

View File

@ -168,7 +168,7 @@ proc waku_destroy(
): cint {.dynlib, exportc.} = ): cint {.dynlib, exportc.} =
checkLibwakuParams(ctx, callback, userData) checkLibwakuParams(ctx, callback, userData)
waku_thread.stopWakuThread(ctx).handleRes(callback, userData) waku_thread.destroyWakuThread(ctx).handleRes(callback, userData)
proc waku_version( proc waku_version(
ctx: ptr WakuContext, callback: WakuCallBack, userData: pointer ctx: ptr WakuContext, callback: WakuCallBack, userData: pointer

View File

@ -18,36 +18,45 @@ type WakuContext* = object
userData*: pointer userData*: pointer
eventCallback*: pointer eventCallback*: pointer
eventUserdata*: pointer eventUserdata*: pointer
running: Atomic[bool] # To control when the thread is running
const git_version* {.strdefine.} = "n/a" const git_version* {.strdefine.} = "n/a"
const versionString = "version / git commit hash: " & waku.git_version const versionString = "version / git commit hash: " & waku.git_version
# To control when the thread is running
# TODO: this should be part of the context so multiple instances can be executed
var running: Atomic[bool]
proc runWaku(ctx: ptr WakuContext) {.async.} = proc runWaku(ctx: ptr WakuContext) {.async.} =
## This is the worker body. This runs the Waku node ## This is the worker body. This runs the Waku node
## and attends library user requests (stop, connect_to, etc.) ## and attends library user requests (stop, connect_to, etc.)
info "Starting Waku", version = versionString
var waku: Waku var waku: Waku
while running.load == true: while true:
await ctx.reqSignal.wait() await ctx.reqSignal.wait()
# Trying to get a request from the libwaku main thread if ctx.running.load == false:
break
## Trying to get a request from the libwaku requestor thread
var request: ptr InterThreadRequest var request: ptr InterThreadRequest
let recvOk = ctx.reqChannel.tryRecv(request) let recvOk = ctx.reqChannel.tryRecv(request)
if recvOk == true: if not recvOk:
error "waku thread could not receive a request"
continue
## Handle the request
let resultResponse = waitFor InterThreadRequest.process(request, addr waku) let resultResponse = waitFor InterThreadRequest.process(request, addr waku)
## Converting a `Result` into a thread-safe transferable response type ## Converting a `Result` into a thread-safe transferable response type
let threadSafeResp = InterThreadResponse.createShared(resultResponse) let threadSafeResp = InterThreadResponse.createShared(resultResponse)
## The error-handling is performed in the main thread ## Send the response back to the thread that sent the request
discard ctx.respChannel.trySend(threadSafeResp) let sentOk = ctx.respChannel.trySend(threadSafeResp)
discard ctx.respSignal.fireSync() if not sentOk:
error "could not send a request to the requester thread",
original_request = $request[]
let fireRes = ctx.respSignal.fireSync()
if fireRes.isErr():
error "could not fireSync back to requester thread", error = fireRes.error
proc run(ctx: ptr WakuContext) {.thread.} = proc run(ctx: ptr WakuContext) {.thread.} =
## Launch waku worker ## Launch waku worker
@ -62,7 +71,7 @@ proc createWakuThread*(): Result[ptr WakuContext, string] =
ctx.respSignal = ThreadSignalPtr.new().valueOr: ctx.respSignal = ThreadSignalPtr.new().valueOr:
return err("couldn't create respSignal ThreadSignalPtr") return err("couldn't create respSignal ThreadSignalPtr")
running.store(true) ctx.running.store(true)
try: try:
createThread(ctx.thread, run, ctx) createThread(ctx.thread, run, ctx)
@ -74,15 +83,19 @@ proc createWakuThread*(): Result[ptr WakuContext, string] =
return ok(ctx) return ok(ctx)
proc stopWakuThread*(ctx: ptr WakuContext): Result[void, string] = proc destroyWakuThread*(ctx: ptr WakuContext): Result[void, string] =
running.store(false) ctx.running.store(false)
let fireRes = ctx.reqSignal.fireSync()
if fireRes.isErr(): let signaledOnTime = ctx.reqSignal.fireSync().valueOr:
return err("error in stopWakuThread: " & $fireRes.error) return err("error in destroyWakuThread: " & $error)
discard ctx.reqSignal.close() if not signaledOnTime:
discard ctx.respSignal.close() return err("failed to signal reqSignal on time in destroyWakuThread")
joinThread(ctx.thread) joinThread(ctx.thread)
?ctx.reqSignal.close()
?ctx.respSignal.close()
freeShared(ctx) freeShared(ctx)
return ok() return ok()
proc sendRequestToWakuThread*( proc sendRequestToWakuThread*(