test(bench): submit-throughput scaling gate for sendRequestToFFIThread (issue #90) (#97)

2026-08-06 15:03:32 +00:00 · 2026-06-25 18:02:45 -03:00 · 2026-06-25 18:02:45 -03:00 · 06771f0a61
commit 06771f0a61
parent 64a332ca8b
15 changed files with 479 additions and 154 deletions
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@ -169,51 +169,15 @@ jobs:
          nimble test_cpp_e2e -y

  check-bindings:
+    # Single OS is enough — codegen output is platform-independent; the Nim
+    # matrix catches version-sensitive output (the PR #39 drift class).
    name: Check generated bindings
    needs: versions
-    # Codegen output is platform-independent — single OS is enough. Matrix
-    # over Nim versions to catch any version-sensitive output. Catches the
-    # class of drift surfaced in PR #39 (C++ regen committed, Rust
-    # overlooked); see `nimble check_bindings` in ffi.nimble.
-    strategy:
-      fail-fast: false
-      matrix:
-        nim-version: ${{ fromJSON(needs.versions.outputs.nim-versions) }}
-    runs-on: ubuntu-22.04
-    env:
-      NIMBLE_VERSION: ${{ needs.versions.outputs.nimble }}
-    steps:
-      - uses: actions/checkout@v4
-
-      - name: Setup Nim
-        uses: jiro4989/setup-nim-action@v2
-        with:
-          nim-version: ${{ matrix.nim-version }}
-          repo-token: ${{ secrets.GITHUB_TOKEN }}
-
-      - name: Install Nimble ${{ env.NIMBLE_VERSION }}
-        run: |
-          cd /tmp && nimble install "nimble@${{ env.NIMBLE_VERSION }}" -y
-          echo "$HOME/.nimble/bin" >> $GITHUB_PATH
-
-      - name: Cache nimble deps
-        id: cache-nimbledeps
-        uses: actions/cache@v4
-        with:
-          path: |
-            nimbledeps/
-            nimble.paths
-          key: ${{ runner.os }}-nimbledeps-${{ matrix.nim-version }}-${{ hashFiles('*.nimble') }}
-          restore-keys: |
-            ${{ runner.os }}-nimbledeps-${{ matrix.nim-version }}-
-            ${{ runner.os }}-nimbledeps-
-
-      - name: Install nimble deps
-        if: steps.cache-nimbledeps.outputs.cache-hit != 'true'
-        run: nimble setup --localdeps -y
-
-      - name: Verify checked-in bindings match generator output
-        run: nimble check_bindings -y
+    uses: ./.github/workflows/nimble-job.yml
+    with:
+      run: nimble check_bindings -y
+      nim-versions:   ${{ needs.versions.outputs.nim-versions }}
+      nimble-version: ${{ needs.versions.outputs.nimble }}

  tests-asan-ubsan:
    name: Tests · ASan+UBSan+LSan
@ -233,6 +197,21 @@ jobs:
      nim-versions:   ${{ needs.versions.outputs.nim-versions }}
      nimble-version: ${{ needs.versions.outputs.nimble }}

+  submit-scaling-gate:
+    # Forcing function, red by design: asserts sendRequestToFFIThread submit
+    # throughput scales with producer-thread count. The per-request global lock
+    # serialises every submit, so this stays red until the lock is replaced with
+    # MPSC ingress — a standing reminder, not a transient failure. Pinned to orc +
+    # unsanitized because the gate is timing-based and the contention it measures
+    # is mm-independent. Full rationale and baseline numbers: tests/bench/README.md.
+    name: Submit Scaling Gate
+    needs: versions
+    uses: ./.github/workflows/nimble-job.yml
+    with:
+      run: NIM_FFI_MM=orc FFI_SUBMIT_PER_THREAD=20000 nimble bench_ffi_submit -y
+      nim-versions:   ${{ needs.versions.outputs.nim-versions }}
+      nimble-version: ${{ needs.versions.outputs.nimble }}
+
  auto-assign:
    name: Auto-assign PR author
    if: github.event_name == 'pull_request' && github.event.action == 'opened'
--- a/.github/workflows/nimble-job.yml
+++ b/.github/workflows/nimble-job.yml
@ -0,0 +1,66 @@
+name: nimble-job
+
+# Single-OS job that sets up the Nim/Nimble toolchain + cached deps, then runs
+# one shell command across the Nim-version matrix. Shared by the ci.yml jobs
+# whose only difference is the command they run (check-bindings, the submit
+# scaling gate, …). Multi-OS / sanitizer matrices live in test.yml /
+# tests-sanitized.yml instead.
+on:
+  workflow_call:
+    inputs:
+      run:
+        required: true
+        type: string
+        description: Shell command to run once the toolchain and deps are ready.
+      nim-versions:
+        required: true
+        type: string
+        description: JSON array of Nim versions to matrix over.
+      nimble-version:
+        required: true
+        type: string
+      runs-on:
+        required: false
+        type: string
+        default: ubuntu-22.04
+
+jobs:
+  run:
+    strategy:
+      fail-fast: false
+      matrix:
+        nim-version: ${{ fromJSON(inputs.nim-versions) }}
+    runs-on: ${{ inputs.runs-on }}
+    name: Nim ${{ matrix.nim-version }}
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Setup Nim
+        uses: jiro4989/setup-nim-action@v2
+        with:
+          nim-version: ${{ matrix.nim-version }}
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Install Nimble ${{ inputs.nimble-version }}
+        run: |
+          cd /tmp && nimble install "nimble@${{ inputs.nimble-version }}" -y
+          echo "$HOME/.nimble/bin" >> $GITHUB_PATH
+
+      - name: Cache nimble deps
+        id: cache-nimbledeps
+        uses: actions/cache@v4
+        with:
+          path: |
+            nimbledeps/
+            nimble.paths
+          key: ${{ runner.os }}-nimbledeps-${{ matrix.nim-version }}-${{ hashFiles('*.nimble') }}
+          restore-keys: |
+            ${{ runner.os }}-nimbledeps-${{ matrix.nim-version }}-
+            ${{ runner.os }}-nimbledeps-
+
+      - name: Install nimble deps
+        if: steps.cache-nimbledeps.outputs.cache-hit != 'true'
+        run: nimble setup --localdeps -y
+
+      - name: Run
+        run: ${{ inputs.run }}
--- a/.github/workflows/tests-sanitized.yml
+++ b/.github/workflows/tests-sanitized.yml
@ -89,5 +89,15 @@ jobs:
      - name: Run unit tests (${{ inputs.sanitizer }})
        run: nimble test_sanitized -y

+      # Correctness only here (exactly-once, no leaks/races). The scaling gate is
+      # off: sanitizers distort thread timing, so throughput scaling is measured
+      # in the non-sanitized Submit Scaling Gate job (ci.yml) instead.
+      - name: Run sendRequestToFFIThread submit stress (${{ inputs.sanitizer }})
+        env:
+          FFI_SUBMIT_PER_THREAD: 2000
+          FFI_SUBMIT_ITERS: 1
+          FFI_SCALING_GATE: 0
+        run: nimble bench_ffi_submit -y
+
      - name: Run C++ e2e tests (${{ inputs.sanitizer }})
        run: nimble test_cpp_e2e_sanitized -y
--- a/examples/timer/cpp_bindings/my_timer.hpp
+++ b/examples/timer/cpp_bindings/my_timer.hpp
@ -146,17 +146,20 @@ inline CborError encode_cbor(CborEncoder& e, const std::optional<T>& v) {

 // ── decode_cbor overloads ───────────────────────────────────────────────

-inline CborError decode_cbor(CborValue& it, bool& out) {
-    if (!cbor_value_is_boolean(&it)) return CborErrorImproperValue;
-    CborError err = cbor_value_get_boolean(&it, &out);
+// After reading a leaf value, the parser must advance past it; both steps
+// short-circuit on the same CborError, so they always travel together.
+inline CborError advance_if_ok(CborValue& it, CborError err) {
    if (err) return err;
    return cbor_value_advance(&it);
 }
+
+inline CborError decode_cbor(CborValue& it, bool& out) {
+    if (!cbor_value_is_boolean(&it)) return CborErrorImproperValue;
+    return advance_if_ok(it, cbor_value_get_boolean(&it, &out));
+}
 inline CborError decode_cbor(CborValue& it, int64_t& out) {
    if (!cbor_value_is_integer(&it)) return CborErrorImproperValue;
-    CborError err = cbor_value_get_int64_checked(&it, &out);
-    if (err) return err;
-    return cbor_value_advance(&it);
+    return advance_if_ok(it, cbor_value_get_int64_checked(&it, &out));
 }
 inline CborError decode_cbor(CborValue& it, int32_t& out) {
    int64_t tmp = 0;
@ -167,15 +170,11 @@ inline CborError decode_cbor(CborValue& it, int32_t& out) {
 }
 inline CborError decode_cbor(CborValue& it, uint64_t& out) {
    if (!cbor_value_is_unsigned_integer(&it)) return CborErrorImproperValue;
-    CborError err = cbor_value_get_uint64(&it, &out);
-    if (err) return err;
-    return cbor_value_advance(&it);
+    return advance_if_ok(it, cbor_value_get_uint64(&it, &out));
 }
 inline CborError decode_cbor(CborValue& it, double& out) {
    if (cbor_value_is_double(&it)) {
-        CborError err = cbor_value_get_double(&it, &out);
-        if (err) return err;
-        return cbor_value_advance(&it);
+        return advance_if_ok(it, cbor_value_get_double(&it, &out));
    }
    if (cbor_value_is_float(&it)) {
        float f = 0.0f;
@ -192,9 +191,8 @@ inline CborError decode_cbor(CborValue& it, std::string& out) {
    CborError err = cbor_value_get_string_length(&it, &len);
    if (err) return err;
    out.resize(len);
-    err = cbor_value_copy_text_string(&it, out.empty() ? nullptr : &out[0], &len, nullptr);
-    if (err) return err;
-    return cbor_value_advance(&it);
+    return advance_if_ok(
+        it, cbor_value_copy_text_string(&it, out.empty() ? nullptr : &out[0], &len, nullptr));
 }

 template<typename T>
@ -223,10 +221,8 @@ inline CborError decode_cbor(CborValue& it, std::vector<std::uint8_t>& out) {
    CborError err = cbor_value_get_string_length(&it, &len);
    if (err) return err;
    out.resize(len);
-    err = cbor_value_copy_byte_string(
-        &it, out.empty() ? nullptr : out.data(), &len, nullptr);
-    if (err) return err;
-    return cbor_value_advance(&it);
+    return advance_if_ok(
+        it, cbor_value_copy_byte_string(&it, out.empty() ? nullptr : out.data(), &len, nullptr));
 }

 template<typename T>
--- a/examples/timer/rust_bindings/src/api.rs
+++ b/examples/timer/rust_bindings/src/api.rs
@ -129,12 +129,13 @@ pub struct MyTimerCtx {

 // SAFETY: The `ptr` field points to an FFIContext owned by the Nim runtime.
 // Every call through the generated FFI proc goes through
-// `sendRequestToFFIThread` on the Nim side, which serialises every request
-// behind `ctx.lock` and dispatches handlers on a single FFI thread, so the
-// pointer is never accessed concurrently from Rust. The Nim-side reentrancy
-// guard (`onFFIThread` threadvar) prevents handlers from re-entering the
-// dispatcher and self-deadlocking. These invariants make it sound to mark
-// the wrapper as Send + Sync.
+// `sendRequestToFFIThread` on the Nim side, which only enqueues the request
+// onto a lock-free MPSC queue (sound from any number of threads) and wakes
+// the single FFI thread that dispatches every handler. The context is thus
+// never mutated non-atomically from the caller's thread. The Nim-side
+// reentrancy guard (`onFFIThread` threadvar) prevents handlers from
+// re-entering the dispatcher. These invariants make it sound to mark the
+// wrapper as Send + Sync.
 unsafe impl Send for MyTimerCtx {}
 unsafe impl Sync for MyTimerCtx {}

--- a/ffi.nimble
+++ b/ffi.nimble
@ -65,6 +65,25 @@ proc sanFlags(san: string): string =
  else:
    raise newException(ValueError, "unknown NIM_FFI_SAN: " & san)

+proc mmModes(): seq[string] =
+  ## Memory-management modes to build under, selected by NIM_FFI_MM (empty = both).
+  case getEnv("NIM_FFI_MM", "")
+  of "orc":
+    @[nimFlagsOrc]
+  of "refc":
+    @[nimFlagsRefc]
+  else:
+    @[nimFlagsOrc, nimFlagsRefc]
+
+proc applyTsanSuppressions() =
+  ## Adds tsan.supp to TSAN_OPTIONS without clobbering options the CI job set.
+  let suppPath = thisDir() & "/tsan.supp"
+  let existing = getEnv("TSAN_OPTIONS")
+  if existing == "":
+    putEnv("TSAN_OPTIONS", "suppressions=" & suppPath)
+  elif "suppressions=" notin existing:
+    putEnv("TSAN_OPTIONS", existing & ":suppressions=" & suppPath)
+
 task buildffi, "Compile the library":
  exec "nim c " & nimFlagsOrc & " --app:lib --noMain ffi.nim"

@ -90,6 +109,17 @@ task bench_codec, "Microbenchmark: cbor vs c (cwire) wire-format codecs":
  # debug build. Not part of `test` — timing is a measurement, not a gate.
  exec "nim c -r " & nimFlagsOrc & " -d:danger tests/bench/bench_codec.nim"

+task bench_ffi_submit,
+  "Concurrent-submit stress + scaling gate for sendRequestToFFIThread":
+  # Honors NIM_FFI_SAN / NIM_FFI_MM like test_sanitized so CI drives it under
+  # asan-ubsan and tsan; FFI_SUBMIT_PER_THREAD sets per-thread volume.
+  let san = getEnv("NIM_FFI_SAN", "none")
+  let extra = sanFlags(san)
+  if san == "tsan":
+    applyTsanSuppressions()
+  for flags in mmModes():
+    exec "nim c -r " & flags & " -d:danger" & extra & " tests/bench/bench_ffi_submit.nim"
+
 task test_cpp_e2e, "Build and run the C++ end-to-end tests for the timer example":
  # Regenerate the C++ bindings so the suite always runs against fresh codegen.
  runOrQuit "nimble genbindings_cpp"
@ -104,23 +134,10 @@ task test_cpp_e2e, "Build and run the C++ end-to-end tests for the timer example
 task test_sanitized,
  "Run all unit tests under a sanitizer (NIM_FFI_SAN) and mm (NIM_FFI_MM)":
  let san = getEnv("NIM_FFI_SAN", "none")
-  let mm = getEnv("NIM_FFI_MM", "")
  let extra = sanFlags(san)
-  let modes =
-    if mm == "orc":
-      @[nimFlagsOrc]
-    elif mm == "refc":
-      @[nimFlagsRefc]
-    else:
-      @[nimFlagsOrc, nimFlagsRefc]
  if san == "tsan":
-    let suppPath = thisDir() & "/tsan.supp"
-    let existing = getEnv("TSAN_OPTIONS")
-    if existing == "":
-      putEnv("TSAN_OPTIONS", "suppressions=" & suppPath)
-    elif "suppressions=" notin existing:
-      putEnv("TSAN_OPTIONS", existing & ":suppressions=" & suppPath)
-  for flags in modes:
+    applyTsanSuppressions()
+  for flags in mmModes():
    for t in unitTests:
      exec "nim c -r " & flags & extra & " tests/unit/" & t & ".nim"

--- a/ffi/codegen/rust.nim
+++ b/ffi/codegen/rust.nim
@ -505,19 +505,18 @@ proc generateApiRs*(
  )
  lines.add("// Every call through the generated FFI proc goes through")
  lines.add(
-    "// `sendRequestToFFIThread` on the Nim side, which serialises every request"
+    "// `sendRequestToFFIThread` on the Nim side, which only enqueues the request"
  )
  lines.add(
-    "// behind `ctx.lock` and dispatches handlers on a single FFI thread, so the"
+    "// onto a lock-free MPSC queue (sound from any number of threads) and wakes"
  )
  lines.add(
-    "// pointer is never accessed concurrently from Rust. The Nim-side reentrancy"
+    "// the single FFI thread that dispatches every handler. The context is thus"
  )
-  lines.add("// guard (`onFFIThread` threadvar) prevents handlers from re-entering the")
-  lines.add(
-    "// dispatcher and self-deadlocking. These invariants make it sound to mark"
-  )
-  lines.add("// the wrapper as Send + Sync.")
+  lines.add("// never mutated non-atomically from the caller's thread. The Nim-side")
+  lines.add("// reentrancy guard (`onFFIThread` threadvar) prevents handlers from")
+  lines.add("// re-entering the dispatcher. These invariants make it sound to mark the")
+  lines.add("// wrapper as Send + Sync.")
  lines.add("unsafe impl Send for $1 {}" % [ctxTypeName])
  lines.add("unsafe impl Sync for $1 {}" % [ctxTypeName])
  lines.add("")
--- a/ffi/ffi_context.nim
+++ b/ffi/ffi_context.nim
@ -7,12 +7,13 @@
 {.passc: "-fPIC".}

 import std/[atomics, locks, options, tables]
-import chronicles, chronos, chronos/threadsync, taskpools/channels_spsc_single, results
+import chronicles, chronos, chronos/threadsync, results
 import
  ./ffi_types,
  ./ffi_events,
  ./ffi_handles,
  ./ffi_thread_request,
+  ./ffi_request_queue,
  ./logging,
  ./cbor_serial

@ -22,10 +23,8 @@ type FFIContext*[T] = object
  myLib*: ptr T # main library object (Waku, LibP2P, SDS, …)
  ffiThread: Thread[(ptr FFIContext[T])]
  eventThread: Thread[(ptr FFIContext[T])]
-  lock: Lock
-  reqChannel: ChannelSPSCSingle[ptr FFIThreadRequest]
-  reqSignal: ThreadSignalPtr
-  reqReceivedSignal: ThreadSignalPtr
+  reqQueue: FFIRequestQueue # lock-free MPSC ingress from foreign threads
+  reqSignal: ThreadSignalPtr # wakes the FFI thread on enqueue
  stopSignal: ThreadSignalPtr
  threadExitSignal: ThreadSignalPtr
    # bounds destroyFFIContext's wait so a blocked loop cannot hang the caller
@ -60,9 +59,10 @@ template closeAndNil(field: untyped) =
    field = nil

 proc deinitContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
-  ## Mirror of `initContextResources`. Threads MUST be joined first;
-  ## fields are nil'd after close so re-init on the same slot is safe.
-  ctx.lock.deinitLock()
+  ## Mirror of `initContextResources`. Threads MUST be joined first (so the
+  ## FFI thread has drained every queued request); fields are nil'd after close
+  ## so re-init on the same slot is safe. `reqQueue`'s stub is inline, so it
+  ## needs no teardown — re-init rebinds head/tail on the next createFFIContext.
  deinitEventRegistry(ctx[].eventRegistry)
  deinitHandleRegistry(ctx[].handles)
  deinitEventQueue(ctx[].eventQueue)
@ -74,7 +74,6 @@ proc deinitContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
    discard
  else:
    closeAndNil(ctx.reqSignal)
-    closeAndNil(ctx.reqReceivedSignal)
    closeAndNil(ctx.stopSignal)
    closeAndNil(ctx.threadExitSignal)
    closeAndNil(ctx.eventQueueSignal)
@ -96,12 +95,11 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
  ## the slot (freeShared or pool.releaseSlot).
  # Nil first so deferred cleanup can't double-close a reused pool slot.
  ctx.reqSignal = nil
-  ctx.reqReceivedSignal = nil
  ctx.stopSignal = nil
  ctx.threadExitSignal = nil
  ctx.eventQueueSignal = nil
  ctx.eventThreadExitSignal = nil
-  ctx.lock.initLock()
+  initRequestQueue(ctx[].reqQueue)
  initEventRegistry(ctx[].eventRegistry)
  initHandleRegistry(ctx[].handles)
  initEventQueue(ctx[].eventQueue)
@ -116,7 +114,6 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
          error = error

  newSignalOrErr(ctx.reqSignal, "reqSignal")
-  newSignalOrErr(ctx.reqReceivedSignal, "reqReceivedSignal")
  newSignalOrErr(ctx.stopSignal, "stopSignal")
  newSignalOrErr(ctx.threadExitSignal, "threadExitSignal")
  newSignalOrErr(ctx.eventQueueSignal, "eventQueueSignal")
--- a/ffi/ffi_request_queue.nim
+++ b/ffi/ffi_request_queue.nim
@ -0,0 +1,57 @@
+## Intrusive multi-producer / single-consumer queue of `ptr FFIThreadRequest`
+## (Vyukov's MPSC algorithm). It replaces the single-slot SPSC channel plus the
+## global submit lock that used to serialise every foreign-thread call: foreign
+## threads `push` concurrently with a single atomic exchange (wait-free), while
+## the FFI thread is the sole `pop`per.
+##
+## The request *is* the queue node — its `next` link is reused intrusively — so
+## enqueueing allocates nothing. The one node the algorithm needs of its own,
+## the stub sentinel, is embedded inline in the queue (hence inline in the
+## owning `FFIContext`): it carries no separate heap allocation, so a context
+## that is intentionally abandoned (wedged-thread shutdown) leaks nothing extra,
+## and it never sits on a per-thread ORC `MemRegion` the way `allocShared` would
+## (the dangling-TLS hazard documented in `ffi_thread_request.nim`).
+
+import std/atomics
+import ./ffi_thread_request
+
+type FFIRequestQueue* = object
+  head: Atomic[ptr FFIThreadRequest] ## producers swap here
+  tail: ptr FFIThreadRequest ## consumer-only cursor
+  stub: FFIThreadRequest ## inline dummy sentinel, never dispatched
+
+proc initRequestQueue*(q: var FFIRequestQueue) =
+  q.stub.next.store(nil, moRelaxed)
+  q.head.store(addr q.stub, moRelaxed)
+  q.tail = addr q.stub
+
+proc push*(q: var FFIRequestQueue, node: ptr FFIThreadRequest) =
+  ## Wait-free enqueue, safe from any number of producer threads.
+  node[].next.store(nil, moRelease)
+  let prev = q.head.exchange(node, moAcquireRelease)
+  prev[].next.store(node, moRelease)
+
+proc pop*(q: var FFIRequestQueue): ptr FFIThreadRequest =
+  ## Single-consumer dequeue. Returns nil when empty, or transiently when a
+  ## producer's exchange has linked a node but its `prev.next` store hasn't
+  ## landed yet — the caller (polling FFI thread) simply retries next wake.
+  let stub = addr q.stub
+  var tail = q.tail
+  var next = tail[].next.load(moAcquire)
+  if tail == stub:
+    if next.isNil():
+      return nil
+    q.tail = next
+    tail = next
+    next = next[].next.load(moAcquire)
+  if not next.isNil():
+    q.tail = next
+    return tail
+  if tail != q.head.load(moAcquire):
+    return nil
+  q.push(stub)
+  next = tail[].next.load(moAcquire)
+  if not next.isNil():
+    q.tail = next
+    return tail
+  nil
--- a/ffi/ffi_thread.nim
+++ b/ffi/ffi_thread.nim
@ -4,50 +4,40 @@
 ## and the `onFFIThread` threadvar. Companion to `event_thread.nim`.
 ##
 ## Responsibilities:
-## - Receive `FFIThreadRequest`s from foreign threads via `reqChannel` and
-##   dispatch them through the user-registered handler table.
+## - Receive `FFIThreadRequest`s from foreign threads via `reqQueue` (a
+##   lock-free MPSC queue) and dispatch them through the user-registered
+##   handler table.
 ## - Advance `ctx.ffiHeartbeat` each loop iteration so the event thread can
 ##   detect a wedged FFI thread.

 proc sendRequestToFFIThread*(
-    ctx: ptr FFIContext, ffiRequest: ptr FFIThreadRequest, timeout = InfiniteDuration
+    ctx: ptr FFIContext, ffiRequest: ptr FFIThreadRequest
 ): Result[void, string] =
  if ctx.eventQueueStuck.load():
    deleteRequest(ffiRequest)
    return err("event queue stuck - library cannot accept new requests")

  if onFFIThread:
-    # Re-entrant dispatch from a handler would self-deadlock on `reqReceivedSignal`.
+    # A handler re-dispatching onto its own FFI thread would enqueue work the
+    # blocked dispatcher can never drain; reject instead of dead-locking.
    deleteRequest(ffiRequest)
    return err(
      "reentrant ffi call: a handler invoked sendRequestToFFIThread on its own context"
    )

-  # Serialise the trySend + fireSync + waitSync — reqChannel is SP and reqReceivedSignal is shared.
-  ctx.lock.acquire()
-  defer:
-    ctx.lock.release()
+  # Lock-free hand-off: the queue is unbounded so enqueue can't fail, and the
+  # request's completion is reported asynchronously through its own callback —
+  # no shared accept-ack, so concurrent producers never serialise here.
+  ctx.reqQueue.push(ffiRequest)

-  let sentOk = ctx.reqChannel.trySend(ffiRequest)
-  if not sentOk:
-    deleteRequest(ffiRequest)
-    return err("Couldn't send a request to the ffi thread")
+  # A failed wake is non-fatal: the request is queued and the FFI thread's
+  # poll-drain dispatches it within one tick regardless of the signal. Return
+  # ok so the caller doesn't fire its error callback for a request that will
+  # still complete normally — returning err here would double-fire the callback.
+  ctx.reqSignal.fireSync().isOkOr:
+    error "failed to wake FFI thread after enqueue (request still queued)",
+      error = error

-  let fireSyncRes = ctx.reqSignal.fireSync()
-  if fireSyncRes.isErr():
-    deleteRequest(ffiRequest)
-    return err("failed fireSync: " & $fireSyncRes.error)
-
-  if fireSyncRes.get() == false:
-    deleteRequest(ffiRequest)
-    return err("Couldn't fireSync in time")
-
-  let res = ctx.reqReceivedSignal.waitSync(timeout)
-  if res.isErr():
-    # FFI thread was signaled and owns the request; don't double-free.
-    return err("Couldn't receive reqReceivedSignal signal")
-
-  # On ok the FFI thread's processRequest deallocShared(req)'s.
  ok()

 proc processRequest[T](
@ -122,30 +112,35 @@ proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
          continue
        pending.del(i)

+    proc drainQueue() =
+      ## Dispatch every request the producers have enqueued. fireSync coalesces,
+      ## so one wake can stand for many submits — drain to empty, not one per
+      ## wake, or queued requests would strand until the next signal.
+      while true:
+        let request = ctx.reqQueue.pop()
+        if request.isNil():
+          break
+        # Tick per dispatch so a large backlog can't flatline the heartbeat and
+        # trip the event thread's wedged-FFI-thread detection mid-drain.
+        discard ctx.ffiHeartbeat.fetchAdd(1)
+        if ctx.myLib.isNil():
+          ctx.myLib = addr ffiReqHandler
+        pending.add processRequest(request, ctx)
+
    while ctx.running.load():
      # Freezes if a sync handler blocks the dispatcher; event thread reads to detect wedged FFI thread.
      discard ctx.ffiHeartbeat.fetchAdd(1)

      reapCompleted()

-      let gotSignal = await ctx.reqSignal.wait().withTimeout(chronos.milliseconds(100))
-      if not gotSignal:
-        continue
+      # Drain regardless of the wait outcome: a missed/coalesced signal must not
+      # leave requests stranded, and the 100ms timeout bounds shutdown latency.
+      discard await ctx.reqSignal.wait().withTimeout(100.milliseconds)
+      drainQueue()

-      var request: ptr FFIThreadRequest
-      if not ctx.reqChannel.tryRecv(request):
-        continue
-
-      if ctx.myLib.isNil():
-        ctx.myLib = addr ffiReqHandler
-
-      pending.add processRequest(request, ctx)
-
-      let fireRes = ctx.reqReceivedSignal.fireSync()
-      if fireRes.isErr():
-        error "could not fireSync back to requester thread", error = fireRes.error
-
-    # Drain so each pending handler's deleteRequest defer runs before exit.
+    # Drain once more so requests enqueued just before `running` flipped still
+    # dispatch and each pending handler's deleteRequest defer runs before exit.
+    drainQueue()
    reapCompleted()
    if pending.len > 0:
      try:
--- a/ffi/ffi_thread_request.nim
+++ b/ffi/ffi_thread_request.nim
@ -10,6 +10,7 @@
 ## chunk's `owner` pointer dangles into reclaimed TLS and the deallocator
 ## segfaults. `malloc`/`free` are process-global and immune to that.

+import std/atomics
 import system/ansi_c
 import results
 import chronos
@ -26,6 +27,10 @@ type FFIThreadRequest* = object
  reqId*: cstring ## Per-proc Req type name used to look up the handler.
  data*: ptr UncheckedArray[byte] ## Owned CBOR-encoded request payload.
  dataLen*: int
+  next*: Atomic[ptr FFIThreadRequest]
+    ## Intrusive MPSC-queue link (see `ffi_request_queue.nim`). Touched only by
+    ## the queue's producers/consumer — the request doubles as its own node, so
+    ## no separate node alloc lands on the per-thread ORC MemRegion.

 proc allocBaseRequest(
    callback: FFICallBack, userData: pointer, reqId: cstring
@ -39,6 +44,7 @@ proc allocBaseRequest(
  ret[].reqId = reqId.alloc()
  ret[].data = nil
  ret[].dataLen = 0
+  ret[].next.store(nil, moRelaxed)
  return ret

 proc copySharedPayload(req: ptr FFIThreadRequest, data: ptr byte, dataLen: int) =
--- a/ffi/internal/ffi_macro.nim
+++ b/ffi/internal/ffi_macro.nim
@ -861,8 +861,8 @@ macro ffi*(args: varargs[untyped]): untyped =
    ## All `.ffi.` procs dispatch through the FFI thread channel and reply
    ## through the callback when the future resolves — the previous "sync
    ## fast-path" that ran inline on the foreign caller thread was removed
-    ## (PR #23 review, items 1–5) because it bypassed `foreignThreadGc`,
-    ## `ctx.lock`, and chronos's single-thread invariant.
+    ## (PR #23 review, items 1–5) because it bypassed `foreignThreadGc`, the
+    ## MPSC ingress hand-off, and chronos's single-thread invariant.
    let helperProc = buildAsyncHelperProc()

    # registerReqFFI lambda: typed params, returns user's typed Result.
--- a/tests/bench/README.md
+++ b/tests/bench/README.md
@ -1,3 +1,36 @@
+# FFI benchmarks
+
+This directory holds Nim micro/stress benchmarks. Neither is part of `nimble test`.
+
+- `bench_codec.nim` — `cbor` vs `c` (cwire) wire-format codec microbenchmark (documented below). Pure measurement, not a gate.
+- `bench_ffi_submit.nim` — concurrent-submit stress test + throughput benchmark for `sendRequestToFFIThread` (documented next). Carries a **scaling gate** that fails CI until the per-request submit lock is replaced.
+
+## `sendRequestToFFIThread` concurrent-submit stress / throughput
+
+`bench_ffi_submit.nim` motivates [issue #90](https://github.com/logos-messaging/nim-ffi/issues/90): every foreign-thread call serialises the whole `trySend + reqSignal.fireSync + reqReceivedSignal.waitSync` cycle under a single `ctx.lock`. The lock is load-bearing because `reqChannel` is single-slot and the accept handshake waits on a *shared* `reqReceivedSignal`, so producers cannot overlap.
+
+The bench fans **K producer threads (1 → 8)** at one context, each firing the same per-thread volume of no-op requests. It times the **submit phase only** — from the start gate until every producer returns from its last `sendRequestToFFIThread` — because that is the path the fix parallelises; completion is bounded by the single FFI thread and deliberately excluded. Each thread count runs `FFI_SUBMIT_ITERS` times (default 5) and the **median** submit/sec is reported, so run-to-run noise can't move the verdict.
+
+It is also a correctness stress test: the aggregate callback count must match the submit count **exactly** (no drops or double-fires), with zero submit errors and (under asan/lsan/tsan) zero leaks or races.
+
+```sh
+nimble bench_ffi_submit
+# smaller / faster (handy under sanitizers — they distort timing, so disable the gate):
+FFI_SUBMIT_PER_THREAD=2000 FFI_SUBMIT_ITERS=1 FFI_SCALING_GATE=0 nimble bench_ffi_submit
+# under a sanitizer (proves no leaks/races; gate off — see below):
+NIM_FFI_SAN=tsan FFI_SUBMIT_PER_THREAD=2000 FFI_SCALING_GATE=0 nimble bench_ffi_submit
+```
+
+Env knobs: `FFI_SUBMIT_PER_THREAD` (volume per producer, default 20000), `FFI_SUBMIT_ITERS` (median sample count, default 5), `FFI_SCALING_GATE` (default `1`; set `0` to report numbers without failing).
+
+### Scaling gate — red until the lock is replaced
+
+By default the bench **fails** (non-zero exit) unless submit throughput at 8 threads is at least `1.5x` the 1-thread rate. This is a forcing function: it cannot pass while `sendRequestToFFIThread` holds `ctx.lock` across the synchronous `reqReceivedSignal` accept, because that serialises every submit no matter how many producers run.
+
+Baseline measured 2026-06-24 (16-core Linux, orc, `-d:danger`, median of 5): submit scaling held at **0.98–1.16x** across threads — flat, as the lock dictates. `1.5x` sits above that noise ceiling (so the lock-bound code fails reliably) and well below the `>=2x` that parallel lock-free MPSC ingress yields on any multicore host (so the fix clears it with margin). Once it lands and this turns green, keep the gate as a regression guard.
+
+The gate runs in the non-sanitized **Submit Scaling Gate** CI job (`.github/workflows/ci.yml`); the sanitized jobs run the same bench with `FFI_SCALING_GATE=0` for leak/race coverage only, since sanitizer instrumentation makes throughput scaling meaningless.
+
 # FFI wire-format codec benchmark

 `bench_codec.nim` is a single-process Nim microbenchmark comparing the two FFI
--- a/tests/bench/bench_ffi_submit.nim
+++ b/tests/bench/bench_ffi_submit.nim
@ -0,0 +1,170 @@
+## Concurrent-submit stress test + throughput bench for `sendRequestToFFIThread`,
+## motivating its per-request submit lock. See tests/bench/README.md for the why.
+
+import std/[atomics, algorithm, strutils, os]
+import results
+import ../../ffi # chronos (Moment/Duration) and the FFI surface both arrive here.
+
+type BenchLib = object
+
+registerReqFFI(NoopRequest, lib: ptr BenchLib):
+  proc(): Future[Result[string, string]] {.async.} =
+    return ok("ok")
+
+var gStart: Atomic[bool]
+var gCompleted: Atomic[int] ## bumped once per callback; also the callback userData
+var gSendErrors: Atomic[int]
+
+let settleTimeout = 30.seconds
+
+## Forcing gate: min submit-throughput scaling (max-threads / 1-thread); red
+## until the per-request submit lock is replaced. See README "Scaling gate".
+const RequiredScaling = 1.5
+
+proc benchCallback(
+    retCode: cint, msg: ptr cchar, len: csize_t, userData: pointer
+) {.cdecl, gcsafe, raises: [].} =
+  let counter = cast[ptr Atomic[int]](userData)
+  discard counter[].fetchAdd(1)
+
+type ProducerArg = object
+  ctx: ptr FFIContext[BenchLib]
+  count: int
+
+proc producerBody(arg: ptr ProducerArg) {.thread, gcsafe.} =
+  while not gStart.load():
+    discard
+  for _ in 0 ..< arg[].count:
+    let req = NoopRequest.ffiNewReq(benchCallback, addr gCompleted)
+    if sendRequestToFFIThread(arg[].ctx, req).isErr():
+      discard gSendErrors.fetchAdd(1)
+
+proc waitForCompletions(target: int): bool =
+  ## Spins until `gCompleted` reaches `target`, bounded by `settleTimeout`.
+  let deadline = Moment.now() + settleTimeout
+  while gCompleted.load() < target:
+    if Moment.now() > deadline:
+      return false
+    os.sleep(1)
+  true
+
+proc median(xs: seq[float]): float =
+  if xs.len == 0:
+    return 0.0
+  let s = xs.sorted()
+  if s.len mod 2 == 1:
+    return s[s.len div 2]
+  (s[s.len div 2 - 1] + s[s.len div 2]) / 2.0
+
+type IterResult = object
+  submitRate: float ## submits/sec over the submit phase only (sends issued)
+  sendErrors: int
+  overruns: int ## callbacks beyond `total` — must be 0 (no double-fire)
+
+proc runOnce(
+    pool: var FFIContextPool[BenchLib], numThreads, perThread: int
+): IterResult =
+  let ctx = pool.createFFIContext().valueOr:
+    quit("createFFIContext failed: " & $error)
+  defer:
+    discard pool.destroyFFIContext(ctx)
+
+  let total = numThreads * perThread
+  gStart.store(false)
+  gCompleted.store(0)
+  gSendErrors.store(0)
+
+  var threads = newSeq[Thread[ptr ProducerArg]](numThreads)
+  var args = newSeq[ProducerArg](numThreads)
+  for i in 0 ..< numThreads:
+    args[i] = ProducerArg(ctx: ctx, count: perThread)
+    createThread(threads[i], producerBody, addr args[i])
+
+  # Times the lock-serialised submit path only; completion (single FFI thread) is excluded.
+  let start = Moment.now()
+  gStart.store(true)
+  joinThreads(threads)
+  let submitSec = (Moment.now() - start).nanoseconds.float / 1_000_000_000.0
+
+  if not waitForCompletions(total):
+    quit("timed out waiting for callbacks: got " & $gCompleted.load() & " of " & $total)
+  os.sleep(50) # let any erroneous extra callbacks land before reading overruns
+
+  IterResult(
+    submitRate: total.float / submitSec,
+    sendErrors: gSendErrors.load(),
+    overruns: max(0, gCompleted.load() - total),
+  )
+
+proc enforceScalingGate(medianRate: seq[float]) =
+  ## Fails the process when submit throughput doesn't scale past RequiredScaling.
+  let scalingMax = medianRate[^1] / medianRate[0]
+  echo ""
+  if scalingMax < RequiredScaling:
+    quit(
+      "SCALING GATE: submit scaling " & formatFloat(scalingMax, ffDecimal, 2) &
+        "x < required " & formatFloat(RequiredScaling, ffDecimal, 2) &
+        "x. The per-request global lock serialises every submit; replace it with " &
+        "MPSC ingress (see tests/bench/README.md) to make this pass."
+    )
+  echo "  scaling gate: ",
+    formatFloat(scalingMax, ffDecimal, 2),
+    "x >= ",
+    formatFloat(RequiredScaling, ffDecimal, 2),
+    "x — submit path scales."
+
+proc main() =
+  let perThread = parseInt(getEnv("FFI_SUBMIT_PER_THREAD", "20000"))
+  let iters = parseInt(getEnv("FFI_SUBMIT_ITERS", "5"))
+  let gateOn = getEnv("FFI_SCALING_GATE", "1") != "0"
+  if perThread < 1 or iters < 1:
+    quit("FFI_SUBMIT_PER_THREAD and FFI_SUBMIT_ITERS must be >= 1")
+  let threadCounts = [1, 2, 4, 8]
+
+  echo "── sendRequestToFFIThread submit throughput (median of ",
+    iters, ") ──────"
+  echo "  ", perThread, " submits per producer thread; noop handler (ok(\"ok\"))"
+  echo ""
+  echo "  ",
+    alignLeft("threads", 9),
+    alignLeft("submits", 10),
+    alignLeft("submit/sec", 16),
+    alignLeft("vs 1-thread", 12)
+
+  var pool: FFIContextPool[BenchLib]
+  var medianRate: seq[float]
+  var allPassed = true
+  for n in threadCounts:
+    var rates: seq[float]
+    var sendErrors = 0
+    var overruns = 0
+    for _ in 0 ..< iters:
+      let r = runOnce(pool, n, perThread)
+      rates.add(r.submitRate)
+      sendErrors += r.sendErrors
+      overruns += r.overruns
+    let med = median(rates)
+    medianRate.add(med)
+    echo "  ",
+      alignLeft($n, 9),
+      alignLeft($(n * perThread), 10),
+      alignLeft(formatFloat(med, ffDecimal, 0), 16),
+      alignLeft(formatFloat(med / medianRate[0], ffDecimal, 2) & "x", 12)
+
+    if sendErrors != 0:
+      echo "  !! ", sendErrors, " submit errors at ", n, " threads"
+      allPassed = false
+    if overruns != 0:
+      echo "  !! ", overruns, " callbacks fired beyond expected at ", n, " threads"
+      allPassed = false
+
+  if not allPassed:
+    quit("stress test FAILED: see !! lines above")
+  echo ""
+  echo "  correctness: callback count matched submits exactly (no drops/dupes)."
+
+  if gateOn:
+    enforceScalingGate(medianRate)
+
+when isMainModule:
+  main()
--- a/tests/unit/test_ffi_context.nim
+++ b/tests/unit/test_ffi_context.nim
@ -489,7 +489,7 @@ suite "Nim-native .ffi. / .ffiCtor. API":

 # Regression for PR #23 review items 1–5: a `.ffi.` body without `await`
 # used to be emitted as an inline-on-foreign-thread fast path, which bypassed
-# `foreignThreadGc`, `ctx.lock`, and chronos's single-thread invariant. The
+# `foreignThreadGc`, the MPSC ingress hand-off, and chronos's single-thread invariant. The
 # sync fast-path was deleted; this test records `getThreadId()` inside a
 # sync body and asserts the handler runs on the FFI thread, not on the
 # caller's thread.
@ -552,10 +552,9 @@ suite "sync-body .ffi. runs on FFI thread (PR #23 regression)":
    check cborDecode(callbackBytes(d), int).value == handlerTid

 # Regression for PR #23 review item 6: reentrancy guard on
-# sendRequestToFFIThread. A handler running on the FFI thread that tries to
-# dispatch back through sendRequestToFFIThread used to self-deadlock waiting
-# on `reqReceivedSignal` (which only the FFI thread can fire). The guard now
-# returns an Err immediately.
+# sendRequestToFFIThread. A handler running on the FFI thread that re-dispatches
+# through sendRequestToFFIThread would enqueue work onto the very queue its
+# blocked dispatcher can never drain. The guard returns an Err immediately.

 var gReentrantNestedRes: Channel[string]
 gReentrantNestedRes.open()