logos-messaging/nim-ffi
1
0
Fork 0
mirror of https://github.com/logos-messaging/nim-ffi.git synced 2026-06-20 16:29:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: reduce comment sizes

Browse Source
This commit is contained in:
Gabriel Cruz 2026-06-02 16:53:04 -03:00
parent 423bd158c0
commit ee2ba0f9e9
No known key found for this signature in database
GPG Key ID: 3C6977037D5A1EF5
3 changed files with 73 additions and 222 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

195
ffi/ffi_context.nim
View File

@ -19,38 +19,25 @@ type FFIContext*[T] = object
ffiThread: Thread[(ptr FFIContext[T])]
# represents the main FFI thread in charge of attending API consumer actions
eventThread: Thread[(ptr FFIContext[T])]
# drains the bounded event queue and runs the heartbeat health check;
# replaces the previous standalone watchdog thread
# drains the event queue and runs the FFI-thread heartbeat check
lock: Lock
reqChannel: ChannelSPSCSingle[ptr FFIThreadRequest]
reqSignal: ThreadSignalPtr # to notify the FFI Thread that a new request is sent
reqReceivedSignal: ThreadSignalPtr
# to signal main thread, interfacing with the FFI thread, that FFI thread received the request
stopSignal: ThreadSignalPtr
# fired by destroyFFIContext so both ffiThread and eventThread can exit promptly
threadExitSignal: ThreadSignalPtr
# fired by ffiThread just before it exits; destroyFFIContext waits on
# this with a bounded timeout instead of joining unconditionally, so a
# blocked event loop cannot hang the caller forever
eventQueueSignal: ThreadSignalPtr
# fired by the FFI thread (via the dispatch templates) when it enqueues
# an event, so the event thread wakes promptly instead of waiting out
# the tick interval
eventThreadExitSignal: ThreadSignalPtr
# fired by the event thread just before it exits; mirrors threadExitSignal
# so destroyFFIContext can do a bounded wait on the event thread too
# fired by ffiThread before exit; bounds destroyFFIContext's wait so
# a blocked event loop cannot hang the caller
eventQueueSignal: ThreadSignalPtr # wakes the event thread on enqueue
eventThreadExitSignal: ThreadSignalPtr # mirrors threadExitSignal for the event thread
userData*: pointer
eventRegistry*: FFIEventRegistry
eventQueue*: EventQueue
# bounded SPSC ring; the FFI thread is the only producer, the event
# thread the only consumer
ffiHeartbeat*: Atomic[int64]
# advanced by the FFI thread on every iteration of its main loop; the
# event thread reads it to detect a wedged FFI thread (>1s without an
# advance after the start-grace window) and fires onNotResponding
eventQueueStuck*: Atomic[bool]
# sticky overflow flag — once the queue saturates, sendRequestToFFIThread
# rejects further calls so the library can't dig itself deeper
# advanced by the FFI thread each loop iteration; event thread reads it
# for liveness
eventQueueStuck*: Atomic[bool] # sticky overflow flag; recovery is destroy+recreate
running: Atomic[bool] # To control when the threads are running
registeredRequests: ptr Table[cstring, FFIRequestProc]
# Pointer to with the registered requests at compile time
@ -63,49 +50,18 @@ var onFFIThread* {.threadvar.}: bool
const git_version* {.strdefine.} = "n/a"
const
EventThreadTickInterval* = 1.seconds
## How often the event thread wakes to do a heartbeat check when no
## events are pending. The dispatch templates also fire
## `eventQueueSignal` on enqueue, so this only bounds the *idle*
## latency between consecutive heartbeat checks.
FFIHeartbeatStartDelay* = 10.seconds
## Grace window after thread startup during which heartbeat stalls
## are ignored — gives the host library (waku / libp2p / …) time to
## come up before we start measuring liveness. Same value the old
## standalone watchdog used.
EventThreadTickInterval* = 1.seconds # bounds idle heartbeat check latency
FFIHeartbeatStartDelay* = 10.seconds # grace window for library startup
FFIHeartbeatStaleThreshold* = 1.seconds
## Once past the start delay, the FFI thread must advance its
## heartbeat at least once per this interval or it is considered
## blocked and onNotResponding fires.
type NotRespondingEvent* = object
## Empty CBOR payload — the event itself is the signal. Consumers
## discriminate on the surrounding `EventEnvelope.eventType`
## (`NotRespondingEventName`); this struct exists so the wire shape
## matches the typed-event contract every other `{.ffiEvent.}`
## produces (`{ eventType, payload }`).
const NotRespondingEventName* = "not_responding"
## Registry key and CBOR `eventType` for `onNotResponding`. Exposed so
## typed listeners can register against the same name the dispatch
## path uses, without depending on a string literal.
proc onNotResponding*(ctx: ptr FFIContext) =
## Fans out the "library is unhealthy" event to every listener
## registered for `not_responding` plus every wildcard listener. The
## payload is `EventEnvelope[NotRespondingEvent]`, CBOR-encoded once
## and reused across listeners — same wire shape as
## `dispatchFFIEventCbor`.
##
## Synchronous, lock-during-invocation by design: this is the global
## "library is unhealthy" notification path and bypasses the event
## queue (which may itself be the thing that's stuck). The dispatch
## templates' lock-during-invocation contract is mirrored here.
##
## Cannot reuse `dispatchFFIEventCbor`: that template resolves the
## registry via the `ffiCurrentEventRegistry` threadvar, which is only
## set on the FFI thread. `onNotResponding` runs on the event thread,
## so it goes through `ctx[].eventRegistry` directly.
## Bypasses the event queue (which may itself be wedged). Cannot reuse
## `dispatchFFIEventCbor`: that template reads `ffiCurrentEventRegistry`,
## a threadvar only set on the FFI thread, but this runs on the event thread.
withLock ctx[].eventRegistry.lock:
let snap =
ctx[].eventRegistry.byEvent.getOrDefault(NotRespondingEventName) &
@ -142,24 +98,15 @@ proc onNotResponding*(ctx: ptr FFIContext) =
proc sendRequestToFFIThread*(
ctx: ptr FFIContext, ffiRequest: ptr FFIThreadRequest, timeout = InfiniteDuration
): Result[void, string] =
# Issue #6: once the event queue has overflowed we stop accepting new
# requests entirely, on the assumption that any further work will just
# produce more events the listener side already can't keep up with.
# Stuck flag is sticky for the context lifetime — recovery requires
# destroy + recreate, matching the issue's "expected malfunctioning".
# NB: we deliberately do NOT call onNotResponding here. The event
# thread fires it once when it observes the stuck flag (its loop is
# the only place where reg.lock is guaranteed not to be held by an
# in-flight listener); calling it from a foreign thread would
# deadlock against a back-pressuring listener mid-invocation.
# Once the event queue overflows, refuse further requests. onNotResponding
# is fired by the event thread (not here) to avoid deadlocking against a
# back-pressuring listener that holds reg.lock.
if ctx.eventQueueStuck.load():
deleteRequest(ffiRequest)
return err("event queue stuck - library cannot accept new requests")
# Reentrancy guard (PR #23 review, item 6): if a handler running on the FFI
# thread tries to dispatch back through this proc, it would wait forever on
# `reqReceivedSignal` — which only this thread can fire — and self-deadlock.
# Return an error instead so the caller can surface it.
# Reentrancy guard: a handler dispatching back through this proc would
# wait forever on `reqReceivedSignal` — only this thread can fire it.
if onFFIThread:
deleteRequest(ffiRequest)
return err(
@ -241,22 +188,13 @@ proc processRequest[T](
except Exception as exc:
error "Unexpected exception in handleRes", error = exc.msg
# ---------------------------------------------------------------------------
# Heartbeat-aware closure capturing for the dispatch threadvar hook
# ---------------------------------------------------------------------------
var ffiEventQueueSignalPtr {.threadvar.}: ThreadSignalPtr
## Stash for the event-thread wakeup signal. Captured here so the
## notify-enqueued hook below has no closure environment.
## Stashed so the hook below has no closure environment.
proc ffiNotifyEventEnqueuedHook() {.gcsafe, raises: [].} =
## Wakes the event thread immediately after a successful enqueue so
## the listener fan-out latency isn't bounded by the tick interval.
if not ffiEventQueueSignalPtr.isNil():
let res = ffiEventQueueSignalPtr.fireSync()
if res.isErr():
# The event thread will still see the queue depth on the next
# tick; logging is enough to flag a misconfigured signal fd.
error "failed to fire eventQueueSignal after enqueue", err = res.error
proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
@ -298,11 +236,8 @@ proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
inc i
while ctx.running.load():
# Heartbeat: the event thread reads this to confirm the FFI thread
# isn't wedged. The 100 ms `reqSignal.wait` below means we advance
# at least ~10x per second under any normal load; a sync handler
# that blocks the dispatcher will freeze the counter, which is
# exactly the failure mode the watchdog used to detect.
# A sync handler blocking the dispatcher freezes this counter; the
# event thread reads it to detect a wedged FFI thread.
discard ctx.ffiHeartbeat.fetchAdd(1)
reapCompleted()
@ -340,20 +275,10 @@ proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
waitFor ffiRun(ctx)
proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
## Drains the bounded event queue and runs the FFI-thread heartbeat
## health check. Replaces the standalone watchdog thread: the event
## thread checks liveness in-band (no probe request round-trip), and
## the dispatch templates' queue-overflow path is the second trigger
## for `onNotResponding`.
##
## The queue stores raw `c_malloc` payloads + names; this thread owns
## them for the duration of dispatch and frees them after the listener
## fan-out returns.
## Drains the event queue and runs the FFI-thread heartbeat check.
## Owns the queued `c_malloc` payloads until dispatch returns.
defer:
# Best-effort: tell stopAndJoinThreads we've exited so its bounded
# wait unblocks. If this fails we still exit; the caller's timeout
# path will take over.
let fireRes = ctx.eventThreadExitSignal.fireSync()
if fireRes.isErr():
error "failed to fire eventThreadExitSignal", err = fireRes.error
@ -366,19 +291,11 @@ proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
var notifiedStuck = false
while ctx.running.load():
# Wake on either an enqueue (eventQueueSignal) or the tick interval —
# whichever comes first. The signal path keeps dispatch latency low
# under load; the timeout path bounds idle latency for the
# heartbeat check.
# Wake on enqueue or tick — whichever first.
discard await ctx.eventQueueSignal.wait().withTimeout(EventThreadTickInterval)
# Drain whatever is currently in the queue. Each iteration:
# 1. Pop one event (queue lock).
# 2. Snapshot listeners + invoke them under reg.lock — the
# lock-during-invocation contract from PR #39 / issue #40 is
# preserved here: a foreign `removeEventListener` blocks
# until the in-flight callback fan-out returns.
# 3. Free the payload.
# Listener fan-out runs under reg.lock — preserves the
# lock-during-invocation contract from PR #39 / issue #40.
while true:
let opt = ctx.eventQueue.tryDequeueEvent()
if opt.isNone:
@ -417,20 +334,13 @@ proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
listener.userData,
)
# Queue-overflow notification: the FFI thread can only set the
# sticky flag (firing onNotResponding from there would deadlock
# against a back-pressuring listener that's holding reg.lock on
# this thread). We fire it once from here, after the drain loop,
# so the slow listener has already released the lock.
# Overflow notification fires here (after drain releases reg.lock)
# rather than from the FFI thread, which would deadlock against an
# in-flight back-pressuring listener.
if not notifiedStuck and ctx.eventQueueStuck.load():
onNotResponding(ctx)
notifiedStuck = true
# Heartbeat staleness check. Skipped during the start-delay grace
# window so a slow library bring-up doesn't fire a spurious
# not_responding. Once we've fired, latch `notifiedStale` until
# the FFI thread proves it's alive again — avoids spamming the
# listener while the FFI thread is still stuck.
if not ctx.running.load():
break
if Moment.now() - startedAt <= FFIHeartbeatStartDelay:
@ -443,6 +353,7 @@ proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
notifiedStale = false
elif not notifiedStale and
Moment.now() - lastHeartbeatChange > FFIHeartbeatStaleThreshold:
# Latch until the FFI thread proves it's alive again.
onNotResponding(ctx)
notifiedStale = true
@ -452,14 +363,10 @@ proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
error "event thread exited with exception", error = exc.msg
proc deinitContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
## Mirror of `initContextResources`: tears down the lock, registry,
## queue, and signal fds in place. The caller is responsible for the
## memory holding `ctx` (free it for heap allocations, return it to the
## pool for slot-allocated contexts). Threads MUST already be joined.
##
## Each field is nil'd after close so a subsequent re-init on the same
## storage (pool slot reuse) doesn't double-close a stale pointer if
## init's deferred cleanup runs.
## Mirror of `initContextResources`: tears down lock, registry, queue,
## and signal fds in place. Threads MUST already be joined. Caller owns
## the memory holding `ctx`. Fields are nil'd after close so a re-init
## on the same slot doesn't double-close.
ctx.lock.deinitLock()
deinitEventRegistry(ctx[].eventRegistry)
deinitEventQueue(ctx[].eventQueue)
@ -515,11 +422,8 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
## Initialises all resources inside an already-allocated FFIContext slot.
## On failure every partially-initialised resource is closed; the caller
## is responsible for releasing the slot (freeShared or pool.releaseSlot).
##
## Defensive: a reused pool slot still holds the previous lifetime's
## signal pointers (set to nil by `deinitContextResources` on destroy,
## but explicit here in case a future path forgets). Nil before
## allocating so the deferred cleanup never double-closes.
# Defensive nil so the deferred cleanup never double-closes if a future
# path forgets to clear stale pointers on a reused pool slot.
ctx.reqSignal = nil
ctx.reqReceivedSignal = nil
ctx.stopSignal = nil
@ -583,11 +487,10 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
return ok()
proc signalStop*[T](ctx: ptr FFIContext[T]): Result[void, string] =
# Error paths intentionally skip onNotResponding: a back-pressuring
# listener may hold reg.lock, and onNotResponding takes it — would
# amplify the stuck state into a deadlock instead of escaping it.
ctx.running.store(false)
# We deliberately do NOT call onNotResponding from these error paths:
# the event thread may be stuck mid-callback holding `eventRegistry.lock`,
# and onNotResponding takes that same lock — exactly the scenario the
# bounded-timeout caller needs to escape from, not amplify into a deadlock.
let reqSignaled = ctx.reqSignal.fireSync().valueOr:
return err("error signaling reqSignal in signalStop: " & $error)
if not reqSignaled:
@ -596,10 +499,7 @@ proc signalStop*[T](ctx: ptr FFIContext[T]): Result[void, string] =
return err("error signaling stopSignal in signalStop: " & $error)
if not stopSignaled:
return err("failed to signal stopSignal on time in signalStop")
# Wake the event thread so it observes `running == false` immediately
# instead of waiting out the tick interval. fireSync failing here is
# not fatal — the event thread will still notice on the next tick — so
# we only log and continue.
# Non-fatal: event thread will see running==false on the next tick.
let evtSignaled = ctx.eventQueueSignal.fireSync()
if evtSignaled.isErr():
error "failed to signal eventQueueSignal in signalStop", error = evtSignaled.error
@ -615,18 +515,15 @@ proc signalStop*[T](ctx: ptr FFIContext[T]): Result[void, string] =
const ThreadExitTimeout* = 1500.milliseconds
proc stopAndJoinThreads*[T](ctx: ptr FFIContext[T]): Result[void, string] =
## Signals the FFI and event threads to stop, waits up to ThreadExitTimeout
## for each to exit, and joins them. On timeout returns err and skips the
## remaining joinThread (leaving the threads live) rather than hanging the
## caller. Resource cleanup (signal fds, lock) is the caller's
## responsibility.
## Signals both threads to stop, waits up to ThreadExitTimeout per thread,
## and joins them. On timeout returns err and skips remaining joins
## (leaving the threads live) rather than hanging the caller. Resource
## cleanup is the caller's responsibility.
##
## Timeout paths skip onNotResponding for the same reason signalStop does.
ctx.signalStop().isOkOr:
return err("signalStop failed: " & $error)
# We deliberately do NOT call onNotResponding from the timeout paths:
# the event thread may be stuck mid-callback holding `eventRegistry.lock`,
# and onNotResponding takes that same lock — exactly the scenario the
# bounded-timeout caller needs to escape from, not amplify into a deadlock.
let ffiExitedOnTime = ctx.threadExitSignal.waitSync(ThreadExitTimeout).valueOr:
return err("error waiting for FFI thread exit: " & $error)

7
ffi/ffi_context_pool.nim
View File

@ -48,11 +48,8 @@ proc destroyFFIContext*[T](
## unsafe.
ctx.stopAndJoinThreads().isOkOr:
return err("destroyFFIContext(pool): " & $error)
# Mirror initContextResources: tear down the lock, registry, queue,
# and signal fds in place. Without this the next slot acquisition would
# re-init an already-initialised lock (UB at the pthread layer) and
# overwrite the existing ThreadSignalPtr fields without closing the
# underlying fds (unbounded fd leak across create/destroy cycles).
# Without this, the next acquisition would re-init an already-initialised
# lock (UB) and leak the previous signal fds.
let deinitRes = ctx.deinitContextResources()
pool.releaseSlot(ctx)
deinitRes.isOkOr:

93
ffi/ffi_events.nim
View File

@ -1,22 +1,8 @@
## Event registry, bounded event queue, and dispatch primitives for FFI
## library-initiated events.
##
## This module owns three concerns so they can evolve together without
## dragging in the rest of `FFIContext`:
##
## 1. A multi-listener registry. Each event name maps to a `seq` of
## listeners; the empty event name `""` is the wildcard channel and
## receives every dispatched event in addition to its own per-name
## subscribers.
## 2. A bounded SPSC event queue. Infrastructure for the dedicated event
## thread (owned by `FFIContext`) to drain encoded events; payloads
## travel via `c_malloc` so transfer across Nim heaps is safe under
## both `--mm:orc` and `--mm:refc`. The dispatch templates do not yet
## enqueue — that rewiring lands alongside the dispatch overhaul.
## 3. The dispatch templates (`dispatchFFIEvent`, `dispatchFFIEventCbor`)
## used by `{.ffiEvent.}`-generated procs. They snapshot the registry
## under its lock, then invoke each listener *outside* the lock so
## re-entrant add/remove from within a handler cannot self-deadlock.
## Event registry, bounded SPSC event queue, and dispatch templates for
## FFI library-initiated events. Empty event name `""` registers a
## wildcard listener that receives every dispatched event. Queue
## payloads travel via `c_malloc` so transfer across Nim heaps is safe
## under both `--mm:orc` and `--mm:refc`.
{.pragma: callback, cdecl, raises: [], gcsafe.}
@ -180,36 +166,28 @@ proc snapshotListeners*(
# ---------------------------------------------------------------------------
const EventQueueCapacity* = 1024
## Maximum number of events that can sit in the queue at once. Sized
## generously — a sustained backlog at this depth almost certainly
## means a user listener is wedged, which is exactly what the stuck
## flag is meant to surface. Each `QueuedEvent` is two pointers plus
## an int (24 B on 64-bit), so the ring is ~24 KiB per context.
## ~24 KiB per context. Sustained backlog at this depth means a
## listener is wedged — what the stuck flag exists to surface.
type
QueuedEvent* = object
## A single event sitting in the bounded queue. All fields are
## raw `c_malloc` pointers — no GC-managed storage — so the queue
## can be a plain `array` without an assignment destructor running
## across thread heaps when an `FFIContextPool` slot is reused.
name*: cstring ## c_malloc'd copy of the event name.
data*: ptr UncheckedArray[byte] ## c_malloc'd CBOR-encoded payload (may be nil).
## All fields are raw `c_malloc` pointers so the buffer survives
## pool-slot reuse across thread heaps without an assignment dtor.
name*: cstring
data*: ptr UncheckedArray[byte]
dataLen*: int
EventQueue* = object
## SPSC ring. Only the FFI thread enqueues; only the event thread
## dequeues. `lock` is sufficient — no need for atomic indices —
## because every operation is short and uncontended.
## SPSC ring: FFI thread enqueues, event thread dequeues. Plain lock
## (no atomic indices) — operations are short and uncontended.
lock*: Lock
head*: int ## Next slot the consumer will read.
tail*: int ## Next slot the producer will write.
count*: int ## Current depth, in [0, EventQueueCapacity].
head*: int
tail*: int
count*: int
buf*: array[EventQueueCapacity, QueuedEvent]
proc initEventQueue*(q: var EventQueue) {.raises: [].} =
## Initialises the queue's lock and zeroes the ring. Must be called
## exactly once on the owning thread before any other thread uses it
## (same constraint as `initEventRegistry`).
## Same single-owning-thread constraint as `initEventRegistry`.
q.lock.initLock()
q.head = 0
q.tail = 0
@ -218,10 +196,7 @@ proc initEventQueue*(q: var EventQueue) {.raises: [].} =
q.buf[i] = QueuedEvent(name: nil, data: nil, dataLen: 0)
proc deinitEventQueue*(q: var EventQueue) {.raises: [].} =
## Frees any pending entries with `c_free` and tears down the lock.
## Called on shutdown (after both producer and consumer threads have
## stopped) and on pool-slot reuse so the next thread to grab the
## slot starts from a clean state.
## Both producer and consumer must have stopped before calling.
for i in 0 ..< EventQueueCapacity:
let e = q.buf[i]
if not e.name.isNil:
@ -237,10 +212,8 @@ proc deinitEventQueue*(q: var EventQueue) {.raises: [].} =
proc tryEnqueueEvent*(
q: var EventQueue, name: cstring, data: ptr UncheckedArray[byte], dataLen: int
): bool {.raises: [], gcsafe.} =
## Pushes `(name, data, dataLen)` onto the queue. The queue takes
## ownership of both `name` and `data` (both must be `c_malloc`'d by
## the caller). Returns false if the queue is full — in that case the
## caller still owns the buffers and must free them.
## Both `name` and `data` must be `c_malloc`'d; on success the queue
## takes ownership. On false the caller still owns and must free them.
withLock q.lock:
if q.count >= EventQueueCapacity:
return false
@ -250,9 +223,7 @@ proc tryEnqueueEvent*(
return true
proc tryDequeueEvent*(q: var EventQueue): Option[QueuedEvent] {.raises: [], gcsafe.} =
## Pops the next entry off the queue and transfers ownership of its
## buffers to the caller (who must `c_free(name)` and `c_free(data)`).
## Returns `none` when the queue is empty.
## Transfers buffer ownership to the caller, who must `c_free` both.
withLock q.lock:
if q.count == 0:
return none(QueuedEvent)
@ -263,7 +234,6 @@ proc tryDequeueEvent*(q: var EventQueue): Option[QueuedEvent] {.raises: [], gcsa
return some(e)
proc eventQueueLen*(q: var EventQueue): int {.raises: [], gcsafe.} =
## Snapshot depth, mainly useful from tests.
withLock q.lock:
return q.count
@ -272,28 +242,15 @@ proc eventQueueLen*(q: var EventQueue): int {.raises: [], gcsafe.} =
# ---------------------------------------------------------------------------
var ffiCurrentEventRegistry* {.threadvar.}: ptr FFIEventRegistry
## Set by the FFI thread at startup so dispatchFFIEvent / dispatchFFIEventCbor
## can find their registry without taking a context pointer per call site.
var ffiCurrentEventQueue* {.threadvar.}: ptr EventQueue
## Bounded queue handle for the dedicated event thread to drain. The
## dispatch templates do not enqueue yet; this is set up so the event
## thread infrastructure has somewhere to read from.
var ffiCurrentEventQueueStuck* {.threadvar.}: ptr Atomic[bool]
## Sticky overflow flag belonging to the owning `FFIContext`. Reserved
## for the dispatch-overhaul follow-up; not consulted yet.
var ffiCurrentNotifyEventEnqueued* {.threadvar.}: proc() {.gcsafe, raises: [].}
## Wakes the event thread after a successful enqueue. Kept as a
## threadvar hook (rather than a queue field) so `ffi_events.nim`
## doesn't have to depend on chronos's `ThreadSignalPtr`. Nil-safe.
## Hook (not a queue field) so this module doesn't depend on chronos's
## ThreadSignalPtr. Nil-safe.
template withFFIEventDispatch(eventName: string, listeners, body: untyped) =
## Shared scaffold for `dispatchFFIEvent` / `dispatchFFIEventCbor`:
## resolves the thread-local registry, snapshots listeners under
## `reg.lock` into the caller-named `listeners` binding, then runs
## `body` inside `foreignThreadGc` + try/except.
## Resolves the thread-local registry, snapshots listeners under
## `reg.lock`, then runs `body` inside `foreignThreadGc` + try/except.
let regPtr = ffiCurrentEventRegistry
if regPtr.isNil():
chronicles.error eventName & " - event registry not set on this thread"