feat(ffi): configurable per-request handler timeout with a finite default (#93) (#108)

This commit is contained in:
Gabriel Cruz 2026-07-06 11:58:04 -03:00 committed by GitHub
parent fb0a1f1077
commit 8dcdcdc76a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 328 additions and 23 deletions

View File

@ -34,6 +34,16 @@ All notable changes to this project are documented in this file.
`nimble genbindings_c` / `genbindings_c_echo` / `check_bindings_c` /
`test_c_e2e` tasks, a `tests/e2e/c` ctest harness, and a
`tests/unit/test_c_codegen.nim` unit suite.
- Configurable per-request handler timeout with a finite default: each
`FFIContext` now carries a `defaultRequestTimeout` (5s) applied to every
handler, replacing the previous unbounded wait so a wedged handler can no
longer hang a foreign caller forever. On trip the caller is unblocked with an
`ffi request timed out after <n>ms` err; the handler is left running (not
cancelled, since a hard-cancel mid-call into the underlying library can leave
it partial), and the callback still fires exactly once. Override per proc with
a `"timeout = <ms>"` spec (e.g. `{.ffi: "timeout = 30000".}`), parsed like the
`abi = ...` spec; runtime-only, codegen ignores it
([#93](https://github.com/logos-messaging/nim-ffi/issues/93)).
- Per-interaction ABI-format annotations: `declareLibrary` now takes an
optional `defaultABIFormat` (`"cbor"` default, or `"c"`) that every
`{.ffi.}` / `{.ffiCtor.}` / `{.ffiDtor.}` / `{.ffiRaw.}` / `{.ffiEvent.}`

View File

@ -72,6 +72,37 @@ proc abiCodegenImplemented*(fmt: ABIFormat): bool =
## seam a future PR flips once the `c` dispatch path is wired.
fmt == ABIFormat.Cbor
proc overrideKey*(override: string): string =
## Lowercased key of a `key = value` pragma override (the text before `=`),
## used to route it to its parser. `"timeout = 30000"` → `"timeout"`.
override.split('=')[0].strip().toLowerAscii()
proc parseTimeoutSpec*(override: string): tuple[ok: bool, ms: int, err: string] =
## Parse a `"timeout = <milliseconds>"` override (whitespace/case tolerant).
## The value must be a positive integer number of milliseconds. On bad
## grammar or value, returns `ok = false` with a human-readable `err`.
let parts = override.split('=')
if parts.len != 2 or overrideKey(override) != "timeout":
return (
false,
0,
"invalid timeout override: '" & override & "'; expected `timeout = <ms>`",
)
let raw = parts[1].strip()
let ms =
try:
parseInt(raw)
except ValueError:
return (
false,
0,
"invalid timeout value: '" & raw &
"'; expected a positive integer of milliseconds",
)
if ms <= 0:
return (false, 0, "timeout must be a positive number of milliseconds, got: " & raw)
(true, ms, "")
proc parseABIFormatName*(name: string): tuple[ok: bool, fmt: ABIFormat] =
## Bare format name (`"c"`/`"cbor"`, case-insensitive) → `ABIFormat`;
## `ok` is false otherwise.
@ -83,28 +114,28 @@ proc parseABIFormatName*(name: string): tuple[ok: bool, fmt: ABIFormat] =
else:
(false, ABIFormat.Cbor)
proc parseAbiSpec*(spec: string): tuple[ok: bool, fmt: ABIFormat, err: string] =
proc parseAbiSpec*(override: string): tuple[ok: bool, fmt: ABIFormat, err: string] =
## Parse an `"abi = <format>"` override (whitespace/case tolerant). On bad
## grammar or format, returns `ok = false` with a human-readable `err`.
let parts = spec.split('=')
let parts = override.split('=')
if parts.len != 2:
return (
false,
ABIFormat.Cbor,
"invalid ABI override '" & spec & "'; expected `abi = c` or `abi = cbor`",
"invalid ABI override: '" & override & "'; expected `abi = c` or `abi = cbor`",
)
if parts[0].strip().toLowerAscii() != "abi":
return (
false,
ABIFormat.Cbor,
"invalid ABI override '" & spec & "'; expected `abi = c` or `abi = cbor`",
"invalid ABI override: '" & override & "'; expected `abi = c` or `abi = cbor`",
)
let (ok, fmt) = parseABIFormatName(parts[1])
if not ok:
return (
false,
ABIFormat.Cbor,
"unknown ABI format '" & parts[1].strip() & "'; valid values are `c` and `cbor`",
"unknown ABI format: '" & parts[1].strip() & "'; valid values are `c` and `cbor`",
)
(true, fmt, "")

View File

@ -39,6 +39,12 @@ type FFIContext*[T] = object
eventQueueStuck*: Atomic[bool] # sticky overflow flag
running: Atomic[bool] # To control when the threads are running
registeredRequests: ptr Table[cstring, FFIRequestProc]
requestTimeouts: ptr Table[cstring, int]
# Per-proc timeout overrides (ms). Points at the compile-time-filled global,
# like registeredRequests, so the FFI thread reads it GC-safely via ctx.
defaultRequestTimeout*: Duration
# Per-handler deadline unless a `{.ffi: "timeout = <ms>".}` override raises
# it; `InfiniteDuration` opts out. See processRequest for the trip behavior.
var onFFIThread* {.threadvar.}: bool
# Re-entrant dispatch guard for `sendRequestToFFIThread`.
@ -49,6 +55,8 @@ const
EventThreadTickInterval* = 1.seconds
FFIHeartbeatStartDelay* = 10.seconds # grace window for library startup
FFIHeartbeatStaleThreshold* = 1.seconds
DefaultRequestTimeout* = 5.seconds
# Finite fallback (issue #93) so a wedged handler can't hang a caller forever.
include ./event_thread
include ./ffi_thread
@ -105,6 +113,7 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
initEventQueue(ctx[].eventQueue)
ctx.ffiHeartbeat.store(0)
ctx.eventQueueStuck.store(false)
ctx.defaultRequestTimeout = DefaultRequestTimeout
var success = false
defer:
@ -120,6 +129,7 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
newSignalOrErr(ctx.eventThreadExitSignal, "eventThreadExitSignal")
ctx.registeredRequests = addr ffi_types.registeredRequests
ctx.requestTimeouts = addr ffi_types.requestTimeoutsMs
ctx.running.store(true)

View File

@ -44,6 +44,47 @@ proc sendRequestToFFIThread*(
ok()
func resolveRequestTimeout[T](reqIdCs: cstring, ctx: ptr FFIContext[T]): Duration =
## Per-proc `{.ffi: "timeout = <ms>".}` override if one was registered for this
## request type, otherwise the context-wide default.
let ms = ctx[].requestTimeouts[].getOrDefault(reqIdCs, 0)
if ms > 0: ms.milliseconds else: ctx.defaultRequestTimeout
proc reportTimeoutIfTripped(
retFut: Future[Result[seq[byte], string]],
request: ptr FFIThreadRequest,
deadline: Duration,
reqId: string,
) {.async.} =
## Waits for the handler or its deadline, whichever comes first. On a trip we
## deliberately do NOT cancel the handler: a hard-cancel mid-call into the
## underlying library (Waku/libp2p) can leave it partially applied, so we
## unblock the caller with a timeout err now and let the handler run to
## completion. `fireCallback`'s once-only guard keeps the two paths from
## answering twice.
if deadline == InfiniteDuration:
return
# Handlers that already completed (e.g. a sync body) skip the timer entirely,
# keeping the per-request cost off the fast path.
if retFut.finished():
return
let timer = sleepAsync(deadline)
# `race` returns the first to finish WITHOUT cancelling the loser, so the
# handler keeps running when the timer wins.
discard await race(retFut, timer)
if not timer.finished():
await timer.cancelAndWait()
if retFut.finished():
return
warn "ffi request timed out; caller unblocked, handler left running",
reqId = reqId, timeoutMs = deadline.milliseconds
fireCallback(
Result[seq[byte], string].err(
"ffi request timed out after " & $deadline.milliseconds & "ms"
),
request,
)
proc processRequest[T](
request: ptr FFIThreadRequest, ctx: ptr FFIContext[T]
) {.async.} =
@ -59,9 +100,15 @@ proc processRequest[T](
else:
ctx[].registeredRequests[][reqIdCs](cast[pointer](request), ctx)
# CatchableError covers CancelledError from the shutdown drain; handleRes must still run.
# CatchableError covers CancelledError from the shutdown drain; handleRes must
# still run, so the timeout race and the handler await share one try — a cancel
# mid-race must not skip the response-and-free below.
let res =
try:
# May answer the caller early with a timeout err; the handler keeps running.
await reportTimeoutIfTripped(
retFut, request, resolveRequestTimeout(reqIdCs, ctx), reqId
)
await retFut
except CatchableError as e:
Result[seq[byte], string].err(

View File

@ -39,6 +39,9 @@ type FFIThreadRequest* = object
## Intrusive ingress-queue link (see `ffi_request_queue.nim`). Touched only
## under the queue's lock; the request doubles as its own node, so no
## separate node alloc lands on the per-thread ORC MemRegion.
responded*: bool
## De-duplicates the callback across the timeout and completion paths. Both
## run on the FFI thread, so a plain flag suffices — no cross-thread race.
func ffiPackScalar*[T](x: T): uint64 =
## Bit-cast one scalar into a `uint64` request slot. Signed ints sign-extend
@ -77,6 +80,7 @@ proc allocBaseRequest(
ret[].data = nil
ret[].dataLen = 0
ret[].next = nil
ret[].responded = false
return ret
proc copySharedPayload(req: ptr FFIThreadRequest, data: ptr byte, dataLen: int) =
@ -200,12 +204,16 @@ proc deleteRequest*(request: ptr FFIThreadRequest) =
c_free(cast[pointer](request[].reqId))
c_free(request)
proc handleRes*(res: Result[seq[byte], string], request: ptr FFIThreadRequest) =
## Fires the registered callback exactly once and frees the request.
## Success payload is CBOR bytes; error payload is the raw UTF-8 error string.
defer:
deleteRequest(request)
proc fireCallback*(res: Result[seq[byte], string], request: ptr FFIThreadRequest) =
## Delivers the response to the foreign callback, at most once per request:
## the timeout path and the handler-completion path both call it, but the
## foreign side must be answered exactly once. Both run on the FFI thread, so
## the plain `responded` flag needs no synchronization. Success payload is the
## encoded response bytes; error payload is the raw UTF-8 error string. Does
## NOT free the request; that stays with `handleRes`.
if request[].responded:
return
request[].responded = true
if res.isErr():
foreignThreadGc:
let msg = if res.error.len > 0: res.error else: EmptyErrorMarker
@ -230,5 +238,12 @@ proc handleRes*(res: Result[seq[byte], string], request: ptr FFIThreadRequest) =
RET_OK, cast[ptr cchar](addr sentinel), 1.csize_t, request[].userData
)
proc handleRes*(res: Result[seq[byte], string], request: ptr FFIThreadRequest) =
## Terminal step of every request: delivers the response (unless a timeout
## already did) and frees the request exactly once.
defer:
deleteRequest(request)
fireCallback(res, request)
proc nilProcess*(reqId: cstring): Future[Result[seq[byte], string]] {.async.} =
return err("This request type is not implemented: " & $reqId)

View File

@ -37,5 +37,11 @@ template foreignThreadGc*(body: untyped) =
## The value is a proc that handles the request asynchronously.
var registeredRequests*: Table[cstring, FFIRequestProc]
## Per-request handler-timeout overrides in milliseconds, keyed by the same Req
## type name as `registeredRequests`. Populated at compile time from a
## `{.ffi: "timeout = <ms>".}` spec; an absent key means "use the context's
## `defaultRequestTimeout`". Like `registeredRequests`, never mutated at run time.
var requestTimeoutsMs*: Table[cstring, int]
### End of FFI utils
################################################################################

View File

@ -24,17 +24,63 @@ proc resolveABIFormat(abiSpecs: seq[NimNode]): ABIFormat {.compileTime.} =
## Resolve one annotation's ABI from its optional `"abi = ..."` string specs
## (last wins), inheriting the library default when absent.
var fmt = currentDefaultABIFormat
for spec in abiSpecs:
if spec.kind notin {nnkStrLit, nnkRStrLit, nnkTripleStrLit}:
for override in abiSpecs:
if override.kind notin {nnkStrLit, nnkRStrLit, nnkTripleStrLit}:
error(
"FFI ABI override must be a string literal like \"abi = c\", got: " & spec.repr
"FFI ABI override must be a string literal like \"abi = c\", got: " &
override.repr
)
let parsed = parseAbiSpec($spec)
let parsed = parseAbiSpec($override)
if not parsed.ok:
error(parsed.err)
fmt = parsed.fmt
fmt
proc resolveFFISpecs(
specs: seq[NimNode]
): tuple[abi: ABIFormat, timeoutMs: int] {.compileTime.} =
## Resolve an annotation's `"abi = ..."` and `"timeout = ..."` string specs
## (last of each wins), inheriting the library-default ABI when absent.
## `timeoutMs == 0` means "no per-proc override" (use the context default).
var abi = currentDefaultABIFormat
var timeoutMs = 0
for override in specs:
if override.kind notin {nnkStrLit, nnkRStrLit, nnkTripleStrLit}:
error(
"FFI override must be a string literal like \"abi = c\" or " &
"\"timeout = 30000\", got: " & override.repr
)
case overrideKey($override)
of "abi":
let parsed = parseAbiSpec($override)
if not parsed.ok:
error(parsed.err)
abi = parsed.fmt
of "timeout":
let parsed = parseTimeoutSpec($override)
if not parsed.ok:
error(parsed.err)
timeoutMs = parsed.ms
else:
error(
"unknown FFI override '" & $override &
"'; expected `abi = ...` or `timeout = ...`"
)
(abi, timeoutMs)
proc registerRequestTimeout(
reqTypeName: NimNode, timeoutMs: int
): NimNode {.compileTime.} =
## Top-level assignment that records a per-proc handler timeout at module init,
## keyed by the same Req type name the dispatcher registry uses. Empty when no
## override was given.
if timeoutMs <= 0:
return newStmtList()
newAssignment(
newTree(nnkBracketExpr, ident("requestTimeoutsMs"), newLit($reqTypeName)),
newLit(timeoutMs),
)
proc gateABIFormat(fmt: ABIFormat, where: string) {.compileTime.} =
## Abort if the selected ABI's codegen isn't wired yet (only `Cbor` is), so a
## `c` request fails loudly instead of emitting CBOR mislabeled as C.
@ -606,7 +652,8 @@ macro ffiRaw*(args: varargs[untyped]): untyped =
requireLibraryDeclared("`.ffiRaw.`")
let prc = args[^1]
gateABIFormat(resolveABIFormat(args[0 ..^ 2]), "`.ffiRaw.` proc")
let (rawAbiFormat, rawTimeoutMs) = resolveFFISpecs(args[0 ..^ 2])
gateABIFormat(rawAbiFormat, "`.ffiRaw.` proc")
let procName = prc[0]
let formalParams = prc[3]
@ -679,7 +726,8 @@ macro ffiRaw*(args: varargs[untyped]): untyped =
registerReqFFI(`reqName`, `paramIdent`: `paramType`):
`anonymousProcNode`
let stmts = newStmtList(registerReq, ffiProc)
let stmts =
newStmtList(registerReq, ffiProc, registerRequestTimeout(reqName, rawTimeoutMs))
when defined(ffiDumpMacros):
echo stmts.repr
@ -749,14 +797,16 @@ macro ffi*(args: varargs[untyped]): untyped =
## proc mylib_send*(w: MyLib, cfg: SendConfig): Future[Result[string, string]] {.ffi.} =
## return ok("done")
# Annotated node is the last vararg; leading args are `"abi = ..."` specs.
# Annotated node is the last vararg; leading args are override specs.
let prc = args[^1]
let abiFormat = resolveABIFormat(args[0 ..^ 2])
let (abiFormat, timeoutMs) = resolveFFISpecs(args[0 ..^ 2])
# A value type stands alone (no library required). Its `c` companion is
# emitted later by `genBindings()`, since a type-pragma macro can only return
# a TypeDef; `cbor` rides the generic overloads. Both abis are valid here.
if prc.kind == nnkTypeDef:
if timeoutMs > 0:
error("`.ffi.` on a type takes no `timeout` override (it only applies to procs)")
gateFFITypeABIFormat(abiFormat, "`.ffi.` type")
var cleanTypeDef = prc.copyNimTree()
if cleanTypeDef[0].kind == nnkPragmaExpr:
@ -1011,7 +1061,9 @@ macro ffi*(args: varargs[untyped]): untyped =
ffiProcRegistry.add(procMeta)
return newStmtList(helperProc, registerReq, ffiProc)
return newStmtList(
helperProc, registerReq, ffiProc, registerRequestTimeout(reqTypeName, timeoutMs)
)
proc scalarPath(): NimNode =
## The scalar fast path lives in `ffi_scalar`; here we only build the shared
@ -1288,7 +1340,7 @@ macro ffiCtor*(args: varargs[untyped]): untyped =
requireLibraryDeclared("`.ffiCtor.`")
let prc = args[^1]
let abiFormat = resolveABIFormat(args[0 ..^ 2])
let (abiFormat, timeoutMs) = resolveFFISpecs(args[0 ..^ 2])
gateABIFormat(abiFormat, "`.ffiCtor.` proc")
let procName = prc[0]
@ -1458,7 +1510,14 @@ macro ffiCtor*(args: varargs[untyped]): untyped =
var `poolIdent`: FFIContextPool[`libTypeName`]
let stmts = newStmtList(
typeDef, ffiNewReqProc, helperProc, processProc, addToReg, poolDecl, ffiProc
typeDef,
ffiNewReqProc,
helperProc,
processProc,
addToReg,
poolDecl,
ffiProc,
registerRequestTimeout(reqTypeName, timeoutMs),
)
when defined(ffiDumpMacros):

View File

@ -38,6 +38,13 @@ proc abitest_echo*(
): Future[Result[int, string]] {.ffi: "abi = cbor".} =
return ok(n)
# Per-proc handler-timeout override (issue #93): parsed like the abi spec and
# recorded in `requestTimeoutsMs`, keyed by the generated Req type name.
proc abitest_slow*(
lib: AbiLib, n: int
): Future[Result[int, string]] {.ffi: "timeout = 30000".} =
return ok(n)
# Event with an explicit ABI override passed after the wire name.
proc abitest_pinged*(p: Pinged) {.ffiEvent("on_pinged", "abi = cbor").}
@ -91,6 +98,30 @@ suite "ABI format parsing":
check parseAbiSpec("abi = bson").ok == false # unknown format
check "bson" in parseAbiSpec("abi = bson").err
suite "handler-timeout spec parsing (issue #93)":
test "overrideKey extracts the lowercased, trimmed key":
check overrideKey("timeout = 30000") == "timeout"
check overrideKey(" ABI = c ") == "abi"
check overrideKey("bare") == "bare"
test "parseTimeoutSpec accepts `timeout = <ms>`, flexible spacing":
check parseTimeoutSpec("timeout = 30000") == (true, 30000, "")
check parseTimeoutSpec("TIMEOUT=100").ms == 100
check parseTimeoutSpec(" timeout = 5 ").ms == 5
test "parseTimeoutSpec rejects malformed specs and non-positive values":
check parseTimeoutSpec("30000").ok == false # missing `timeout =`
check parseTimeoutSpec("abi = c").ok == false # wrong key
check parseTimeoutSpec("timeout = 1 = 2").ok == false # too many `=`
check parseTimeoutSpec("timeout = abc").ok == false # not an integer
check parseTimeoutSpec("timeout = 0").ok == false # must be positive
check parseTimeoutSpec("timeout = -5").ok == false # must be positive
test "a `timeout` override is recorded; a plain proc has no entry":
# Populated at module init from the annotations above.
check requestTimeoutsMs["AbitestSlowReq".cstring] == 30000
check not requestTimeoutsMs.hasKey("AbitestPingReq".cstring)
suite "ABI proc-dispatch readiness (why c is still gated on procs)":
test "cbor proc-dispatch is wired; c proc-dispatch is gated":
# This predicate is what the proc-form macros consult: `cbor` is wired

View File

@ -11,6 +11,7 @@ type CallbackData = object
lock: Lock
cond: Cond
called: bool
callCount: int
retCode: cint
msg: array[1024, byte]
msgLen: int
@ -34,6 +35,7 @@ proc testCallback(
copyMem(addr d[].msg[0], msg, n)
d[].msgLen = n
d[].called = true
inc d[].callCount
signal(d[].cond)
release(d[].lock)
@ -226,6 +228,12 @@ suite "destroyFFIContext refc workaround":
check false
return
# This case stresses the refc destroy workaround, not the request timeout:
# the 50k-allocation handler can outrun the finite default deadline on a slow
# sanitizer/ARM runner, tripping a spurious timeout err. Opt out so the check
# below observes the handler's real result.
ctx.defaultRequestTimeout = InfiniteDuration
var d: CallbackData
initCallbackData(d)
defer:
@ -610,3 +618,91 @@ suite "reentrancy guard (PR #23 review, item 6)":
let nestedMsg = gReentrantNestedRes.recv()
check nestedMsg.startsWith("err:")
check "reentrant ffi call" in nestedMsg
# Per-proc handler timeout (issue #93): a `{.ffi: "timeout = <ms>".}` override
# bounds how long a handler may run before the caller is unblocked with an err.
# The handler is NOT cancelled — it keeps running — so the callback must still
# fire exactly once.
type TimeoutConfig {.ffi.} = object
dummy: int
proc testlib_slow_timeout*(
lib: SimpleLib, cfg: TimeoutConfig
): Future[Result[string, string]] {.ffi: "timeout = 100".} =
await sleepAsync(500.milliseconds)
return ok("slow-timeout-done")
proc testlib_under_deadline*(
lib: SimpleLib, cfg: TimeoutConfig
): Future[Result[string, string]] {.ffi: "timeout = 1000".} =
await sleepAsync(50.milliseconds)
return ok("under-deadline-done")
proc createSimpleCtx(): ptr FFIContext[SimpleLib] =
## Spins up a SimpleLib context via the ctor and returns it (nil on failure).
var ctorD: CallbackData
initCallbackData(ctorD)
defer:
deinitCallbackData(ctorD)
var cfg = cborEncode(TestlibCreateCtorReq(config: SimpleConfig(initialValue: 1)))
let ctorRet =
testlib_create(encodedPtr(cfg), cfg.len.csize_t, testCallback, addr ctorD)
if ctorRet.isNil():
return nil
waitCallback(ctorD)
if ctorD.retCode != RET_OK:
return nil
let ctxAddr = ctorAddrFromCbor(callbackBytes(ctorD))
if ctxAddr == 0:
return nil
cast[ptr FFIContext[SimpleLib]](ctxAddr)
suite "per-proc request timeout (issue #93)":
test "handler past its deadline yields a timeout err, fired exactly once":
let ctx = createSimpleCtx()
check not ctx.isNil()
defer:
check SimpleLibFFIPool.destroyFFIContext(ctx).isOk()
var d: CallbackData
initCallbackData(d)
defer:
deinitCallbackData(d)
var reqBytes = cborEncode(TestlibSlowTimeoutReq(cfg: TimeoutConfig(dummy: 0)))
let ret = testlib_slow_timeout(
ctx, testCallback, addr d, encodedPtr(reqBytes), reqBytes.len.csize_t
)
check ret == RET_OK
waitCallback(d)
check d.retCode == RET_ERR
check "timed out" in callbackErr(d)
# The handler (500 ms) is still running past the 100 ms deadline; once it
# finishes it must NOT deliver a second callback.
os.sleep(700)
check d.callCount == 1
test "handler finishing under its deadline returns normally":
let ctx = createSimpleCtx()
check not ctx.isNil()
defer:
check SimpleLibFFIPool.destroyFFIContext(ctx).isOk()
var d: CallbackData
initCallbackData(d)
defer:
deinitCallbackData(d)
var reqBytes = cborEncode(TestlibUnderDeadlineReq(cfg: TimeoutConfig(dummy: 0)))
let ret = testlib_under_deadline(
ctx, testCallback, addr d, encodedPtr(reqBytes), reqBytes.len.csize_t
)
check ret == RET_OK
waitCallback(d)
check d.retCode == RET_OK
check cborDecode(callbackBytes(d), string).value == "under-deadline-done"
check d.callCount == 1