diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 20f2310..7fcc1b4 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -22,6 +22,9 @@ The public API stays stable while scheduler selection, queue topology, and instr - `ThreadPool.statsSnapshot() Stats` - `ThreadPool.purgeWorkerCaches() !void` +`ThreadPool.purgeWorkerCaches()` is quiescent-only. It returns +`error.PoolBusy` unless `active_tasks == 0 && queued_tasks == 0`. + ## Scheduler Variants ### `global_queue` @@ -34,9 +37,14 @@ The public API stays stable while scheduler selection, queue topology, and instr ### `work_stealing` - One bounded central inject queue for external submitters. -- One bounded local deque per worker. -- Local submissions from worker threads prefer the local deque. +- One owner-local LIFO slot plus one bounded local queue per worker. +- Local submissions from worker threads prefer the LIFO slot, then the local + queue. - Idle workers refill from the inject queue, then attempt steals from peer deques. +- Worker wake selection is configurable: + - `scan_parked_workers` keeps the legacy submit-path parked-worker scan. + - `idle_queue` uses a bounded idle-worker ticket queue to pick wake + candidates without scanning every worker. - Included as the low-contention candidate and still benchmarked on every baseline run. ### `lane_extension` @@ -49,24 +57,37 @@ The public API stays stable while scheduler selection, queue topology, and instr ### Inject Queue - Bounded ring buffer. -- Mutex-protected, no unbounded growth. +- Vyukov-style MPMC sequence queue, no unbounded growth. - Backpressure is explicit: `QueueFull`, or blocking retry with timeout. ### Local Worker Queue -- Bounded deque. -- Owner pushes and pops from the bottom. -- Steals happen from the top. -- The local deque uses a lock-free Chase-Lev design. The owner thread pushes and pops wait-free (no CAS on the fast path). Stealers use a CAS on the top index. Buffer size is constrained to power-of-2 for bitmask wrapping. +- Bounded owner-local queue with pluggable implementation via `Config.queue_kind`. +- `queue_kind = .ring` is the default Tokio/Go-style SPMC ring. +- `queue_kind = .chase_lev` remains available as the classic Chase-Lev deque. +- Owner pushes and pops from the bottom / tail side; steals happen from the top / head side. ## Memory Model - Task descriptors are preallocated from a bounded descriptor array. - Descriptor reuse happens only after task completion and, for waitable tasks, after `wait`. - Worker scratch allocates one bounded buffer per worker lazily on first use. +- On the normal owning-worker path, scratch begin/alloc/finish are lock-free. - Scratch buffers can be purged explicitly or after configured idle periods. +- Explicit purge is only legal while the pool is quiescent. - All pool-owned allocations are counted in `total_alloc_bytes`, `total_free_bytes`, `live_bytes`, and `peak_live_bytes`. +## Runtime Presets + +- `RuntimePreset.standard` preserves the stable defaults. +- `RuntimePreset.tokio_parity` is an explicit benchmark-focused preset that + currently enables: + - `timestamp_mode = .disabled` + - `wake_strategy = .idle_queue` + +The preset exists as an opt-in configuration surface. It is not silently +forced by the benchmark harness. + ## Task Lifecycle 1. Submission validates pool state and acquires a descriptor from the bounded freelist. diff --git a/docs/HANDOFF-2026-04-14.md b/docs/HANDOFF-2026-04-14.md index c1e355b..cd9916c 100644 --- a/docs/HANDOFF-2026-04-14.md +++ b/docs/HANDOFF-2026-04-14.md @@ -2,13 +2,64 @@ > **For the next agent:** this is a standalone handoff. Read it first before touching any perf code. It captures what shipped, what failed, what's preserved as audit trails, current host state, and what NOT to try next. +## Late-session update — current dirty workspace + +This file originally captured the docs-only / Option 3 close-out state on +`main`. The current workspace now has uncommitted scheduler changes on top of +that baseline. They are documented here and in +`docs/performance/2026-04-14-scratch-fastpath-idle-queue/README.md`. + +- **Workspace status:** dirty tree, not on `main`. Modified files: + `src/memory/scratch.zig`, `src/pool.zig`, `src/root.zig`, + `src/types.zig`, `tests/unit.zig`, `tests/stress.zig`. +- **Reference review:** `docs/codex-review-2026-04-14.md` drove this attempt. +- **What changed in the workspace:** removed the per-task scratch mutex from + the worker fast path; strengthened `purgeWorkerCaches()` to reject unless + `active_tasks == 0 && queued_tasks == 0`; added an opt-in idle-worker queue + wake strategy; added `runtime_preset = .tokio_parity` which currently maps + to `timestamp_mode = .disabled` and `wake_strategy = .idle_queue`. +- **Correctness note:** the scratch fast-path refactor initially exposed a + double-purge race between idle-time purge and external quiescent purge. That + showed up as an intermittent `SIGSEGV` in + `stress.test.repeated churn returns to a bounded steady state after purge`. + The final workspace fixes this by serializing only overlapping purge callers + on a slow-path mutex while keeping `beginTask()`, `alloc()`, and + `finishTask()` lock-free on the owning-worker path. +- **Validation on the final workspace:** `zig build test --summary all`, + `zig build test-integration --summary all`, and + `zig build test-stress --summary all` all pass; a focused 30x loop on the + purge/churn stress passes; the required 20x `ReleaseFast` stress gate also + passes. +- **Performance verdict on the final workspace:** same-session balanced-order + A/B (n=10 each) versus a clean baseline worktree at `HEAD` is still + negative: + - `tokio_parity/flood_1m`: 487,840.5 -> 465,650.0 ops/s median (**-4.5%**) + - `tokio_parity/contended_spawn_8x`: 469,570.0 -> 411,959.0 ops/s median + (**-12.3%**) +- **Host-load caveat:** these numbers were gathered while the host was under + substantial unrelated load (`load average: 12.89, 13.63, 11.13` with several + `geode` processes consuming CPU). That must be acknowledged for benchmark + variance, but it is **not** an excuse to ignore the fixed correctness bug or + to claim a perf win that the same-session A/B does not show. + ## TL;DR (60 seconds) +- **Current workspace is dirty with uncommitted perf code.** `main` still + points at the docs-only Option 3 close-out (`456c865`), but the working tree + now contains the scratch fast-path + idle-queue attempt summarized above. - **Project goal:** Zig thread pool pursuing Tokio `rt_multi_threaded` performance parity on `flood_1m` benchmark. Current gap: ~12-38× from Tokio (5-15M target vs ~390K steady-state on current host). - **Main tip (2026-04-14):** `456c865` — docs-only commit about an abandoned Option 3 attempt. No perf-affecting code changes on main this session. - **Single-atomic optimization surface is exhausted** for flood_1m on this architecture. Three attempts confirmed: G3 (null/+1.7%), wake-gate (−11/−2.5/−8.9%), Option 3 (**−48%**). Structural change required for further flood_1m wins. -- **Host is drifted.** Morning baseline of 566K ops/s does not reproduce. Current same-session reference: ~400K ops/s. Cause: 4+ days uptime + long-lived user processes; not fixable without reboot. Geode co-tenant was stopped this session; moved perf +5% (within noise). -- **Open decisions for user:** sentinel revert (pending), next-direction (multi-session options only remain), whether to reboot for a clean 566K baseline attempt. +- **The latest workspace attempt is correctness-clean but not perf-positive.** + Scratch fast path + idle queue + parity preset plumbing are implemented and + stress-tested, but the same-session A/B is still negative on both required + Tokio-parity scenarios. +- **Host is drifted and currently loaded.** Morning baseline of 566K ops/s + does not reproduce. Current same-session reference is host-load-sensitive, + and the final workspace measurements were taken under 12-13 load average. +- **Open decisions for user:** sentinel revert (still pending), whether to + keep iterating on the idle-queue path versus slicing back to the scratch-only + win, and whether to reboot for a clean-host remeasure. ## Project Overview @@ -19,6 +70,15 @@ - `/home/benji/src/devnw/zig/thread-pool-phase-g2` — older worktree at `eae94ce [feat/phase-g2-queued-tasks-poll]` (inherited from prior session, not active). - `/home/benji/src/devnw/zig/thread-pool-option3` — this session's Option 3 implementation at `371d8ce [feat/option3-worker-local-incoming]` (NOT merged, preserved as audit trail). +**Current workspace state:** + +- Dirty working tree with the scratch fast-path + idle-queue attempt. +- The benchmark harness in `benchmarks/bench_tokio_parity.zig` was restored to + its stable config; the new parity behavior is available through + `Config.runtime_preset = .tokio_parity` rather than forced by the bench. +- Detailed evidence archive: + `docs/performance/2026-04-14-scratch-fastpath-idle-queue/README.md`. + **Primary target benchmark:** `benchmarks/bench_tokio_parity.zig` — specifically the `flood_1m` scenario (single submitter, 1M tasks). Built via `zig build -Doptimize=ReleaseFast bench-tokio-parity`. The resulting binary ends up at `.zig-cache/o//thread-pool-tokio-parity` (the `bench-tokio-parity` target auto-runs the bench; use direct binary invocation for n-count control). **Secondary benchmarks in same binary:** `contended_spawn_8x` (8 concurrent submitters). @@ -201,6 +261,10 @@ Brainstorms at `docs/superpowers/brainstorms/`: 1. **Sentinel revert.** `claim_sentinel_enabled = true` at `src/pool.zig:167` adds ~2 atomic ops per dispatch in ReleaseFast. Phase H fix is shipped; the TODO at line 163 says revert to `std.debug.runtime_safety`. Option 3 didn't ship, so the "keep through Option 3, revert after" plan is moot. Can revert independently now. 2. **Next direction.** Remaining options from the wrap-up's ranking: + - **Current dirty-tree question:** keep iterating on the idle-queue path, + or split out the scratch fast-path / quiescent-purge changes as a smaller + correctness-and-overhead slice. The combined attempt is tested but not + benchmark-positive on this host. - **Option 2: park/wake contract redesign** — multi-session, high-risk. Replace condvar-based parking with futex-direct waiter lists so `queued_tasks` SC-fetchAdd can relax to `.monotonic`. Every submit saves one `mfence` on x86. Substantial redesign; must revalidate Phase H ghost-dispatch defense against new wake model. - **Hybrid Option 3:** contention-aware routing — use incoming only when `incoming_cursor` fetchAdd shows multi-submitter contention; otherwise direct to inject. Would preserve flood_1m's hot-line advantage while capturing the measured +18% contended_spawn_8x spread. Conjectural, no plan. - **Async-context-aware submit:** Tokio-style — track which async task the submitter runs in, enable true locality. Fundamentally different distribution model. Multi-session. diff --git a/docs/PERFORMANCE_TRACKING.md b/docs/PERFORMANCE_TRACKING.md index 263b4bf..9a80327 100644 --- a/docs/PERFORMANCE_TRACKING.md +++ b/docs/PERFORMANCE_TRACKING.md @@ -4,6 +4,61 @@ Rolling record of micro-benchmark numbers as the Tokio-parity plan progresses. All numbers from local dev machine unless noted. CI baselines tracked in `benchmarks/results/baseline.json`. +## 2026-04-14 late session — scratch fast path + idle queue attempt + +**Status:** implemented in the current working tree only, not on `main`. +Detailed archive: `docs/performance/2026-04-14-scratch-fastpath-idle-queue/`. + +### What changed + +- `WorkerScratch.beginTask()`, `alloc()`, and `finishTask()` are now lock-free + on the normal owning-worker path. +- `ThreadPool.purgeWorkerCaches()` / `Core.purgeWorkerCaches()` now reject + unless the pool is quiescent: `active_tasks == 0 && queued_tasks == 0`. +- Added `WakeStrategy` with an opt-in `idle_queue` mode that replaces the + submit-path O(N) parked-worker scan with a bounded idle-worker ticket queue. +- Added `RuntimePreset.tokio_parity`, which currently enables + `timestamp_mode = .disabled` and `wake_strategy = .idle_queue`. + +### Correctness outcome + +The first cut of the scratch fast-path refactor exposed a real race: +idle-time purge and external quiescent purge could overlap and double-free a +scratch buffer. That caused an intermittent `SIGSEGV` in +`stress.test.repeated churn returns to a bounded steady state after purge`. + +Final workspace fix: + +- keep the task fast path lock-free +- serialize only overlapping purge callers on a slow-path purge mutex + +Validation on the final workspace: + +- `zig build test --summary all` PASS +- `zig build test-integration --summary all` PASS +- `zig build test-stress --summary all` PASS +- focused purge/churn stress loop: 30/30 PASS +- required `ReleaseFast` stress gate: 20/20 PASS + +### Same-session A/B (balanced order, n=10 each, ReleaseFast) + +Baseline was a clean worktree at `HEAD`; candidate was the final dirty +workspace. Host load during this measurement was high +(`load average: 12.89, 13.63, 11.13` with unrelated `geode` processes +consuming CPU), so treat these as relative same-session numbers only. + +| Scenario | Baseline median ops/s | Baseline p95 | Candidate median ops/s | Candidate p95 | Delta | +|---|---:|---:|---:|---:|---:| +| `tokio_parity/flood_1m` | 487,840.5 | 537,976 | 465,650.0 | 541,624 | -4.5% median | +| `tokio_parity/contended_spawn_8x` | 469,570.0 | 510,690 | 411,959.0 | 453,933 | -12.3% median | + +### Verdict + +This attempt improved the implementation shape and closed a real purge race, +but it did **not** meet the performance success criteria on this host. Do not +claim Tokio/Go parity progress from this patch as-is, and do not flip the +default wake strategy based on these numbers. + ## Verdict vs Targets | Target | v0.2.0 baseline | After plan | Met? | diff --git a/docs/README.md b/docs/README.md index dd14feb..c9e4eca 100644 --- a/docs/README.md +++ b/docs/README.md @@ -18,3 +18,7 @@ A Zig library for thread-pool - [Contributing](../CONTRIBUTING.md) - [Security](../SECURITY.md) - [Changelog](../CHANGELOG.md) +- [Current Handoff](HANDOFF-2026-04-14.md) +- [Performance Tracking](PERFORMANCE_TRACKING.md) +- [Design](DESIGN.md) +- [2026-04-14 Codex Review](codex-review-2026-04-14.md) diff --git a/docs/codex-review-2026-04-14.md b/docs/codex-review-2026-04-14.md new file mode 100644 index 0000000..7614225 --- /dev/null +++ b/docs/codex-review-2026-04-14.md @@ -0,0 +1,274 @@ +# Performance Review of zig/thread-pool Toward Tokio and Go Parity + +## Scope and current repo state + +The `thread-pool` project is a standalone Zig thread-pool library for the `devnw/zig` ecosystem, offering bounded task descriptors, fixed worker caps, a global-queue baseline, and a work-stealing scheduler behind a single API. citeturn13view0 The stated goal for the current performance initiative is parity with Tokio’s multithreaded runtime on the `tokio_parity/flood_1m` benchmark (1M trivial tasks, single submitter) and improved multi-submitter performance on `contended_spawn_8x`. citeturn9view0turn13view1 + +The repository is hosted on entity["company","GitLab","code hosting platform"] and the most recent `main`-branch work (as of April 14, 2026) is documentation-heavy, consolidating prior merges (notably Phase G2 re-land and Phase H race fix) and preserving abandoned performance experiments as audit trails. citeturn3view0turn9view0 The session handoff explicitly warns that mainline performance did not change meaningfully “this session,” and that the current host is drifted (absolute numbers should not be compared across sessions without controlling host state). citeturn9view0turn30view2 + +Two recent, relevant negative-result branches are especially important in understanding today’s “blockers”: + +- **Option 3 worker-local incoming queues**: decisive `flood_1m` regression (≈ −48% median), while improving `contended_spawn_8x` (≈ +18% median). citeturn30view0turn9view0 +- **Wake-scan parked-count gate**: three attempts to short-circuit the O(N) parked-worker scan regressed `flood_1m` (−11%, −2.5%, −8.9%), with root cause attributed to cacheline contention and atomic RMW traffic. citeturn30view1turn20view0 + +The practical implication of those results (and earlier profiling history tracked in `docs/PERFORMANCE_TRACKING.md`) is that micro-optimizations around “single atomics” are no longer sufficient to close the remaining Tokio/Go gap; structural scheduler/wake-model adjustments are required. citeturn25view3turn30view2 + +## Architecture review + +### Benchmark that defines “Tokio parity” in this repo + +The repository’s Tokio-parity benchmark binary implements two scenarios: + +- `flood_1m`: single submitter loop submits 1,000,000 trivial tasks using `.backpressure = .block` (bounded submission), then waits for completion. citeturn13view1 +- `contended_spawn_8x`: 8 OS threads submit concurrently (100,000 total tasks), then wait. citeturn13view1 + +The benchmark config uses: + +- `max_threads = 0` → “CPU count” (capped internally at 64). citeturn13view1turn26view0 +- `queue_capacity = 4096` and `local_queue_capacity = 256`. citeturn13view1turn26view0 +- work-stealing scheduler. citeturn13view1turn25view2 + +This matters because it entrenches two expensive invariants: + +- **bounded descriptors** (descriptor array sized to `queue_capacity`), so flood workloads necessarily cycle descriptor acquire/release and can trigger blocking when “in-flight” saturates. citeturn18view0turn13view1 +- **OS-thread parking/unparking pressure**: if submit throughput is not high enough to keep workers busy, they will park frequently and require wakes, driving futex/condvar overhead into the critical path. citeturn18view0turn25view3 + +### Current scheduler mechanics in `src/pool.zig` + +The hot path for `submit()` is: + +1. Acquire a descriptor from a bounded freelist (`free_top_tagged` ABA-tagged CAS stack). citeturn18view0 +2. Fill descriptor fields; mark state `.queued`. citeturn18view0 +3. Enqueue to: + - worker-local LIFO slot or local queue when submitting from a worker thread, otherwise + - global inject queue (`BoundedRing` Vyukov-style MPMC). citeturn18view0turn9view0 +4. Update counters: `submitted.fetchAdd(.monotonic)` plus `queued_tasks.fetchAdd(.seq_cst)` (for the park/wake handshake), tracking peak queue depth. citeturn18view0turn25view3 +5. Wake one parked worker by scanning `is_parked` across the entire worker array (round-robin start), using a `seq_cst` load per worker until a parked worker is found, then acquiring that worker’s park mutex and signaling its condition variable. citeturn18view0turn30view1 + +When a worker goes idle (can’t find tasks in LIFO, local queue, inject queue, or via stealing), the park path does: + +- optional spin loop (`spin_before_park_iters`, default 0). citeturn18view0turn26view0 +- lock `park_mutex`, then execute a **seq_cst total-order handshake**: `is_parked.store(.seq_cst)` and `queued_tasks.load(.seq_cst)` to avoid a lost-wakeup scenario, then `timedWait` on `park_cond` for `idle_park_after_ns` (default 250µs). citeturn18view0turn25view3 + +On the worker side, `runTask` uses an atomic state transition `.queued → .running` (Phase H fix) and decrements `queued_tasks` upon CAS success (to prevent shutdown deadlocks in the presence of “ghost duplicates”). citeturn18view0turn9view0 + +### Observability and “extras” on the hot path + +Two library features materially affect per-task overhead: + +- **timestamp mode**: `full` records timestamps and updates latency histograms; `disabled` skips most timestamp reads and histogram updates. citeturn13view2turn18view0 +- **per-worker scratch buffer**: each task calls `WorkerScratch.beginTask()` and `finishTask()` around task execution. Today, `beginTask()` takes a mutex lock every task, even when tasks never allocate scratch memory. citeturn27view0turn18view0 + +The scratch lock is not an academic concern: it inserts a synchronization primitive into the inner loop of worker execution (and into the “noop task” parity benchmarks), which Tokio and Go avoid by relying more heavily on thread-local or per-worker data with minimal locking. citeturn27view0turn21view1turn29search1 + +## Findings: what is currently blocking throughput + +### The “Tokio/Go gap” is mostly structural, not a missing micro-optimization + +The handoff and re-profile documents consistently converge on the same topline: even optimistic “single-atomic ceiling” improvements do not get close to the 5–15M ops/s Tokio target; they move the needle into the ~0.8–1.1M range at best, leaving most of the gap intact. citeturn30view2turn9view0 + +This is reinforced by the project’s own historical phase tracking, where multiple “hot path” ideas either regressed or proved to be off the critical path once profiled (e.g., lock-free backpressure rewrite showed negligible benefit; other phases were reverted after measurement). citeturn25view3turn30view1 + +The blockers are therefore not “one missing trick,” but a cluster of compounding costs that show up strongly on the `flood_1m` workload shape. + +### Primary blocker: wake strategy forces expensive per-submit work under producer-bound load + +The current `enqueueTask` wake behavior is expensive in two ways: + +- It performs an O(N) scan of `is_parked` across `workers.len` *on every successfully enqueued task*, using `seq_cst` loads. citeturn18view0turn30view1 +- When it finds a parked worker, it takes a mutex and signals a condition variable—an operation that can drive futex wake overhead into the dominant cost when workers are parking frequently. citeturn18view0turn25view3 + +The project already verified that wake-path changes have meaningful ceilings (wake-scan stubs and earlier “remove wake scan” experiments), but attempts to “gate” them with a shared counter regressed because introducing shared atomic RMW traffic during frequent park/unpark cycles cost more than the scan savings. citeturn30view1turn20view0 + +The key insight is: **the wake bottleneck is not just the scan; it’s the interaction between frequent parking and being forced to do substantial work per submit to recover from that parking.** citeturn25view3turn18view0 + +### Secondary blocker: aggressive parking + seq_cst handshake keeps the system in a “wake heavy” regime + +The `idle()` path parks a worker as soon as `nextTask()` returns null, with default `spin_before_park_iters = 0`. citeturn18view0turn26view0 In producer-bound workloads where the submitter can’t keep the queue saturated (which is exactly what the re-profile describes: submitter 100% running, workers mostly blocked), this makes workers churn between runnable and parked states. citeturn30view2turn18view0 + +The correctness of the lost-wakeup guard depends on a seq_cst total order between `queued_tasks` and `is_parked`. That enforces stronger ordering than Tokio or Go typically use in their local fast paths, and it constrains how “cheap” a wake optimization can be without reworking this handshake. citeturn18view0turn21view1turn22view1 + +### Tertiary blocker: per-task scratch locking on the worker fast path + +Even for a no-op task, every execution path calls: + +- `worker.scratch.beginTask()` (mutex lock + “ensure buffer” logic), then +- `worker.scratch.finishTask()`. citeturn18view0turn27view0 + +This injects a lock into the worker inner loop, which is not free even when uncontended. It becomes especially hard to justify in the parity benchmarks, where scratch is not used by the workload at all. citeturn13view1turn27view0 + +### Confirmed dead-ends that should not be repeated + +Two confirmed blockers are worth stating plainly because they constrain “what to fix next”: + +- **Option 3 (worker-local incoming queues)** cannot be shipped as implemented if `flood_1m` remains a top KPI. It dramatically worsens single-submitter throughput because it forces a submitter to rotate across many cold cachelines; the same design is plausibly beneficial for multi-producer contention. citeturn30view0turn9view0 +- **A single global “parked_count” gate** is net-negative under the observed park/unpark dynamics; any wake optimization must avoid a shared RMW hotspot written by every worker. citeturn30view1turn20view0 + +## How Tokio and Go avoid these costs + +### Tokio’s multithreaded runtime does not scan all workers to find someone to wake + +Tokio’s documentation for the multithreaded runtime describes essentially the same queue topology (global + per-worker local queues, work-stealing, and a LIFO slot optimization). citeturn21view1turn25view3 The crucial difference is that Tokio explicitly coordinates idle workers so the runtime can quickly choose a worker to notify. + +In Tokio’s scheduler implementation, worker wake-up uses an idle coordinator: `notify_parked_local` calls `idle.worker_to_notify(...)` and then directly un-parks the chosen worker via its `unpark` handle—no linear scan of all workers’ parked flags. citeturn23view0turn28search0 This is a fundamentally different wake-selection strategy than the Zig pool’s “scan `is_parked[]` every submit” approach. citeturn18view0turn30view1 + +Tokio’s parker itself is also a small state machine around an atomic state plus a condvar/driver mechanism, with strong emphasis on not losing notifications. citeturn22view1 The Zig pool has the same “don’t lose wakeups” constraint, but currently pays for it with a global seq_cst counter and cross-thread scan. citeturn18view0turn25view3 + +### Go’s runtime similarly relies on local run queues and work-stealing, but avoids per-task OS wakeups + +Your `RingQueue` is explicitly inspired by Tokio and Go’s run queue logic: single producer pushes/pops from a tail, stealers CAS the head and steal in batches (“steal half”), mirroring patterns in `runtime/proc.go`. citeturn27view1turn29search1 + +However, Go’s scheduler is not built around waking OS threads per unit of work. Goroutines are scheduled within the runtime on existing OS threads bound to “P” tokens, and the system is designed to keep runnable work available to running threads through local queues, global queue balancing, and work stealing. citeturn29search1turn27view1 The overall effect is the same: **wake selection and work availability are designed so the system does not require a heavy OS-thread park/unpark loop for trivial work.** + +## Fixes that unblock measurable gains + +This section is deliberately blunt: the observed gap is too large to claim “Tokio/Go parity” from a handful of patches. The fixes below are the changes that remove the biggest structural brakes *so the project can actually pursue* the next order-of-magnitude improvements without repeatedly hitting the same ceiling. They are also chosen to respect the repository’s demonstrated lessons: avoid shared RMW hotspots, measure with paired A/B, and keep correctness gates (Phase H sentinel, stress tests) in place during scheduler work. citeturn9view0turn30view1 + +### Remove per-task scratch locking from the worker hot path + +Right now, the parity benchmark’s noop tasks pay a scratch mutex lock/unlock for every task because `runTask` unconditionally calls `beginTask()`, and `beginTask()` always takes a mutex. citeturn18view0turn27view0 + +You can remove that lock from the hot path safely by strengthening the `purgeWorkerCaches` precondition so that `purge()` is only called when **both** `active_tasks == 0` and `queued_tasks == 0`, i.e., there is not only no running work but also no pending work that could start immediately after the check. (Today it checks only `active_tasks`.) citeturn18view0turn27view0 + +Once purge is constrained this way, scratch state is single-thread-owned during normal execution, and the mutex can be removed (or moved behind a slow-path only). + +A concrete patch sketch (illustrative; adjust formatting to match repository style): + +```diff +diff --git a/src/pool.zig b/src/pool.zig +@@ + pub fn purgeWorkerCaches(self: *Core) !void { +- if (self.counters.active_tasks.load(.acquire) != 0) return error.PoolBusy; ++ if (self.counters.active_tasks.load(.acquire) != 0) return error.PoolBusy; ++ if (self.counters.queued_tasks.load(.acquire) != 0) return error.PoolBusy; + for (self.workers) |*worker| { + worker.scratch.purge(); + } + } + +diff --git a/src/memory/scratch.zig b/src/memory/scratch.zig +@@ + pub const WorkerScratch = struct { +@@ +- /// Serializes purge() and beginTask()/alloc() so that the owning worker +- /// and a remote caller (e.g. Core.purgeWorkerCaches) cannot race on the +- /// backing buffer. Without this guard, two concurrent purge() calls would +- /// double-free the allocation. +- mutex: std.Thread.Mutex = .{}, ++ // Invariant: purge() is only called when the pool is quiescent: ++ // active_tasks == 0 and queued_tasks == 0 (see Core.purgeWorkerCaches). +@@ + pub fn beginTask(self: *WorkerScratch) !void { +- self.mutex.lock(); +- defer self.mutex.unlock(); + try self.ensureBufferLocked(); + if (self.buffer) |buf| { + self.fba = std.heap.FixedBufferAllocator.init(buf); + self.current_bytes = 0; + } + } +@@ + pub fn alloc(self: *WorkerScratch, comptime T: type, count: usize) ![]T { +- self.mutex.lock(); +- defer self.mutex.unlock(); + try self.ensureBufferLocked(); + if (self.fba == null) return error.ScratchUnavailable; +@@ + pub fn purge(self: *WorkerScratch) void { +- self.mutex.lock(); +- defer self.mutex.unlock(); + if (self.buffer) |buf| { + self.accounting.recordFree(buf.len); + self.allocator.free(buf); + } + self.buffer = null; + self.fba = null; + self.current_bytes = 0; + } +``` + +This removes a mutex from the per-task inner loop and aligns scratch behavior with how the benchmarks actually use it. citeturn27view0turn13view1 + +### Replace O(N) wake scanning with an idle-worker queue that avoids a shared RMW hotspot + +The wake-scan parked-count gate failed because it added a shared counter written by every worker on every park/unpark transition. citeturn30view1turn20view0 The right replacement pattern is closer to Tokio: track idle workers so the submitter can wake one without scanning the whole pool. citeturn23view0turn28search0 + +A practical approach that respects the “avoid a shared RMW hotspot” lesson is: + +- Maintain an idle set as **multiple shards**, each shard a bounded MPMC ring of worker indices (or a Treiber stack) to reduce contention. +- Workers enqueue themselves into their shard *only when actually transitioning into the parked/waiting state*, and only if not already enqueued. +- Submitters dequeue from shards to get a candidate parked worker to signal, avoiding O(N) scanning and reducing the frequency with which the submit path touches per-worker cachelines. + +This is not “free”—it changes the park/unpark transition cost—but it targets the most damaging property of the current design: doing O(N) work on every submit. citeturn18view0turn25view3 + +A minimal implementation strategy (one shard to start, then shard if needed) looks like this: + +- Add to `Worker`: + - `idle_enqueued: atomic u8` (0/1) to prevent duplicate inserts and bound queue growth even when a worker wakes via timeout. +- Add to `Core`: + - `idle_queue: BoundedRing(u32)` sized to `max_threads` (or `2*max_threads`) and normalized to a power of two. + +Then modify: + +- `idle()` to: + - set `is_parked = 1` + - enqueue its index into `idle_queue` before the `queued_tasks` check + - proceed to `timedWait` + +- `enqueueTask` to: + - after successful enqueue + counter increments, try `idle_queue.pop()` and signal that worker if it is still parked, then clear `idle_enqueued`. + +This preserves the existing correctness model (Phase G2/Phase H semantics) while removing the linear scan from the submit path. citeturn18view0turn9view0turn30view1 + +The project already has strong evidence that wake-path improvements have a large ceiling; this is the kind of structural change that can plausibly capture more of that ceiling than a global counter gate can, because it reduces *read amplification* on the submitter rather than adding a new shared write hotspot. citeturn30view2turn30view1 + +### Stop paying `seq_cst` on a global counter forever by moving the lost-wakeup guarantee into a per-worker parker + +Even if you remove the scan, `queued_tasks.fetchAdd(.seq_cst)` is still on the submit hot path specifically to participate in the SC handshake that prevents the IRIW lost-wakeup scenario. citeturn18view0turn25view3 Tokio avoids building lost-wakeup prevention on a global queue-depth counter; it uses a per-parker atomic state machine that is updated on park/unpark paths. citeturn22view1turn23view0 + +If Tokio/Go parity is truly the goal (not “best effort within today’s architecture”), the long-term fix is to: + +- introduce a per-worker `park_state` (e.g., EMPTY / PARKED / NOTIFIED) +- make `unpark()` a state transition (often a single atomic swap) plus a conditional signal +- remove the need for `queued_tasks` to be seq_cst purely for wake correctness + +This is a larger change than the two patches above, but it’s the lever that removes a global ordering primitive from the hottest submit path. Without doing this, you will keep running into ceilings where the submitter is “100% running” and largely performing synchronization and wake maintenance. citeturn30view2turn18view0turn22view1 + +### Make “parity mode” a first-class configuration + +Right now, the parity benchmarks exercise: + +- scratch begin/finish every task, citeturn18view0turn27view0 +- timestamp mode defaults, citeturn13view2turn18view0 +- full scheduler semantics, including tight park/unpark loops. citeturn18view0turn13view1 + +Tokio/Go “trivial task throughput” numbers are typically achieved in regimes designed to minimize OS-level thread parking and to keep hot workers active with local queues and cheap notification. citeturn23view0turn29search1 + +To avoid repeatedly debating whether benchmark numbers reflect “optional features,” the project should treat “parity mode” as an explicit config preset: + +- `timestamp_mode = disabled` +- scratch fast-path that is effectively lock-free unless scratch is used +- wake strategy set to “idle-set / O(1) notify” rather than scan +- optionally: parameters tuned to minimize parking churn (longer spin, larger inject batch, etc.), but behind explicit config knobs rather than hidden behavior + +This makes benchmark comparisons honest: you’re comparing scheduler and wake machinery vs scheduler and wake machinery, not “scheduler + extra cross-thread memory subsystem” vs “scheduler.” citeturn13view0turn13view2turn21view1 + +## Measurement and risk controls + +The measurement discipline documented in the handoff is not optional; it’s necessary to avoid re-learning the same lessons: + +- Use paired same-session A/B comparisons and at least `n=10` for `flood_1m`. citeturn9view0turn30view1 +- Apply the “20× ReleaseFast stress gate” to any submit-path or wake-path change; keep the Phase H claim sentinel enabled while doing so. citeturn9view0turn18view0 +- Treat host drift as real; do not anchor on stale absolute baselines (including the earlier 566K number unless revalidated under clean host state). citeturn9view0turn30view2 + +For the two concrete patches suggested above, the acceptance criteria should be framed as: + +- **Patch: scratch lock removal** + - No regression in `test-stress` (≥20×). + - Benchmark deltas should be visible both in `flood_1m` and in any worker-heavy scenario (this change affects worker-side overhead). citeturn27view0turn18view0 + +- **Patch: idle-worker queue notify** + - No lost-wakeup hangs under stress (this touches the most sensitive correctness surface). + - A/B improvement should show most strongly in the parity benchmarks, where submit-side scan overhead is paid at high frequency. citeturn13view1turn18view0 + +Finally, one hard truth from the existing evidence should guide expectations: even “perfect” versions of these fixes are unlikely to magically reach 5–15M ops/s without deeper structural alignment with Tokio/Go’s idle management and wake semantics. The fixes above are the ones that remove the current architecture’s biggest brakes so that a more Tokio-like idle coordinator and per-worker parker state machine can be implemented without constantly fighting O(N) per-submit work and per-task synchronization overhead. citeturn30view2turn23view0turn18view0 diff --git a/docs/performance/2026-04-14-scratch-fastpath-idle-queue/README.md b/docs/performance/2026-04-14-scratch-fastpath-idle-queue/README.md new file mode 100644 index 0000000..8d9a0cf --- /dev/null +++ b/docs/performance/2026-04-14-scratch-fastpath-idle-queue/README.md @@ -0,0 +1,116 @@ +# Scratch Fast Path + Idle Queue Attempt — 2026-04-14 + +## Status + +Implemented in the current working tree only. Not merged to `main`. + +This archive covers the late 2026-04-14 attempt driven by +`docs/codex-review-2026-04-14.md`: + +- remove per-task scratch locking from the worker hot path +- replace submit-path O(N) parked-worker scanning with an idle-worker queue +- add a first-class parity-mode preset +- preserve the existing `queued_tasks` / `is_parked` lost-wakeup model + +## Code changes in the workspace + +- `src/memory/scratch.zig` + - `beginTask()`, `alloc()`, and `finishTask()` are lock-free on the normal + owning-worker path + - backing buffer allocation stays lazy + - code comments now document the quiescent-purge invariant +- `src/pool.zig` + - `purgeWorkerCaches()` now returns `error.PoolBusy` unless + `active_tasks == 0 && queued_tasks == 0` + - added idle-worker ticket queues and the `idle_queue` wake strategy + - kept the existing SC `queued_tasks` / `is_parked` handshake intact +- `src/types.zig` / `src/root.zig` + - exported `WakeStrategy` and `RuntimePreset` + - `RuntimePreset.tokio_parity` currently enables + `timestamp_mode = .disabled` and `wake_strategy = .idle_queue` +- `tests/unit.zig` / `tests/stress.zig` + - added regression coverage for quiescent purge, idle-queue wake behavior, + and lost-wakeup stress under `wake_strategy = .idle_queue` + +## Correctness finding from the attempt + +The first cut of the scratch fast-path refactor introduced a real race: +idle-time purge and external quiescent purge could overlap and double-free a +scratch buffer. The symptom was an intermittent `SIGSEGV` in +`stress.test.repeated churn returns to a bounded steady state after purge`. + +Final fix: + +- keep the task fast path lock-free +- add a slow-path-only purge mutex so overlapping purge callers serialize + +That preserves the hot-path goal while closing the double-free. + +## Validation on the final workspace + +- `zig build test --summary all` PASS +- `zig build test-integration --summary all` PASS +- `zig build test-stress --summary all` PASS +- focused purge/churn regression loop: + `zig build test-stress -- --test-filter "repeated churn returns to a bounded steady state after purge"` + 30/30 PASS +- required `ReleaseFast` stress gate: + `zig build test-stress --summary all` + 20/20 PASS + +## Benchmark method + +Same-session A/B versus a clean baseline worktree at `HEAD`: + +- baseline worktree: `/tmp/thread-pool-baseline` +- candidate: current dirty workspace +- build mode: `ReleaseFast` +- run count: n=10 baseline, n=10 candidate +- ordering: balanced order, alternating which binary runs first per pair + +Raw outputs: + +- `baseline-balanced-n10.txt` +- `after-balanced-n10.txt` + +## Host state during measurement + +The host was under substantial unrelated load during the benchmark run: + +- `uptime`: `load average: 12.89, 13.63, 11.13` +- `ps -eo pid,comm,%cpu,%mem --sort=-%cpu | head -n 12` showed several + unrelated `geode` processes consuming significant CPU + +This must be acknowledged for benchmark variance and absolute numbers. It does +**not** invalidate the correctness regression that was fixed, and it does not +justify claiming a perf win that the same-session A/B does not show. + +## Results + +| Scenario | Baseline median ops/s | Baseline p95 | Candidate median ops/s | Candidate p95 | Delta | +|---|---:|---:|---:|---:|---:| +| `tokio_parity/flood_1m` | 487,840.5 | 537,976 | 465,650.0 | 541,624 | -4.5% median | +| `tokio_parity/contended_spawn_8x` | 469,570.0 | 510,690 | 411,959.0 | 453,933 | -12.3% median | + +## Additional note on parity preset + +During the session, the benchmark harness was briefly changed to force +`runtime_preset = .tokio_parity`. On this host that made the measured parity +benchmarks worse, so the harness was restored to its stable config. The preset +remains available as an explicit runtime option; it is not forced by the +benchmark binary. + +## Verdict + +This attempt improved the implementation shape and closed a real purge race, +but it did **not** meet the performance success criteria: + +- `flood_1m` did not show a same-session improvement +- `contended_spawn_8x` regressed materially on the measured host +- the idle-queue wake path is therefore not ready to become the default + +The next session should treat this as: + +- a correctness-clean, tested workspace +- useful infrastructure for further wake-path investigation +- not yet a benchmark-positive patch to ship as-is diff --git a/docs/performance/2026-04-14-scratch-fastpath-idle-queue/after-balanced-n10.txt b/docs/performance/2026-04-14-scratch-fastpath-idle-queue/after-balanced-n10.txt new file mode 100644 index 0000000..e1e5d51 --- /dev/null +++ b/docs/performance/2026-04-14-scratch-fastpath-idle-queue/after-balanced-n10.txt @@ -0,0 +1,20 @@ +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2174519971 ops_per_sec=459872 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=220296915 ops_per_sec=453933 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=1955764804 ops_per_sec=511309 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=233805854 ops_per_sec=427705 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2362874274 ops_per_sec=423213 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=236875468 ops_per_sec=422163 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2193616734 ops_per_sec=455868 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=245277167 ops_per_sec=407702 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2253258364 ops_per_sec=443802 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=226697725 ops_per_sec=441116 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2121215011 ops_per_sec=471428 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=269695196 ops_per_sec=370789 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2106527636 ops_per_sec=474715 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=253447143 ops_per_sec=394560 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=1884849028 ops_per_sec=530546 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=269099552 ops_per_sec=371610 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2390447979 ops_per_sec=418332 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=240259897 ops_per_sec=416216 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=1846298615 ops_per_sec=541624 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=251820142 ops_per_sec=397109 diff --git a/docs/performance/2026-04-14-scratch-fastpath-idle-queue/baseline-balanced-n10.txt b/docs/performance/2026-04-14-scratch-fastpath-idle-queue/baseline-balanced-n10.txt new file mode 100644 index 0000000..51d4027 --- /dev/null +++ b/docs/performance/2026-04-14-scratch-fastpath-idle-queue/baseline-balanced-n10.txt @@ -0,0 +1,20 @@ +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2104931378 ops_per_sec=475075 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=220876438 ops_per_sec=452742 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2930059638 ops_per_sec=341290 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=204415039 ops_per_sec=489201 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=1986808630 ops_per_sec=503320 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=222514362 ops_per_sec=449409 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2057590461 ops_per_sec=486005 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=195813361 ops_per_sec=510690 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2201313083 ops_per_sec=454274 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=206582086 ops_per_sec=484069 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=1920730209 ops_per_sec=520635 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=223056920 ops_per_sec=448316 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2042165391 ops_per_sec=489676 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=232207557 ops_per_sec=430649 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=1858818286 ops_per_sec=537976 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=203437444 ops_per_sec=491552 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2093585423 ops_per_sec=477649 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=205397846 ops_per_sec=486860 +tokio_parity scenario=flood_1m tasks=1000000 elapsed_ns=2025504682 ops_per_sec=493704 +tokio_parity scenario=contended_spawn_8x tasks=100000 submitters=8 elapsed_ns=219745897 ops_per_sec=455071 diff --git a/src/memory/scratch.zig b/src/memory/scratch.zig index 68df826..23ce0e3 100644 --- a/src/memory/scratch.zig +++ b/src/memory/scratch.zig @@ -39,12 +39,16 @@ pub const WorkerScratch = struct { fba: ?std.heap.FixedBufferAllocator = null, current_bytes: usize = 0, peak_bytes: usize = 0, - /// Serializes purge() and beginTask()/alloc() so that the owning worker - /// and a remote caller (e.g. Core.purgeWorkerCaches) cannot race on the - /// backing buffer. Without this guard, two concurrent purge() calls would - /// double-free the allocation. - mutex: std.Thread.Mutex = .{}, - + /// Slow-path serialization for overlapping purge callers. The normal + /// owning-worker task path remains lock-free; this only prevents + /// external quiescent purge and the worker's own idle-time purge from + /// racing each other and double-freeing the backing buffer. + purge_mutex: std.Thread.Mutex = .{}, + /// Invariant: the worker task fast path is single-thread-owned. `purge()` + /// is only legal once the pool is quiescent (`active_tasks == 0 && + /// queued_tasks == 0`) or from the owning worker's idle path after it has + /// left task execution. That lets `beginTask()`, `alloc()`, and + /// `finishTask()` stay lock-free on the normal path. pub fn init(allocator: std.mem.Allocator, accounting: Accounting, cap_bytes: usize) WorkerScratch { return .{ .allocator = allocator, @@ -54,12 +58,11 @@ pub const WorkerScratch = struct { } pub fn beginTask(self: *WorkerScratch) !void { - self.mutex.lock(); - defer self.mutex.unlock(); - try self.ensureBufferLocked(); + self.current_bytes = 0; if (self.buffer) |buf| { self.fba = std.heap.FixedBufferAllocator.init(buf); - self.current_bytes = 0; + } else { + self.fba = null; } } @@ -73,9 +76,7 @@ pub const WorkerScratch = struct { } pub fn alloc(self: *WorkerScratch, comptime T: type, count: usize) ![]T { - self.mutex.lock(); - defer self.mutex.unlock(); - try self.ensureBufferLocked(); + try self.ensureBuffer(); if (self.fba == null) return error.ScratchUnavailable; var fba = &self.fba.?; const slice = try fba.allocator().alloc(T, count); @@ -85,8 +86,8 @@ pub const WorkerScratch = struct { } pub fn purge(self: *WorkerScratch) void { - self.mutex.lock(); - defer self.mutex.unlock(); + self.purge_mutex.lock(); + defer self.purge_mutex.unlock(); if (self.buffer) |buf| { self.accounting.recordFree(buf.len); self.allocator.free(buf); @@ -104,8 +105,7 @@ pub const WorkerScratch = struct { return self.peak_bytes; } - /// Must be called with self.mutex held. - fn ensureBufferLocked(self: *WorkerScratch) !void { + fn ensureBuffer(self: *WorkerScratch) !void { if (self.cap_bytes == 0 or self.buffer != null) return; const buf = try self.allocator.alloc(u8, self.cap_bytes); self.accounting.recordAlloc(buf.len); @@ -136,3 +136,81 @@ test "worker scratch respects fixed cap and purge" { scratch.purge(); try std.testing.expectEqual(@as(usize, 0), scratch.bytesCurrent()); } + +test "worker scratch defers allocation until first alloc and reuses the buffer" { + var total_alloc = std.atomic.Value(u64).init(0); + var total_free = std.atomic.Value(u64).init(0); + var total_count = std.atomic.Value(u64).init(0); + var live = std.atomic.Value(u64).init(0); + var peak = std.atomic.Value(u64).init(0); + const accounting = Accounting{ + .total_alloc_bytes = &total_alloc, + .total_free_bytes = &total_free, + .total_alloc_count = &total_count, + .live_bytes = &live, + .peak_live_bytes = &peak, + }; + + var scratch = WorkerScratch.init(std.testing.allocator, accounting, 64); + + try scratch.beginTask(); + scratch.finishTask(); + try std.testing.expectEqual(@as(u64, 0), total_count.load(.acquire)); + try std.testing.expectEqual(@as(u64, 0), live.load(.acquire)); + + try scratch.beginTask(); + _ = try scratch.alloc(u8, 16); + scratch.finishTask(); + try std.testing.expectEqual(@as(u64, 1), total_count.load(.acquire)); + try std.testing.expectEqual(@as(u64, 64), live.load(.acquire)); + + try scratch.beginTask(); + _ = try scratch.alloc(u8, 8); + scratch.finishTask(); + try std.testing.expectEqual(@as(u64, 1), total_count.load(.acquire)); + + scratch.purge(); + try std.testing.expectEqual(@as(u64, 64), total_free.load(.acquire)); + try std.testing.expectEqual(@as(u64, 0), live.load(.acquire)); +} + +test "worker scratch serializes overlapping purge callers" { + var total_alloc = std.atomic.Value(u64).init(0); + var total_free = std.atomic.Value(u64).init(0); + var total_count = std.atomic.Value(u64).init(0); + var live = std.atomic.Value(u64).init(0); + var peak = std.atomic.Value(u64).init(0); + const accounting = Accounting{ + .total_alloc_bytes = &total_alloc, + .total_free_bytes = &total_free, + .total_alloc_count = &total_count, + .live_bytes = &live, + .peak_live_bytes = &peak, + }; + + var scratch = WorkerScratch.init(std.testing.allocator, accounting, 64); + try scratch.beginTask(); + _ = try scratch.alloc(u8, 16); + scratch.finishTask(); + + const Purger = struct { + fn run(shared: *WorkerScratch, start: *std.atomic.Value(u8)) void { + while (start.load(.acquire) == 0) { + std.atomic.spinLoopHint(); + } + shared.purge(); + } + }; + + var start = std.atomic.Value(u8).init(0); + const t1 = try std.Thread.spawn(.{}, Purger.run, .{ &scratch, &start }); + const t2 = try std.Thread.spawn(.{}, Purger.run, .{ &scratch, &start }); + start.store(1, .release); + t1.join(); + t2.join(); + + try std.testing.expectEqual(@as(u64, 1), total_count.load(.acquire)); + try std.testing.expectEqual(@as(u64, 64), total_free.load(.acquire)); + try std.testing.expectEqual(@as(u64, 0), live.load(.acquire)); + try std.testing.expectEqual(@as(usize, 0), scratch.bytesCurrent()); +} diff --git a/src/pool.zig b/src/pool.zig index c42eccc..91b2c7f 100644 --- a/src/pool.zig +++ b/src/pool.zig @@ -165,6 +165,9 @@ const LIFO_SLOT_EMPTY: u32 = 0xFFFF_FFFF; /// flip this back to `std.debug.runtime_safety` or remove the sentinel /// entirely. const claim_sentinel_enabled: bool = true; +const idle_ticket_none: u8 = 0; +const idle_ticket_queued: u8 = 1; +const idle_ticket_consumed: u8 = 2; const Worker = struct { thread: ?std.Thread = null, @@ -268,6 +271,9 @@ const Core = struct { free_indices: []u32, free_top_tagged: std.atomic.Value(u64) = std.atomic.Value(u64).init(0), inject_queue: BoundedRing(u32), + idle_queues: ?[]BoundedRing(u32) = null, + idle_queue_shard_mask: usize = 0, + idle_ticket_states: ?[]std.atomic.Value(u8) = null, sync: Sync = .{}, counters: Counters align(std.atomic.cache_line) = .{}, worker_stats_buffer: []types.WorkerStats, @@ -275,6 +281,38 @@ const Core = struct { pub fn init(self: *Core, allocator: std.mem.Allocator, config: types.Config) !void { var normalized = config.normalized(); try normalized.validate(); + var idle_queues: ?[]BoundedRing(u32) = null; + var idle_queue_shard_mask: usize = 0; + var idle_ticket_states: ?[]std.atomic.Value(u8) = null; + if (normalized.wake_strategy == .idle_queue) { + const idle_queue_shards = idleQueueShardCount(normalized.max_threads); + const idle_queue_slots = idleQueueCapacity(idleQueueShardSlots(normalized.max_threads, idle_queue_shards)); + const queues = try allocator.alloc(BoundedRing(u32), idle_queue_shards); + var idle_queues_transferred = false; + errdefer if (!idle_queues_transferred) allocator.free(queues); + var idle_queues_inited: usize = 0; + errdefer if (!idle_queues_transferred) { + while (idle_queues_inited > 0) { + idle_queues_inited -= 1; + queues[idle_queues_inited].deinit(); + } + }; + for (queues) |*idle_queue| { + idle_queue.* = try BoundedRing(u32).init(allocator, idle_queue_slots); + idle_queues_inited += 1; + } + + const tickets = try allocator.alloc(std.atomic.Value(u8), normalized.max_threads); + errdefer if (!idle_queues_transferred) allocator.free(tickets); + for (tickets) |*ticket| { + ticket.* = std.atomic.Value(u8).init(idle_ticket_none); + } + + idle_queues = queues; + idle_queue_shard_mask = queues.len - 1; + idle_ticket_states = tickets; + idle_queues_transferred = true; + } self.* = .{ .allocator = allocator, @@ -284,6 +322,9 @@ const Core = struct { .descriptors = try allocator.alloc(TaskDescriptor, normalized.queue_capacity), .free_indices = try allocator.alloc(u32, normalized.queue_capacity), .inject_queue = try BoundedRing(u32).init(allocator, normalized.queue_capacity), + .idle_queues = idle_queues, + .idle_queue_shard_mask = idle_queue_shard_mask, + .idle_ticket_states = idle_ticket_states, .worker_stats_buffer = try allocator.alloc(types.WorkerStats, normalized.max_threads), }; errdefer self.deinitOwned(); @@ -302,6 +343,15 @@ const Core = struct { self.recordOwnedAlloc(self.free_indices.len * @sizeOf(u32)); self.recordOwnedAlloc(self.worker_stats_buffer.len * @sizeOf(types.WorkerStats)); self.recordOwnedAlloc(self.inject_queue.capacity() * @sizeOf(u32)); + if (self.idle_queues) |queues| { + self.recordOwnedAlloc(queues.len * @sizeOf(BoundedRing(u32))); + for (queues) |*idle_queue| { + self.recordOwnedAlloc(idle_queue.capacity() * @sizeOf(u32)); + } + } + if (self.idle_ticket_states) |tickets| { + self.recordOwnedAlloc(tickets.len * @sizeOf(std.atomic.Value(u8))); + } for (self.descriptors, 0..) |*descriptor, idx| { descriptor.* = .{}; @@ -540,11 +590,59 @@ const Core = struct { pub fn purgeWorkerCaches(self: *Core) !void { if (self.counters.active_tasks.load(.acquire) != 0) return error.PoolBusy; + if (self.counters.queued_tasks.load(.acquire) != 0) return error.PoolBusy; for (self.workers) |*worker| { worker.scratch.purge(); } } + fn wakeOneIdleWorkerFromQueue(self: *Core) void { + const idle_queues = self.idle_queues orelse return; + const idle_ticket_states = self.idle_ticket_states orelse return; + const start = @as(usize, @truncate(self.counters.submitted.load(.monotonic))) & self.idle_queue_shard_mask; + var shard_off: usize = 0; + while (shard_off < idle_queues.len) : (shard_off += 1) { + const shard_index = (start + shard_off) & self.idle_queue_shard_mask; + var idle_queue = &idle_queues[shard_index]; + while (idle_queue.pop()) |candidate| { + const idx: usize = @intCast(candidate); + const worker = &self.workers[idx]; + idle_ticket_states[idx].store(idle_ticket_consumed, .release); + + if (worker.is_parked.load(.seq_cst) != 1) continue; + + worker.park_mutex.lock(); + if (worker.is_parked.load(.monotonic) == 1) { + worker.park_cond.signal(); + worker.park_mutex.unlock(); + return; + } + worker.park_mutex.unlock(); + } + } + } + + fn clearConsumedIdleTicket(self: *Core, worker_index: usize) void { + if (self.config.wake_strategy != .idle_queue) return; + const idle_ticket_states = self.idle_ticket_states orelse return; + if (idle_ticket_states[worker_index].load(.acquire) == idle_ticket_consumed) { + idle_ticket_states[worker_index].store(idle_ticket_none, .release); + } + } + + fn publishIdleWorker(self: *Core, worker_index: usize) void { + if (self.config.wake_strategy != .idle_queue) return; + const idle_queues = self.idle_queues orelse return; + const idle_ticket_states = self.idle_ticket_states orelse return; + if (idle_ticket_states[worker_index].cmpxchgStrong(idle_ticket_none, idle_ticket_queued, .acq_rel, .acquire)) |_| { + return; + } + const shard_index = worker_index & self.idle_queue_shard_mask; + if (!idle_queues[shard_index].tryPush(@intCast(worker_index))) { + idle_ticket_states[worker_index].store(idle_ticket_none, .release); + } + } + fn submitInternal(self: *Core, task_fn: TaskFn, ctx: *anyopaque, options: types.SubmitOptions, allow_block: bool) !types.TaskHandle { const deadline = if (allow_block and options.backpressure == .block_with_timeout and options.timeout_ns != null) nowNs() + options.timeout_ns.? @@ -610,32 +708,35 @@ const Core = struct { } _ = self.counters.submitted.fetchAdd(1, .monotonic); - self.incrementQueued(); - - // Wake one parked worker. Scan `is_parked` across workers starting at a - // round-robin offset so load spreads across the pool. The submit-before-park - // handshake (see idle()) guarantees: if any worker is about to miss this - // task because it's heading to park, its `is_parked := 1` store is ordered - // with its reload of `queued_tasks`. So either we observe `is_parked == 1` - // and signal it, OR the worker reloads `queued_tasks` and sees our push — - // no lost wakeup. No broadcast fallback is needed. - const submitted = self.counters.submitted.load(.monotonic); - const hint_idx: usize = @intCast(submitted % self.workers.len); - var off: usize = 0; - while (off < self.workers.len) : (off += 1) { - const idx = (hint_idx + off) % self.workers.len; - const w = &self.workers[idx]; - if (w.is_parked.load(.seq_cst) == 1) { - w.park_mutex.lock(); - // Re-check under the mutex: the worker may have unparked between our - // load and this acquire. The signal is a no-op if it isn't parked — - // but we do not want to hold the mutex for a worker that is running. - if (w.is_parked.load(.monotonic) == 1) { - w.park_cond.signal(); + _ = self.incrementQueued(); + if (self.config.wake_strategy == .scan_parked_workers) { + // Wake one parked worker. Scan `is_parked` across workers starting at a + // round-robin offset so load spreads across the pool. The submit-before-park + // handshake (see idle()) guarantees: if any worker is about to miss this + // task because it's heading to park, its `is_parked := 1` store is ordered + // with its reload of `queued_tasks`. So either we observe `is_parked == 1` + // and signal it, OR the worker reloads `queued_tasks` and sees our push — + // no lost wakeup. No broadcast fallback is needed. + const submitted = self.counters.submitted.load(.monotonic); + const hint_idx: usize = @intCast(submitted % self.workers.len); + var off: usize = 0; + while (off < self.workers.len) : (off += 1) { + const idx = (hint_idx + off) % self.workers.len; + const w = &self.workers[idx]; + if (w.is_parked.load(.seq_cst) == 1) { + w.park_mutex.lock(); + // Re-check under the mutex: the worker may have unparked between our + // load and this acquire. The signal is a no-op if it isn't parked — + // but we do not want to hold the mutex for a worker that is running. + if (w.is_parked.load(.monotonic) == 1) { + w.park_cond.signal(); + } + w.park_mutex.unlock(); + break; } - w.park_mutex.unlock(); - break; } + } else { + self.wakeOneIdleWorkerFromQueue(); } return if (options.waitable) @@ -870,6 +971,7 @@ const Core = struct { fn idle(self: *Core, worker_index: usize) void { var worker = &self.workers[worker_index]; + self.clearConsumedIdleTicket(worker_index); // Phase G2 (re-land 2026-04-13): no `counters.queued_tasks` poll. // The submit path writes that line with .seq_cst on every submit; // 56 workers × spin_iters shared loads per park cycle dominated @@ -918,6 +1020,14 @@ const Core = struct { // reads that store's specific value — if the SC load observes an earlier // value, no HB edge forms and the IRIW scenario is permitted. worker.is_parked.store(1, .seq_cst); + // With the idle-queue wake strategy, publish our wake ticket after + // `is_parked := 1` and before the `queued_tasks` SC reload. That keeps + // the same lost-wakeup surface as the legacy scan path: after the + // submitter's SC fetchAdd, either it dequeues this ticket and signals + // under `park_mutex`, or our SC load below sees the queued task and we + // skip waiting. Stale queue entries are safe: the submit path validates + // `is_parked` under the mutex before signaling. + self.publishIdleWorker(worker_index); if (self.counters.queued_tasks.load(.seq_cst) != 0 or self.state.load(.acquire) != .running) { worker.is_parked.store(0, .release); worker.park_mutex.unlock(); @@ -1071,6 +1181,18 @@ const Core = struct { lq.deinit(); } self.inject_queue.deinit(); + if (self.idle_queues) |queues| { + for (queues) |*idle_queue| { + self.recordOwnedFree(idle_queue.capacity() * @sizeOf(u32)); + idle_queue.deinit(); + } + self.recordOwnedFree(queues.len * @sizeOf(BoundedRing(u32))); + self.allocator.free(queues); + } + if (self.idle_ticket_states) |tickets| { + self.recordOwnedFree(tickets.len * @sizeOf(std.atomic.Value(u8))); + self.allocator.free(tickets); + } self.recordOwnedFree(self.worker_stats_buffer.len * @sizeOf(types.WorkerStats)); self.allocator.free(self.worker_stats_buffer); self.recordOwnedFree(self.free_indices.len * @sizeOf(u32)); @@ -1083,13 +1205,14 @@ const Core = struct { self.allocator.free(self.workers); } - fn incrementQueued(self: *Core) void { + fn incrementQueued(self: *Core) u64 { // Must be .seq_cst: the submit-path park handshake requires this store // to participate in the SC total order so that the IRIW scenario // (submitter reads is_parked=0 AND worker reads queued_tasks=0) is // excluded. See the comment in idle() for the full argument. const queue_depth = self.counters.queued_tasks.fetchAdd(1, .seq_cst) +% 1; updatePeak(&self.counters.queue_depth_peak, queue_depth); + return queue_depth; } fn decrementQueued(self: *Core) void { @@ -1180,8 +1303,10 @@ pub const ThreadPool = struct { } /// Release scratch memory held by idle workers. Returns - /// `error.PoolBusy` if any tasks are currently active, since - /// purging during execution would corrupt scratch allocators. + /// `error.PoolBusy` unless the pool is quiescent + /// (`active_tasks == 0 && queued_tasks == 0`), since purging during + /// execution or with queued work remaining would corrupt scratch + /// allocator ownership. pub fn purgeWorkerCaches(self: *ThreadPool) !void { return self.core.purgeWorkerCaches(); } @@ -1202,3 +1327,19 @@ fn updatePeakU32(counter: *std.atomic.Value(u32), value: u32) void { current = previous.?; } } + +fn idleQueueCapacity(max_threads: usize) usize { + const desired = @max(max_threads * 2, 2); + return std.math.ceilPowerOfTwo(usize, desired) catch desired; +} + +fn idleQueueShardCount(max_threads: usize) usize { + const capped = @min(@max(max_threads, 1), 8); + var shards: usize = 1; + while ((shards << 1) <= capped) : (shards <<= 1) {} + return shards; +} + +fn idleQueueShardSlots(max_threads: usize, shard_count: usize) usize { + return std.math.divCeil(usize, max_threads, shard_count) catch max_threads; +} diff --git a/src/root.zig b/src/root.zig index 920863b..55f83fe 100644 --- a/src/root.zig +++ b/src/root.zig @@ -20,6 +20,8 @@ pub const SchedulerVariant = types.SchedulerVariant; pub const QueueKind = types.QueueKind; pub const TaskLane = types.TaskLane; pub const BackpressurePolicy = types.BackpressurePolicy; +pub const WakeStrategy = types.WakeStrategy; +pub const RuntimePreset = types.RuntimePreset; pub const SubmitOptions = types.SubmitOptions; pub const ShutdownMode = types.ShutdownMode; pub const TaskHandle = types.TaskHandle; diff --git a/src/types.zig b/src/types.zig index 415bb69..cc2563a 100644 --- a/src/types.zig +++ b/src/types.zig @@ -33,6 +33,23 @@ pub const TimestampMode = enum { disabled, }; +/// Controls how submitters pick a worker to wake after enqueueing work. +/// `scan_parked_workers` preserves the legacy O(N) parked-worker scan. +/// `idle_queue` uses a bounded idle-worker ticket queue to select wake +/// candidates without reading every worker's `is_parked` flag. +pub const WakeStrategy = enum { + scan_parked_workers, + idle_queue, +}; + +/// Named runtime presets for benchmark-focused configuration bundles. +/// Presets intentionally only touch clearly documented knobs; callers can +/// still override unrelated fields directly in the `Config` literal. +pub const RuntimePreset = enum { + standard, + tokio_parity, +}; + /// Task lifecycle state machine: /// /// free ──> reserved ──> queued ──> running ──> completed @@ -120,6 +137,8 @@ pub const Config = struct { /// tasks with `.lane = .blocking`. allow_blocking_lane: bool = false, timestamp_mode: TimestampMode = .full, + wake_strategy: WakeStrategy = .scan_parked_workers, + runtime_preset: RuntimePreset = .standard, /// Maximum number of tasks a stealer takes from a single victim per steal /// attempt. With `queue_kind = .ring`, true batch-steal via stealHalf is /// used and values > 1 are correct. With `queue_kind = .chase_lev`, @@ -129,6 +148,13 @@ pub const Config = struct { pub fn normalized(self: Config) Config { var cfg = self; + switch (cfg.runtime_preset) { + .standard => {}, + .tokio_parity => { + cfg.timestamp_mode = .disabled; + cfg.wake_strategy = .idle_queue; + }, + } const cpu_guess = @max(std.Thread.getCpuCount() catch 1, 1); if (cfg.max_threads == 0) { cfg.max_threads = @min(cpu_guess, 64); @@ -334,3 +360,9 @@ test "Config normalized rounds queue_capacity to power of two" { const cfg = (Config{ .queue_capacity = 1000 }).normalized(); try std.testing.expectEqual(@as(usize, 1024), cfg.queue_capacity); } + +test "Config tokio_parity preset enables idle queue wake and disabled timestamps" { + const cfg = (Config{ .runtime_preset = .tokio_parity }).normalized(); + try std.testing.expectEqual(TimestampMode.disabled, cfg.timestamp_mode); + try std.testing.expectEqual(WakeStrategy.idle_queue, cfg.wake_strategy); +} diff --git a/tests/stress.zig b/tests/stress.zig index 516f73e..d3590c1 100644 --- a/tests/stress.zig +++ b/tests/stress.zig @@ -73,6 +73,7 @@ test "park handshake: no lost wakeups under submit-while-parking race" { .scheduler_variant = .work_stealing, .idle_park_after_ns = 50_000, .spin_before_park_iters = 4, + .wake_strategy = .idle_queue, }); defer pool.deinit(); @@ -526,6 +527,7 @@ test "phase g2: aggressive spin flood - no lost wakeups" { .local_queue_capacity = 256, .scheduler_variant = .work_stealing, .spin_before_park_iters = 4096, + .wake_strategy = .idle_queue, }); defer pool.deinit(); diff --git a/tests/unit.zig b/tests/unit.zig index 3243b2b..bf1e6a4 100644 --- a/tests/unit.zig +++ b/tests/unit.zig @@ -72,6 +72,16 @@ test "worker cache purge resets current scratch usage" { } } +test "worker cache purge rejects queued work even when no task is active" { + var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.work_stealing)); + defer pool.deinit(); + + pool.core.counters.queued_tasks.store(1, .release); + defer pool.core.counters.queued_tasks.store(0, .release); + + try std.testing.expectError(error.PoolBusy, pool.purgeWorkerCaches()); +} + test "waitable handle generation increments across reuses" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.global_queue)); defer pool.deinit(); @@ -552,6 +562,27 @@ test "per-worker park: submit wakes a parked worker" { try std.testing.expect(stats.wakeups > 0); } +test "idle queue wake strategy: submit wakes a parked worker" { + var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ + .max_threads = 4, + .queue_capacity = 128, + .idle_park_after_ns = 10_000_000, + .wake_strategy = .idle_queue, + }); + defer pool.deinit(); + + std.Thread.sleep(50_000_000); + + var wg = pool.waitGroup(); + var counter = common.Counter{}; + _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); + wg.wait(); + + try std.testing.expectEqual(@as(u64, 1), counter.value.load(.acquire)); + const stats = pool.statsSnapshot(); + try std.testing.expect(stats.wakeups > 0); +} + test "Stats.started and .completed aggregate from per-worker counters" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 2,