Pool worker fibers and coalesce reactor wake-ups

bench/README.md

@@ -37,33 +37,40 @@ Knobs (both sides accept the same ones): `BENCH_N` (default 100000),
 ## Results (Apple M2, 8 cores, Ruby 4.0.5, 2026-07-07)
 
 100k requests per cell; NSGI at `BENCH_THREADS=8`; allocations are Ruby
-objects allocated per request.
+objects allocated per request. Measured with fiber pooling and coalesced
+reactor wake-ups in place.
 
 | Scenario | NSGI req/s | NSGI allocs | Rack req/s | Rack allocs |
 | -------- | ---------- | ----------- | ---------- | ----------- |
-| 12 headers, no body | 161k | 12.0 | 520k | 22.0 |
-| 60 headers, no body | 163k | 12.0 | 146k | 70.0 |
-| 12 headers, 16 KB body | 165k | 12.0 | 468k | 25.0 |
+| 12 headers, no body | 336k | 10.0 | 520k | 22.0 |
+| 20 headers, no body | ~337k | 10.0 | 377k | 30.0 |
+| 24 headers, no body | ~337k | 10.0 | 300k | 34.0 |
+| 60 headers, no body | 339k | 10.0 | 146k | 70.0 |
+| 12 headers, 16 KB body | 338k | 10.0 | 468k | 25.0 |
 
-Single-threaded NSGI is 60k req/s (16.5 us/req): the cross-thread handoff
+Single-threaded NSGI is 99k req/s (10.1 us/req): the cross-thread handoff
 (pipe write, condvar, context switches) dominates when the reactor sleeps
-between every request. Concurrency amortizes it to ~6 us/req.
+between every request. Concurrency amortizes it to ~3 us/req.
 
 ## Reading the numbers
 
 - **NSGI's costs are flat.** Header count and body size change nothing:
   the host passes pointers, and the app pays only for what it touches
-  (12 allocations per request regardless, about half of which is
-  fiber/reactor machinery).
+  (10 allocations per request regardless).
 - **Rack's costs scale with the request.** Every header becomes a fresh
   String whether or not the app reads it (22 to 70 allocations from 12 to
   60 headers, with throughput dropping 3.6x), and the body is copied into
   `rack.input` (16 KB of memory traffic per request that NSGI simply does
   not perform).
-- **The crossover is real.** At 60 headers, NSGI outruns even this
-  handoff-free idealized Rack. Against a real Rack server (which adds
-  socket handling, parsing, and its own thread handoffs on top), the
-  crossover comes much earlier.
-- **The fixed handoff is NSGI's price.** ~4-6 us/request buys VM isolation
-  from foreign threads. Batching more jobs per reactor wake-up and pooling
-  fibers are the known levers if this ever dominates a real workload.
+- **The crossover sits inside typical traffic.** NSGI overtakes this
+  handoff-free idealized Rack at ~22 request headers; ordinary browser
+  requests carry 15 to 25. Against a real Rack server (which adds socket
+  handling, parsing, and its own thread handoffs on top), the crossover
+  comes earlier still.
+- **The fixed handoff is NSGI's price.** ~2-3 us/request buys VM isolation
+  from foreign threads. Two optimizations already landed here: worker
+  fibers are pooled across requests instead of created per request, and
+  wake-ups are coalesced (only the producer that flips the wake flag
+  writes to the pipe; the reactor resets it when it drains the queue
+  empty). Remaining levers if the handoff ever dominates a real workload:
+  batching response signaling, or sharding across VMs.

ext/nsgi/src/lib.rs

@@ -122,6 +122,13 @@ unsafe impl Sync for Job {}
 struct Runtime {
     queue: Mutex<VecDeque<Arc<Job>>>,
     wake_write: c_int,
+    /// Wake coalescing: true while a wake byte is in flight (written but not
+    /// yet followed by the reactor draining the queue empty). Producers only
+    /// write when they flip this false -> true; the reactor resets it when a
+    /// pop finds the queue empty. Both transitions happen under the queue
+    /// mutex, so a producer that skips the write is guaranteed its job is
+    /// seen by the drain loop the pending wake triggers.
+    wake_signaled: std::sync::atomic::AtomicBool,
     /// Maximum number of queued (not yet dispatched) jobs before new calls
     /// are rejected with a 503; 0 means unbounded. From `NSGI_MAX_PENDING`.
     max_pending: usize,
@@ -169,6 +176,7 @@ fn boot() -> Result<Runtime, ()> {
         Ok(Ok(())) => Ok(Runtime {
             queue: Mutex::new(VecDeque::new()),
             wake_write,
+            wake_signaled: std::sync::atomic::AtomicBool::new(false),
             max_pending,
         }),
         Ok(Err(message)) => {
@@ -335,9 +343,16 @@ unsafe extern "C" fn native_pop(_module: VALUE) -> VALUE {
     let Some(runtime) = runtime() else {
         return Qnil as VALUE;
     };
-    match runtime.queue.lock().unwrap().pop_front() {
+    let mut queue = runtime.queue.lock().unwrap();
+    match queue.pop_front() {
         Some(job) => unsafe { rb_ull2inum(Arc::into_raw(job) as usize as u64) },
-        None => Qnil as VALUE,
+        None => {
+            // Queue drained: let the next producer write a fresh wake byte.
+            runtime
+                .wake_signaled
+                .store(false, std::sync::atomic::Ordering::Release);
+            Qnil as VALUE
+        }
     }
 }
 
@@ -640,16 +655,22 @@ fn handle_impl(req: *const NsgiRequest) -> NsgiResponse {
         cv: Condvar::new(),
     });
 
+    let need_wake;
     {
         let mut queue = runtime.queue.lock().unwrap();
         if runtime.max_pending > 0 && queue.len() >= runtime.max_pending {
             return static_503();
         }
         queue.push_back(Arc::clone(&job));
+        need_wake = !runtime
+            .wake_signaled
+            .swap(true, std::sync::atomic::Ordering::AcqRel);
+    }
+    // Wake the reactor, coalesced: only the producer that flipped the flag
+    // writes; everyone else's job rides the wake already in flight.
+    if need_wake {
+        unsafe { libc::write(runtime.wake_write, b"!".as_ptr() as *const c_void, 1) };
     }
-    // Wake the reactor; one byte per job. EINTR aside, a full pipe just
-    // means plenty of wake-ups are already pending.
-    unsafe { libc::write(runtime.wake_write, b"!".as_ptr() as *const c_void, 1) };
 
     let mut done = job.done.lock().unwrap();
     while done.is_none() {

lib/nsgi.rb

@@ -40,23 +40,43 @@ def load_app(path)
     end
 
     # The reactor: invoked by the native bridge on the VM thread once the
-    # application is loaded; never returns. Each queued job becomes its own
+    # application is loaded; never returns. Each queued job runs in its own
     # Fiber, so requests interleave whenever application code waits (sleep,
-    # IO, ...) instead of blocking the whole VM.
+    # IO, ...) instead of blocking the whole VM. Worker fibers are pooled:
+    # after finishing a request they park themselves and are resumed with
+    # the next job instead of being created (and GC'd) per request.
     def __run(wake_fd)
       scheduler = @scheduler || Scheduler.new
       Fiber.set_scheduler(scheduler)
       wake = IO.for_fd(wake_fd, autoclose: false)
+      pool = []
       loop do
         while (job = Native.__pop)
-          Fiber.schedule { Native.__process(job) }
+          worker = pool.pop
+          worker = pool.pop while worker && !worker.alive?
+          worker ||= __worker(pool)
+          worker.resume(job)
         end
         scheduler.__step(wake)
       rescue Exception => e
         warn(e.full_message) if ENV["NSGI_DEBUG"]
       end
     end
 
+    # A reusable request worker. Runs one job per resume; between jobs it
+    # parks itself back into the pool and yields to the reactor. Pushing
+    # before yielding is safe: the VM is single-threaded, so the reactor
+    # can only pop this fiber after the yield actually suspends it.
+    def __worker(pool)
+      Fiber.new do |job|
+        loop do
+          Native.__process(job)
+          pool << Fiber.current
+          job = Fiber.yield
+        end
+      end
+    end
+
     # Entry point invoked by the native bridge, once per request, inside that
     # request's Fiber. Builds the lazy Request façade over the native handle
     # and returns a normalized `[Integer, Array<[String, String]>, String or