mirror of
https://github.com/logos-messaging/nim-ffi.git
synced 2026-07-16 21:10:19 +00:00
feat(ffi): async teardown hook for {.ffiDtor.} (#115)
This commit is contained in:
parent
5eaa799b7d
commit
3e57751e3a
@ -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.
|
||||
|
||||
@ -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=<ms>`.
|
||||
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).
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
86
tests/unit/test_ffi_teardown.nim
Normal file
86
tests/unit/test_ffi_teardown.nim
Normal file
@ -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 `lib<name>NimMain`; 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()
|
||||
Loading…
x
Reference in New Issue
Block a user