feat(pipeline): park idle pool workers to fix the thread-oversubscription cliff

crates/fgumi-pipeline-core/src/builder.rs

@@ -1222,6 +1222,32 @@ impl Pipeline {
         // with Serial+sticky+Affinity targeting). Indexed by worker id.
         let sticky_owners = assign_sticky_owners(&steps, &owners, n_threads);
 
+        // 1b. Compute which workers are "pinned" — the sole eligible dispatcher
+        // of some step: an Exclusive owner, a sticky owner, or the affinity
+        // target of a `Serial` step (including a non-sticky affinity like the
+        // `correct` chain's pinned `GroupByQueryname`). A pinned worker must NOT
+        // deep-park on the shared event-count: `notify_one` cannot target a
+        // specific worker, so a push meant to wake a pinned step's owner could
+        // wake a peer that `Skip`s it, leaving the owner asleep. Pinned workers
+        // keep the exponential-backoff sleep (they are ≤ a handful of N and were
+        // never the oversubscription problem). Computed while `steps` is still
+        // alive, before `build_worker_storage` consumes it.
+        let pinned_workers: Vec<bool> = {
+            let mut pinned = vec![false; n_threads];
+            for (w, p) in pinned.iter_mut().enumerate() {
+                *p = owners.contains(&Some(w)) || sticky_owners[w].is_some();
+            }
+            for step in &steps {
+                if step.kind() == crate::step::StepKind::Serial
+                    && let Some(target) = step.affinity().target_worker(n_threads)
+                    && target < n_threads
+                {
+                    pinned[target] = true;
+                }
+            }
+            pinned
+        };
+
         // 2. Build per-step contexts (input + output handles). Domain counters
         // are allocated only when telemetry is on (`counters_enabled`), so the
         // telemetry-off path keeps its byte-identical zero-allocation route —
@@ -1309,6 +1335,21 @@ impl Pipeline {
 
         let signal_arc = Arc::clone(&signal);
 
+        // 4-0. Pool event-count parker (Layer 2 oversubscription fix). Only for
+        // the multi-worker path: with one worker there is no peer to notify and
+        // nothing to park behind. Bound to the signal *before* any worker spawns
+        // so a terminal transition (cancel/error) can wake every parked worker.
+        // The signal holds only a `Weak`, so this `Arc` (kept alive here for the
+        // whole run scope) is what keeps the parker live while workers could be
+        // parked.
+        let parker: Option<Arc<crate::runtime::event_count::PoolEventCount>> = if n_threads > 1 {
+            let ec = Arc::new(crate::runtime::event_count::PoolEventCount::new(n_threads));
+            signal_arc.bind_event_count(&ec);
+            Some(ec)
+        } else {
+            None
+        };
+
         // 4a. Optional deadlock-detection monitor. Spawns a watcher
         // thread that periodically samples the stats snapshot; if no
         // step's `progress + finished` counter has advanced for
@@ -1542,6 +1583,7 @@ impl Pipeline {
                 let liveness_clone = Arc::clone(&liveness);
                 // Detached drivers take board slots after the pool workers.
                 let board_clone = worker_board.clone();
+                let parker_clone = parker.clone();
                 let state_slot = n_threads + driver_idx;
                 let thread_name = match group.label() {
                     DetachedGroup::Shared(label) => format!("fgumi-driver-{label}"),
@@ -1568,6 +1610,7 @@ impl Pipeline {
                                     &liveness_clone,
                                     board_clone.as_deref(),
                                     state_slot,
+                                    parker_clone.as_deref(),
                                 );
                             }))
                         {
@@ -1630,6 +1673,9 @@ impl Pipeline {
                     // the zero-cost path.
                     worker_board.as_deref(),
                     0,
+                    // n_threads == 1: no parker (nobody to notify / park behind).
+                    None,
+                    pinned_workers[0],
                 );
             })) {
                 signal_arc.cancel();
@@ -1654,6 +1700,8 @@ impl Pipeline {
                 // Pool worker `worker_id` takes board slot `worker_id` (telemetry
                 // on); `None` on the zero-cost path.
                 let board_clone = worker_board.clone();
+                let parker_clone = parker.clone();
+                let pinned = pinned_workers[worker_id];
 
                 let handle = thread::Builder::new()
                     .name(format!("fgumi-worker-{worker_id}"))
@@ -1683,6 +1731,8 @@ impl Pipeline {
                                     scheduler_clone.as_ref(),
                                     board_clone.as_deref(),
                                     worker_id,
+                                    parker_clone.as_deref(),
+                                    pinned,
                                 );
                             }))
                         {

crates/fgumi-pipeline-core/src/runtime/detached.rs

@@ -337,6 +337,12 @@ pub fn run_detached_driver(
     liveness: &crate::liveness::LivenessCounter,
     board: Option<&crate::runtime::worker_state::WorkerStateBoard>,
     state_slot: usize,
+    // The pool's event-count, so this driver's productive pushes wake parked
+    // pool workers (the notify seam in `dispatch_one_step`). Passed with
+    // `pinned = true` below so the driver itself keeps its `Park` backoff and
+    // never deep-parks on the shared event-count — it is a notifier, not a
+    // waiter (spec §5). `None` when the pool has no parker (`n_threads == 1`).
+    parker: Option<&crate::runtime::event_count::PoolEventCount>,
 ) {
     let primary = group.primary_step();
     let mut row = build_driver_storage(group.into_steps(), contexts.inputs.len());
@@ -352,6 +358,11 @@ pub fn run_detached_driver(
         &DrainFirstScheduler,
         board,
         state_slot,
+        parker,
+        // `pinned`: a detached driver is not a pool waiter. This routes its idle
+        // to the `Park` backoff (its existing behaviour) rather than the shared
+        // event-count wait, while still passing `parker` above for the notify.
+        true,
     );
 }
 
@@ -496,6 +507,7 @@ mod tests {
             &crate::liveness::LivenessCounter::new(1),
             None,
             0,
+            None,
         );
     }