diff --git a/ffi/event_thread.nim b/ffi/event_thread.nim index 21cc6c2..21ccdb9 100644 --- a/ffi/event_thread.nim +++ b/ffi/event_thread.nim @@ -116,20 +116,24 @@ proc eventRun[T](ctx: ptr FFIContext[T]) {.async.} = var hb = HeartbeatMonitor.init(ctx) var notifiedStuck = false # latched forever — eventQueueStuck is sticky terminal. - while ctx.running.load(): + # Keep draining after `running` flips false until the FFI thread has exited, so + # events emitted by an async {.ffiDtor.} teardown are still dispatched. + while ctx.running.load() or not ctx.ffiThreadExited.load(): # Wake on enqueue or tick — whichever first. discard await ctx.eventQueueSignal.wait().withTimeout(EventThreadTickInterval) ctx.drainEventQueue() - # Fire after drain so reg.lock is free — FFI-thread would deadlock here. - if not notifiedStuck and ctx.eventQueueStuck.load(): - onNotResponding(ctx) - notifiedStuck = true + # Liveness only applies while running; skip it during the teardown drain. + if ctx.running.load(): + # Fire after drain so reg.lock is free — FFI-thread would deadlock here. + if not notifiedStuck and ctx.eventQueueStuck.load(): + onNotResponding(ctx) + notifiedStuck = true + hb.check(ctx) - if not ctx.running.load(): - break - hb.check(ctx) + # Catch anything enqueued between the last drain and the FFI thread's exit. + ctx.drainEventQueue() proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} = ## Drains the event queue and runs the FFI-thread heartbeat check. diff --git a/ffi/ffi_context.nim b/ffi/ffi_context.nim index 27ed961..751e63c 100644 --- a/ffi/ffi_context.nim +++ b/ffi/ffi_context.nim @@ -37,6 +37,9 @@ type FFIContext*[T] = object ffiHeartbeat*: Atomic[int64] # advanced each FFI-thread loop; event thread reads for liveness eventQueueStuck*: Atomic[bool] # sticky overflow flag + ffiThreadExited*: Atomic[bool] + # set once the FFI thread (including any async {.ffiDtor.} teardown) is done; + # keeps the event thread draining until then so teardown-emitted events land running: Atomic[bool] # To control when the threads are running registeredRequests: ptr Table[cstring, FFIRequestProc] requestTimeouts: ptr Table[cstring, int] @@ -126,6 +129,7 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] = initEventQueue(ctx[].eventQueue) ctx.ffiHeartbeat.store(0) ctx.eventQueueStuck.store(false) + ctx.ffiThreadExited.store(false) ctx.defaultRequestTimeout = DefaultRequestTimeout var success = false @@ -199,7 +203,7 @@ proc signalStop*[T](ctx: ptr FFIContext[T]): Result[void, string] = ## connections) on the FFI thread before it exits — a graceful shutdown can ## outlast the default, and being cut short leaks the context instead of waiting. ## Override at compile time with `-d:ffiThreadExitTimeoutMs=`. -const ThreadExitTimeoutMs* {.intdefine.} = 1500 +const ThreadExitTimeoutMs* {.intdefine: "ffiThreadExitTimeoutMs".} = 1500 const ThreadExitTimeout* = ThreadExitTimeoutMs.milliseconds proc stopAndJoinThreads*[T](ctx: ptr FFIContext[T]): Result[void, string] = diff --git a/ffi/ffi_thread.nim b/ffi/ffi_thread.nim index c4af476..f98834b 100644 --- a/ffi/ffi_thread.nim +++ b/ffi/ffi_thread.nim @@ -150,6 +150,12 @@ proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} = onFFIThread = false # Free handle refs on the FFI thread that allocated them (refc heap is thread-local). ctx[].handles.releaseAll() + # Teardown has run and no more events will be emitted from this thread; let + # the event thread stop draining and exit. Wake it so it notices without + # waiting a full tick. + ctx.ffiThreadExited.store(true) + ctx.eventQueueSignal.fireSync().isOkOr: + error "failed to wake event thread on FFI thread exit", err = error # Unblocks destroyFFIContext's bounded wait so cleanup can proceed. let fireRes = ctx.threadExitSignal.fireSync() if fireRes.isErr(): diff --git a/ffi/internal/ffi_macro.nim b/ffi/internal/ffi_macro.nim index e440e71..6b263b9 100644 --- a/ffi/internal/ffi_macro.nim +++ b/ffi/internal/ffi_macro.nim @@ -1585,6 +1585,10 @@ macro ffiDtor*(args: varargs[untyped]): untyped = ## The wire format follows the library default and can be overridden with ## `{.ffiDtor: "abi = c".}` / `{.ffiDtor: "abi = cbor".}`. ## + ## Example (sync): + ## proc echo_destroy*(e: Echo) {.ffiDtor.} = + ## e.close() + ## ## Example (async): ## proc waku_destroy*(w: Waku): Future[void] {.ffiDtor.} = ## await w.stop()