From dda79fab343e46e694c9bad8388aab33d2c15d1c Mon Sep 17 00:00:00 2001 From: Ivan FB Date: Fri, 19 Jun 2026 23:27:08 +0200 Subject: [PATCH] feat: recycle pooled FFI contexts; compile-time request ids MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two foreign-host concurrency fixes for refc, both needed so a Go host can hammer the FFI under load without corrupting Nim's per-thread GC. 1. Compile-time request ids. ffiNewReq / the method + ctor wrappers built the request id with `$T` at runtime, allocating a Nim GC string on the foreign caller's (often transient) thread. Emit a `cstring` literal of the type name instead — no allocation on the caller thread. 2. Recycle pooled contexts instead of destroy/recreate. Restores the release/v0.1 model that v0.2 dropped: a pool slot's worker + event threads and signal fds are built once and reused. The ffiDtor now requests a synchronous recycle (drain in-flight handlers, free the lib, clear listeners, release the slot) on the FFI thread, keeping the threads alive; createFFIContext reuses an initialised slot. Without this every create/destroy churned ~6 signal fds, so fd numbers climbed past FD_SETSIZE (1024) and ThreadSignalPtr.waitSync's select() failed with EINVAL under create/destroy load. Adds CtxLifecycle (Active/RecyclePending/Recycling), ctx-level inUse/tryClaim/release/markAsActive, requestRecycle (waits on a new recycleDoneSignal), freeLib (refc GC_unref / orc =destroy of ctor-owned libs), recycleContext, FFIEventRegistry.clearListeners, and roots the ctor-stored ref lib under refc (GC_ref, balanced in freeLib). Co-Authored-By: Claude Opus 4.8 --- ffi/ffi_context.nim | 70 +++++++++++++++++++++++++++++++++++-- ffi/ffi_context_pool.nim | 71 ++++++++++++++++++++++++-------------- ffi/ffi_events.nim | 7 ++++ ffi/ffi_thread.nim | 61 ++++++++++++++++++++++++++++++++ ffi/internal/ffi_macro.nim | 38 ++++++++++++++++---- 5 files changed, 212 insertions(+), 35 deletions(-) diff --git a/ffi/ffi_context.nim b/ffi/ffi_context.nim index c3ccfc0..fb35d8a 100644 --- a/ffi/ffi_context.nim +++ b/ffi/ffi_context.nim @@ -2,7 +2,7 @@ {.passc: "-fPIC".} -import std/[atomics, locks, options, tables] +import std/[atomics, locks, options, sequtils, tables] import chronicles, chronos, chronos/threadsync, results import ./ffi_types, @@ -15,8 +15,30 @@ import export ffi_events, ffi_handles +type CtxLifecycle* {.pure.} = enum + ## State machine guarding a pooled FFI context (Atomic on FFIContext). + ## Active -> RecyclePending when the ffiDtor requests recycle + ## RecyclePending -> Recycling FFI loop claimed it, draining handlers + ## Recycling -> Active createFFIContext reuses the slot + Active + RecyclePending + Recycling + type FFIContext*[T] = object myLib*: ptr T # main library object (Waku, LibP2P, SDS, …) + myLibRefd*: bool + # refc only: true once myLib[] (a ref) has been GC_ref'd to root it against + # the cycle collector. Balanced by GC_unref in freeLib. + myLibOwned*: bool + # true once a ctor stored a createShared'd lib into myLib (vs the worker's + # stack fallback). freeLib only frees/destroys owned libs. + inUse*: Atomic[bool] + # Whether this pooled context is claimed. The recycle handler clears it on + # the FFI thread so the slot returns to the pool without recreating threads. + lifecycle*: Atomic[CtxLifecycle] + recycleDoneSignal: ThreadSignalPtr + # fired by the recycle handler once the lib is freed and the slot released; + # the synchronous recycleFFIContext caller waits on it. ffiThread: Thread[(ptr FFIContext[T])] eventThread: Thread[(ptr FFIContext[T])] reqQueueBank: RequestQueueBank @@ -42,6 +64,9 @@ var onFFIThread* {.threadvar.}: bool const git_version* {.strdefine.} = "n/a" const + RecycleWaitTimeout* = 5.seconds + ## Caller-side bound for synchronous recycle; the FFI-thread drain itself is + ## bounded by RecycleTimeout, so this only guards against a wedged worker. EventThreadTickInterval* = 1.seconds FFIHeartbeatStartDelay* = 10.seconds FFIHeartbeatStaleThreshold* = 1.seconds @@ -73,7 +98,8 @@ proc deinitContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] = deinitHandleRegistry(ctx[].handles) deinitEventQueue(ctx[].eventQueue) when defined(gcRefc): - # ThreadSignalPtr.close() under refc hangs via signal-handler re-entry; leak the bounded fd. + # ThreadSignalPtr.close() under refc hangs via signal-handler re-entry; the + # recycle pool makes full destroy rare, so the leaked fd stays bounded. discard else: closeAndNil(ctx.reqSignal) @@ -81,6 +107,7 @@ proc deinitContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] = closeAndNil(ctx.threadExitSignal) closeAndNil(ctx.eventQueueSignal) closeAndNil(ctx.eventThreadExitSignal) + closeAndNil(ctx.recycleDoneSignal) ok() template newSignalOrErr(field: untyped, name: string) = @@ -95,6 +122,10 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] = ctx.threadExitSignal = nil ctx.eventQueueSignal = nil ctx.eventThreadExitSignal = nil + ctx.recycleDoneSignal = nil + ctx.myLibOwned = false + ctx.myLibRefd = false + ctx.lifecycle.store(CtxLifecycle.Active) initRequestQueue(ctx[].reqQueueBank) initEventRegistry(ctx[].eventRegistry) initHandleRegistry(ctx[].handles) @@ -117,6 +148,7 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] = newSignalOrErr(ctx.threadExitSignal, "threadExitSignal") newSignalOrErr(ctx.eventQueueSignal, "eventQueueSignal") newSignalOrErr(ctx.eventThreadExitSignal, "eventThreadExitSignal") + newSignalOrErr(ctx.recycleDoneSignal, "recycleDoneSignal") ctx.registeredRequests = addr ffi_types.registeredRequests @@ -167,6 +199,40 @@ proc signalStop*[T](ctx: ptr FFIContext[T]): Result[void, string] = error "failed to signal eventQueueSignal in signalStop", error = error ok() +proc tryClaim*[T](ctx: ptr FFIContext[T]): bool = + ## Atomically claim a free pooled context (false -> true). + var expected = false + ctx.inUse.compareExchange(expected, true) + +proc release*[T](ctx: ptr FFIContext[T]) = + ctx.inUse.store(false) + +proc isInUse*[T](ctx: ptr FFIContext[T]): bool = + ctx.inUse.load() + +proc markAsActive*[T](ctx: ptr FFIContext[T]) = + ## Reused context: its worker threads are still alive; re-arm for requests. + ctx.lifecycle.store(CtxLifecycle.Active) + +proc requestRecycle*[T](ctx: ptr FFIContext[T]): Result[void, string] = + ## Ask the FFI thread to drain, free the lib and release the slot, WITHOUT + ## stopping its worker/event threads, so the next createFFIContext reuses them. + ## Synchronous: waits on recycleDoneSignal. No fd churn -> no select() limit. + var expected = CtxLifecycle.Active + if not ctx.lifecycle.compareExchange(expected, CtxLifecycle.RecyclePending): + return err("requestRecycle: context is not Active (already recycling)") + + let fired = ctx.reqSignal.fireSync().valueOr: + return err("requestRecycle: failed to signal the FFI thread: " & $error) + if not fired: + return err("requestRecycle: failed to signal the FFI thread in time") + + let done = ctx.recycleDoneSignal.waitSync(RecycleWaitTimeout).valueOr: + return err("requestRecycle: failed waiting for recycle: " & $error) + if not done: + return err("requestRecycle: recycle did not complete in time") + ok() + ## Per-thread exit wait before stopAndJoinThreads leaks ctx rather than hanging; async ## `{.ffiDtor.}` teardown can outlast the default. Override `-d:ffiThreadExitTimeoutMs=`. const ThreadExitTimeoutMs* {.intdefine: "ffiThreadExitTimeoutMs".} = 1500 diff --git a/ffi/ffi_context_pool.nim b/ffi/ffi_context_pool.nim index 6c0f62d..a6e351a 100644 --- a/ffi/ffi_context_pool.nim +++ b/ffi/ffi_context_pool.nim @@ -13,37 +13,43 @@ type StaticCtxReady FFIContextPool*[T] = object - ## Fixed pool of FFI contexts, plus the one `{.ffiStatic.}` context. - # Each live context holds 5 ThreadSignalPtrs — one fd each on Linux, two (a - # socketpair) elsewhere. Under refc a destroyed context cannot close them - # (see `deinitContextResources`), so churn leaks fds unbounded. - slots: array[MaxFFIContexts, FFIContext[T]] - inUse: array[MaxFFIContexts, Atomic[bool]] + ## Fixed pool of FFI contexts, plus the one `{.ffiStatic.}` context. Each + ## slot's worker + event threads and signal fds are built once (on first + ## use) and reused across create/recycle cycles — recycle keeps them alive, + ## so repeated create/destroy does not churn fds. Bounds ThreadSignalPtr fds + ## at MaxFFIContexts * (signals per ctx). + contexts: array[MaxFFIContexts, FFIContext[T]] + initialized: array[MaxFFIContexts, Atomic[bool]] staticCtx: Atomic[pointer] staticState: Atomic[StaticCtxState] -proc acquireSlot[T](pool: var FFIContextPool[T]): Result[ptr FFIContext[T], string] = - for i in 0 ..< MaxFFIContexts: - var expected = false - if pool.inUse[i].compareExchange(expected, true): - return ok(pool.slots[i].addr) - err("FFI context pool exhausted (max " & $MaxFFIContexts & " contexts)") - proc releaseSlot[T](pool: var FFIContextPool[T], ctx: ptr FFIContext[T]) = + ## Full-teardown release: the slot must be rebuilt before it serves again. for i in 0 ..< MaxFFIContexts: - if pool.slots[i].addr == ctx: - pool.inUse[i].store(false) - return + if pool.contexts[i].addr == ctx: + pool.initialized[i].store(false) + break + ctx.release() proc createFFIContext*[T]( pool: var FFIContextPool[T] ): Result[ptr FFIContext[T], string] = - let ctx = pool.acquireSlot().valueOr: - return err("createFFIContext: acquireSlot failed: " & $error) - initContextResources(ctx).isOkOr: - pool.releaseSlot(ctx) - return err("createFFIContext: initContextResources failed: " & $error) - ok(ctx) + ## Acquires a context from the fixed pool. A slot's worker is built once on + ## first use and reused (markAsActive) on every later acquisition. + for i in 0 ..< MaxFFIContexts: + let ctx = pool.contexts[i].addr + if not ctx.tryClaim(): + continue + if pool.initialized[i].load(): + # Reused slot: a prior recycle drained and released it; worker still alive. + ctx.markAsActive() + return ok(ctx) + initContextResources(ctx).isOkOr: + ctx.release() + return err("createFFIContext: initContextResources failed: " & $error) + pool.initialized[i].store(true) + return ok(ctx) + err("FFI context pool exhausted (max " & $MaxFFIContexts & " contexts)") proc isStaticCtx[T](pool: var FFIContextPool[T], ctx: ptr FFIContext[T]): bool = ## True while `ctx` is the pool's static context, including mid-teardown. @@ -51,16 +57,29 @@ proc isStaticCtx[T](pool: var FFIContextPool[T], ctx: ptr FFIContext[T]): bool = # pointer covers `Destroying` too. pool.staticCtx.load() == cast[pointer](ctx) +proc recycleFFIContext*[T]( + pool: var FFIContextPool[T], ctx: ptr FFIContext[T] +): Result[void, string] = + ## Normal teardown: drains in-flight handlers, frees the lib and returns the + ## slot to the pool WITHOUT stopping its threads, so a later createFFIContext + ## reuses them. Synchronous (waits for the FFI thread to finish draining). + # Recycling it would release the slot while `staticState` still points at it. + if pool.isStaticCtx(ctx): + return err("recycleFFIContext(pool): the {.ffiStatic.} context outlives every ctx") + ctx.requestRecycle() + proc destroyFFIContext*[T]( pool: var FFIContextPool[T], ctx: ptr FFIContext[T] ): Result[void, string] = - ## On thread-exit timeout the slot is leaked; closing live-thread resources is unsafe. + ## Full teardown: stops/joins the threads and frees resources, marking the slot + ## uninitialised so a later createFFIContext rebuilds it; normal cleanup uses + ## recycleFFIContext. On thread-exit timeout the slot is leaked; closing + ## live-thread resources is unsafe. # Destroying it would release the slot while `staticState` still points at it. if pool.isStaticCtx(ctx): return err("destroyFFIContext(pool): the {.ffiStatic.} context outlives every ctx") ctx.stopAndJoinThreads().isOkOr: return err("destroyFFIContext(pool): " & $error) - # Required: next acquisition would otherwise re-init a live lock (UB). let deinitRes = ctx.deinitContextResources() pool.releaseSlot(ctx) deinitRes.isOkOr: @@ -118,6 +137,6 @@ proc isValidCtx*[T](pool: var FFIContextPool[T], ctx: pointer): bool = if ctx.isNil(): return false for i in 0 ..< MaxFFIContexts: - if cast[pointer](pool.slots[i].addr) == ctx: - return pool.inUse[i].load() + if cast[pointer](pool.contexts[i].addr) == ctx: + return cast[ptr FFIContext[T]](ctx).isInUse() false diff --git a/ffi/ffi_events.nim b/ffi/ffi_events.nim index 0131875..717cdbd 100644 --- a/ffi/ffi_events.nim +++ b/ffi/ffi_events.nim @@ -35,6 +35,13 @@ proc deinitEventRegistry*(reg: var FFIEventRegistry) = reg.byEvent = default(Table[string, seq[FFIEventListener]]) reg.nextId = 0'u64 +proc clearListeners*(reg: var FFIEventRegistry) {.raises: [].} = + ## Drops all listeners (used when a context is recycled for reuse) without + ## touching the lock — the event thread keeps using it across recycles. + withLock reg.lock: + reg.byEvent.clear() + reg.nextId = 0'u64 + proc addEventListener*( reg: var FFIEventRegistry, eventName: string, diff --git a/ffi/ffi_thread.nim b/ffi/ffi_thread.nim index aa8577f..d6617fb 100644 --- a/ffi/ffi_thread.nim +++ b/ffi/ffi_thread.nim @@ -16,6 +16,10 @@ proc sendRequestToFFIThread*( "reentrant ffi call: a handler invoked sendRequestToFFIThread on its own context" ) + if ctx.lifecycle.load() != CtxLifecycle.Active: + deleteRequest(ffiRequest) + return err("FFI context is not accepting requests (being recycled)") + # Wake only when the push found the queue empty: waking per submit kills scaling, and a skipped wake just waits the consumer's 100ms poll. let shouldWake = ctx.reqQueueBank.pushRequest(ffiRequest) @@ -82,6 +86,56 @@ proc processRequest[T]( except Exception as e: error "Unexpected exception in handleRes", error = e.msg +const RecycleTimeout = 1500.milliseconds + ## Bounds how long the recycle handler waits for in-flight handlers before it + ## cancels them, so a wedged handler cannot block reuse forever. + +proc freeLib[T](ctx: ptr FFIContext[T]) {.gcsafe.} = + ## Releases the library object the ctor stored in ctx.myLib. Only owned libs + ## (createShared'd by a ctor) are freed; the worker's stack fallback is not. + if not ctx.myLibOwned or ctx.myLib.isNil(): + ctx.myLib = nil + return + when not defined(gcRefc): + try: + {.cast(gcsafe).}: + `=destroy`(ctx.myLib[]) + except Exception: + discard + else: + when T is ref: + if ctx.myLibRefd: + GC_unref(ctx.myLib[]) + ctx.myLibRefd = false + freeShared(ctx.myLib) + ctx.myLib = nil + ctx.myLibOwned = false + +proc recycleContext[T]( + ctx: ptr FFIContext[T], ongoing: ptr seq[Future[void]] +) {.async.} = + ## Drain in-flight handlers, free the lib, clear listeners and release the + ## slot — all WITHOUT stopping the worker/event threads, so the next + ## createFFIContext reuses them (no fd churn). Then fire recycleDoneSignal. + ongoing[].keepItIf(not it.finished()) + var drained = ongoing[].len == 0 + if not drained: + drained = await allFutures(ongoing[]).withTimeout(RecycleTimeout) + if not drained: + for fut in ongoing[]: + if not fut.finished(): + fut.cancelSoon() + drained = await allFutures(ongoing[]).withTimeout(RecycleTimeout) + + freeLib(ctx) + clearListeners(ctx[].eventRegistry) + ongoing[].setLen(0) + ctx.release() + + let fireRes = ctx.recycleDoneSignal.fireSync() + if fireRes.isErr(): + error "failed to fire recycleDoneSignal", err = fireRes.error + var ffiEventQueueSignalPtr {.threadvar.}: ThreadSignalPtr # Stashed so the hook has no closure env. @@ -152,6 +206,13 @@ proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} = while ctx.running.load(): ctx.proveAlive() + # Recycle requested by the ffiDtor: drain + free lib + release the slot, + # keeping this thread alive for the next createFFIContext to reuse. + var expected = CtxLifecycle.RecyclePending + if ctx.lifecycle.compareExchange(expected, CtxLifecycle.Recycling): + await recycleContext(ctx, addr pending) + continue + cleanFinishedRequests() # Block until a submit signals us, or at most 100ms. diff --git a/ffi/internal/ffi_macro.nim b/ffi/internal/ffi_macro.nim index 71590f9..d2d87e5 100644 --- a/ffi/internal/ffi_macro.nim +++ b/ffi/internal/ffi_macro.nim @@ -392,13 +392,18 @@ proc buildFFINewReqProc(reqTypeName, body: NimNode): NimNode = `reqObjIdent`.`fieldName` = `fieldName` ) + let reqNameLit = newLit( + if reqTypeName.kind == nnkPostfix: + $reqTypeName[1] + else: + $reqTypeName + ) newBody.add( quote do: - let typeStr = $T # Encode into shared memory, avoiding a second seq[byte] copy. let (sharedData, sharedLen) = cborEncodeShared(`reqObjIdent`) return FFIThreadRequest.initFromOwnedShared( - callback, userData, typeStr.cstring, sharedData, sharedLen + callback, userData, cstring(`reqNameLit`), sharedData, sharedLen ) ) @@ -1077,10 +1082,15 @@ proc buildFFIProc( ffiBody.add(stmt) let reqPtrIdent = genSym(nskLet, "reqPtr") + let reqNameLit = newLit( + if reqTypeName.kind == nnkPostfix: + $reqTypeName[1] + else: + $reqTypeName + ) ffiBody.add quote do: - let typeStr = $`reqTypeName` let `reqPtrIdent` = FFIThreadRequest.initFromPtr( - callback, userData, typeStr.cstring, reqCbor, int(reqCborLen) + callback, userData, cstring(`reqNameLit`), reqCbor, int(reqCborLen) ) ffiBody.add buildSendAndReply(reqPtrIdent) @@ -1208,11 +1218,16 @@ proc buildCtorFFINewReqProc(reqTypeName: NimNode, paramNames: seq[string]): NimN let retType = newTree(nnkPtrTy, ident("FFIThreadRequest")) formalParams = @[retType] & formalParams + let reqNameLit = newLit( + if reqTypeName.kind == nnkPostfix: + $reqTypeName[1] + else: + $reqTypeName + ) var newBody = newStmtList() newBody.add quote do: - let typeStr = $T return FFIThreadRequest.initFromPtr( - callback, userData, typeStr.cstring, reqCbor, int(reqCborLen) + callback, userData, cstring(`reqNameLit`), reqCbor, int(reqCborLen) ) let newReqProc = newProc( @@ -1299,9 +1314,18 @@ proc buildCtorProcessFFIRequestProc( return err($error) let myLibIdent = newDotExpr(newTree(nnkDerefExpr, ctxIdent), ident("myLib")) + let myLibOwnedIdent = newDotExpr(newTree(nnkDerefExpr, ctxIdent), ident("myLibOwned")) + let myLibRefdIdent = newDotExpr(newTree(nnkDerefExpr, ctxIdent), ident("myLibRefd")) newBody.add quote do: `myLibIdent` = createShared(`libTypeName`) `myLibIdent`[] = `libValIdent` + `myLibOwnedIdent` = true + # Root the ref lib under refc: it lives only via this ptr in non-GC + # createShared memory, invisible to the cycle collector. freeLib unroots it. + when defined(gcRefc): + when `libTypeName` is ref: + GC_ref(`myLibIdent`[]) + `myLibRefdIdent` = true newBody.add quote do: return ok($cast[uint](`ctxIdent`)) @@ -1649,7 +1673,7 @@ macro ffiDtor*(args: varargs[untyped]): untyped = let poolIdent = ident($libTypeName & "FFIPool") ffiBody.add quote do: let `destroyResIdent` = - `poolIdent`.destroyFFIContext(cast[ptr FFIContext[`libTypeName`]](ctx)) + `poolIdent`.recycleFFIContext(cast[ptr FFIContext[`libTypeName`]](ctx)) if `destroyResIdent`.isErr(): return RET_ERR