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 5f3407d..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] @@ -58,6 +61,19 @@ const DefaultRequestTimeout* = 5.seconds # Finite fallback (issue #93) so a wedged handler can't hang a caller forever. +type FFITeardownProc*[T] = proc(lib: ptr T): Future[void] {.async.} + +proc ffiTeardownHook*[T](): var FFITeardownProc[T] = + ## Per-library teardown slot (one `{.global.}` per `T`), assigned at module init + ## by a non-empty `{.ffiDtor.}` and awaited by the FFI thread before it exits. + ## + ## A runtime slot, not an overloaded `ffiTeardown` resolved via `mixin`: the + ## constructor force-instantiates the FFI thread body before the dtor (declared + ## later in the source) is visible, so an overload would bind the no-op default + ## and the teardown would silently never run. + var hook {.global.}: FFITeardownProc[T] + hook + include ./event_thread include ./ffi_thread @@ -113,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 @@ -180,9 +197,14 @@ proc signalStop*[T](ctx: ptr FFIContext[T]): Result[void, string] = error "failed to signal eventQueueSignal in signalStop", error = error ok() -## Bound on how long clearContext waits for the FFI thread to exit before -## leaking ctx rather than hanging the caller. -const ThreadExitTimeout* = 1500.milliseconds +## Bound on how long stopAndJoinThreads waits for a worker thread to exit before +## leaking ctx rather than hanging the caller. Configurable because an async +## `{.ffiDtor.}` runs its teardown (e.g. `switch.stop()` over many live +## 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: "ffiThreadExitTimeoutMs".} = 1500 +const ThreadExitTimeout* = ThreadExitTimeoutMs.milliseconds proc stopAndJoinThreads*[T](ctx: ptr FFIContext[T]): Result[void, string] = ## On timeout, returns err and skips remaining joins (leaves threads live). diff --git a/ffi/ffi_thread.nim b/ffi/ffi_thread.nim index 9f6eac2..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(): @@ -210,4 +216,15 @@ proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} = except CatchableError as e: error "draining pending FFI requests on shutdown raised", error = e.msg + # In-flight requests drained; run the library's async shutdown (e.g. + # `switch.stop()`) on this event loop before the thread joins. Only if a + # `{.ffiDtor.}` registered a hook and a request populated `myLib`. Exceptions + # are logged, never propagated: the thread must still fire threadExitSignal. + let teardown = ffiTeardownHook[T]() + if not teardown.isNil() and not ctx.myLib.isNil(): + try: + await teardown(ctx.myLib) + except CatchableError as e: + error "library teardown raised on shutdown", error = e.msg + waitFor ffiRun(ctx) diff --git a/ffi/internal/ffi_macro.nim b/ffi/internal/ffi_macro.nim index 69be972..6b263b9 100644 --- a/ffi/internal/ffi_macro.nim +++ b/ffi/internal/ffi_macro.nim @@ -1575,26 +1575,38 @@ macro ffiCtor*(args: varargs[untyped]): untyped = return stmts macro ffiDtor*(args: varargs[untyped]): untyped = - ## Defines a C-exported destructor that tears down the FFIContext after the - ## body runs. + ## Defines a C-exported destructor that tears down the FFIContext. ## - ## The annotated proc must have exactly one parameter of the library type. - ## The body contains any library-level cleanup to run before context teardown. + ## The annotated proc must have exactly one parameter of the library type. It + ## may be sync (no return type) or async (`Future[void]`) — an async dtor can + ## `await` a graceful library shutdown (e.g. `switch.stop()`) whose futures + ## live on the FFI event loop. ## ## The wire format follows the library default and can be overridden with ## `{.ffiDtor: "abi = c".}` / `{.ffiDtor: "abi = cbor".}`. ## - ## Example: - ## proc waku_destroy*(w: Waku) {.ffiDtor.} = - ## w.cleanup() + ## Example (sync): + ## proc echo_destroy*(e: Echo) {.ffiDtor.} = + ## e.close() + ## + ## Example (async): + ## proc waku_destroy*(w: Waku): Future[void] {.ffiDtor.} = + ## await w.stop() ## ## The generated C-exported proc has the signature: ## int waku_destroy(void* ctx) ## - ## It extracts the library value from ctx, runs the body, then calls - ## destroyFFIContext to tear down the FFI thread and free the context. + ## A non-empty body is lifted into an async impl registered in the library's + ## `ffiTeardownHook` slot; the FFI thread awaits it on its own event loop, after + ## draining in-flight requests and just before it exits — so the body runs on + ## the worker thread, not the host (calling) thread. The C wrapper signals the + ## thread to stop and blocks (up to `ThreadExitTimeout`) until it, and the + ## teardown, finish, then frees the context. An empty/`discard` body registers + ## no hook. + ## ## Returns RET_OK on success, RET_ERR on failure (null/invalid ctx, or - ## destroyFFIContext failure). + ## destroyFFIContext failure — e.g. a teardown that outlasts ThreadExitTimeout, + ## which leaks the context rather than hanging the caller). requireBeforeGenBindings("`.ffiDtor.`") requireLibraryDeclared("`.ffiDtor.`") @@ -1612,6 +1624,18 @@ macro ffiDtor*(args: varargs[untyped]): untyped = let libParamName = formalParams[1][0] let libTypeName = formalParams[1][1] + # A dtor is sync (no return type) or async (`Future[void]`); reject anything + # else up front rather than emitting an obscure downstream error. + let retTypeNode = formalParams[0] + let retIsFutureVoid = + retTypeNode.kind == nnkBracketExpr and $retTypeNode[0] == "Future" and + retTypeNode.len == 2 and $retTypeNode[1] == "void" + if retTypeNode.kind != nnkEmpty and not retIsFutureVoid: + error( + "ffiDtor: proc must return nothing (sync) or Future[void] (async), got: " & + retTypeNode.repr + ) + let procNameStr = block: let raw = $procName if raw.endsWith("*"): @@ -1639,16 +1663,26 @@ macro ffiDtor*(args: varargs[untyped]): untyped = if ctx.isNil or cast[ptr FFIContext[`libTypeName`]](ctx)[].myLib.isNil: return RET_ERR - ffiBody.add quote do: - let `libParamName` = cast[ptr FFIContext[`libTypeName`]](ctx)[].myLib[] - let isNoop = bodyNode.kind == nnkEmpty or ( bodyNode.kind == nnkStmtList and bodyNode.len == 1 and bodyNode[0].kind == nnkDiscardStmt ) - if not isNoop: - ffiBody.add(bodyNode) + + # Lift the body into an async impl registered in the per-library + # `ffiTeardownHook`, which the FFI thread awaits at shutdown (see ffi_thread.nim + # and ffiTeardownHook's docstring). The C wrapper no longer runs the body. + let teardownImplName = genSym(nskProc, "ffiTeardownImpl") + let teardownRegistration = + if isNoop: + newEmptyNode() + else: + quote: + proc `teardownImplName`(lib: ptr `libTypeName`): Future[void] {.async.} = + let `libParamName` = lib[] + `bodyNode` + + ffiTeardownHook[`libTypeName`]() = `teardownImplName` let poolIdent = ident($libTypeName & "FFIPool") ffiBody.add quote do: @@ -1690,7 +1724,7 @@ macro ffiDtor*(args: varargs[untyped]): untyped = when not declared(`poolIdent`): var `poolIdent`: FFIContextPool[`libTypeName`] - let stmts = newStmtList(poolDecl, ffiProc) + let stmts = newStmtList(teardownRegistration, poolDecl, ffiProc) when defined(ffiDumpMacros): echo stmts.repr diff --git a/tests/unit/test_ffi_teardown.nim b/tests/unit/test_ffi_teardown.nim new file mode 100644 index 0000000..7778791 --- /dev/null +++ b/tests/unit/test_ffi_teardown.nim @@ -0,0 +1,86 @@ +import std/[atomics, os] +import unittest2 +import results +import ffi + +# Exercises the async `{.ffiDtor.}` teardown hook: a non-empty dtor body is +# lifted into an `ffiTeardown` overload the FFI thread awaits on its own event +# loop right before it exits. `destroyFFIContext` must block until that +# teardown finishes. + +type TeardownLib = object + +# declareLibrary emits an importc `libNimMain`; this test links as a plain +# exe, so stub it (mirrors test_ffi_context.nim). +{.emit: "void libteardownlibNimMain(void) {}".} + +declareLibrary("teardownlib", TeardownLib) + +var gTeardownRan: Atomic[bool] +var gTeardownThreadId: Atomic[int] + +type NoopConfig {.ffi.} = object + dummy: int + +proc teardownlib_create*( + config: NoopConfig +): Future[Result[TeardownLib, string]] {.ffiCtor.} = + return ok(TeardownLib()) + +proc teardownlib_destroy*(lib: TeardownLib): Future[void] {.ffiDtor.} = + ## Async teardown: sleeps on the FFI event loop, then records that it ran and + ## on which thread. If destroy didn't wait, `gTeardownRan` would still be false + ## when the test checks it. + await sleepAsync(200.milliseconds) + gTeardownThreadId.store(getThreadId()) + gTeardownRan.store(true) + +proc noopCallback( + retCode: cint, msg: ptr cchar, len: csize_t, userData: pointer +) {.cdecl, gcsafe, raises: [].} = + discard + +proc encodedPtr(bytes: var seq[byte]): ptr byte = + if bytes.len == 0: + nil + else: + cast[ptr byte](addr bytes[0]) + +proc createCtxWithLib(): ptr FFIContext[TeardownLib] = + ## Spins up a context via the ctor and waits until `myLib` is populated (set on + ## the FFI thread when the ctor request is dispatched), so teardown has a lib. + var cfg = cborEncode(TeardownlibCreateCtorReq(config: NoopConfig(dummy: 0))) + let ret = teardownlib_create(encodedPtr(cfg), cfg.len.csize_t, noopCallback, nil) + if ret.isNil(): + return nil + let ctx = cast[ptr FFIContext[TeardownLib]](ret) + var tries = 0 + while ctx[].myLib.isNil() and tries < 500: + os.sleep(5) + inc tries + ctx + +suite "async {.ffiDtor.} teardown hook": + test "destroy blocks until the async teardown body completes": + let ctx = createCtxWithLib() + check not ctx.isNil() + check not ctx[].myLib.isNil() + + gTeardownRan.store(false) + gTeardownThreadId.store(0) + let callerTid = getThreadId() + + check TeardownlibFFIPool.destroyFFIContext(ctx).isOk() + + # If destroy returned before awaiting the teardown, this would be false. + check gTeardownRan.load() + # And it must have run on the FFI thread, not the caller's. + check gTeardownThreadId.load() != 0 + check gTeardownThreadId.load() != callerTid + + test "teardown runs exactly once per context": + gTeardownRan.store(false) + let ctx = createCtxWithLib() + check not ctx.isNil() + check TeardownlibFFIPool.destroyFFIContext(ctx).isOk() + check gTeardownRan.load()