From d578e15747da2378eded4a74b5c1831459100948 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 20 Oct 2026 11:30:21 -0700 Subject: [PATCH 1/3] Run a single execution worker on the calling thread execute_work_queue spawned and joined a scoped thread even when it had one worker, and the engine's default is one worker. On F5 tier M under the application-root profile, spawning and joining that thread took about 31% of each tick, while the work it ran was about 2% of the run. A single worker now runs the same loop inline. Several workers still each get a thread. Nothing it computes changes. Carried onto main from 43a8ddda on claude/stack-envelope, as an isolated candidate with no dependency on M1, M2 or lazy blockers. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/parallel/exec.rs | 93 ++++++++++++++++----------- 1 file changed, 54 insertions(+), 39 deletions(-) diff --git a/crates/warp-core/src/parallel/exec.rs b/crates/warp-core/src/parallel/exec.rs index 9c39a7f1..28f8619d 100644 --- a/crates/warp-core/src/parallel/exec.rs +++ b/crates/warp-core/src/parallel/exec.rs @@ -1079,50 +1079,18 @@ where let next_unit = AtomicUsize::new(0); + // One worker gains nothing from a thread: it would run the same loop + // while this thread waits to join it. Run it here instead. + if workers == 1 { + return vec![drain_units(units, &next_unit, &resolve_store)]; + } + std::thread::scope(|s| { let handles: Vec<_> = (0..workers) .map(|_| { - let units = &units; let next_unit = &next_unit; let resolve_store = &resolve_store; - - s.spawn(move || -> WorkerResult { - let mut delta = TickDelta::new(); - let mut evidence = Vec::new(); - - // Work-stealing loop: claim units until none remain - loop { - let unit_idx = next_unit.fetch_add(1, Ordering::Relaxed); - if unit_idx >= units.len() { - break; - } - - let unit = &units[unit_idx]; - - // Resolve view for this warp (per-unit, NOT cached across units) - let Some(store) = resolve_store(&unit.warp_id) else { - return WorkerResult::MissingStore(unit.warp_id); - }; - - // Execute items SERIALLY (no nested threading!) - for (idx, item) in unit.items.iter().enumerate() { - match execute_item_enforced(store, item, idx, unit, delta) { - Ok((next_delta, record)) => { - delta = next_delta; - evidence.push(record); - } - Err(mut poisoned) => { - poisoned.prepend_evidence(evidence); - return WorkerResult::Poisoned(poisoned); - } - } - } - - // View dropped here - no long-lived borrows across warps - } - - WorkerResult::Success(WorkerExecution::new(delta, evidence)) - }) + s.spawn(move || drain_units(units, next_unit, resolve_store)) }) .collect(); @@ -1136,6 +1104,53 @@ where }) } +/// One worker's loop: claims units until none remain, executing each unit's +/// items serially against its warp's store. +fn drain_units<'state, F>( + units: &[WorkUnit], + next_unit: &AtomicUsize, + resolve_store: &F, +) -> WorkerResult +where + F: Fn(&WarpId) -> Option<&'state GraphStore> + Sync, +{ + let mut delta = TickDelta::new(); + let mut evidence = Vec::new(); + + // Work-stealing loop: claim units until none remain + loop { + let unit_idx = next_unit.fetch_add(1, Ordering::Relaxed); + if unit_idx >= units.len() { + break; + } + + let unit = &units[unit_idx]; + + // Resolve view for this warp (per-unit, NOT cached across units) + let Some(store) = resolve_store(&unit.warp_id) else { + return WorkerResult::MissingStore(unit.warp_id); + }; + + // Execute items SERIALLY (no nested threading!) + for (idx, item) in unit.items.iter().enumerate() { + match execute_item_enforced(store, item, idx, unit, delta) { + Ok((next_delta, record)) => { + delta = next_delta; + evidence.push(record); + } + Err(mut poisoned) => { + poisoned.prepend_evidence(evidence); + return WorkerResult::Poisoned(poisoned); + } + } + } + + // View dropped here - no long-lived borrows across warps + } + + WorkerResult::Success(WorkerExecution::new(delta, evidence)) +} + struct ItemExecutionOutcome { delta: TickDelta, evidence: ExecutionFootprintEvidence, From 5d97ab0932863646b9b6df23ec448a770aa3009a Mon Sep 17 00:00:00 2001 From: James Ross Date: Wed, 21 Oct 2026 04:53:05 -0700 Subject: [PATCH 2/3] Witness error propagation and document inline worker threads Add two witnesses that a single worker running inline on the caller's thread reports failures exactly as scoped worker threads do: - rule_panic_propagates_identically_for_every_worker_count: one of twenty scopes panics in its executor; for workers 1, 2, 4, 8, 16, 32 the commit unwinds with the executor's payload, retains no evidence, keeps the poisoned item's evidence as a failure artifact, and leaves the state root unchanged. - missing_store_propagates_identically_for_every_worker_count: an unresolvable warp yields exactly one MissingStore for 1, 2, 4, 8 workers, every other worker an empty success. Both pass with the inline path and with it disabled (the old threaded one-worker path), and both fail against a mutant whose inline path swallows failures. Document on execute_work_queue, EngineBuilder::workers and in the changelog that with one worker rules run on the caller's thread and stack, so native rules must not rely on a fresh thread-local environment per tick. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 ++ crates/warp-core/src/engine_impl.rs | 4 + crates/warp-core/src/parallel/exec.rs | 60 ++++++++++- .../parallel_engine_worker_invariance.rs | 99 +++++++++++++++++++ 4 files changed, 167 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 928e9360..ac1fa64f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -95,6 +95,12 @@ ### Changed +- With one execution worker (the engine default), a tick now runs its rules + on the thread that commits it instead of spawning a thread and waiting for + it. Results, their order, and panic payloads are unchanged; two or more + workers still each get a scoped thread. Native rules run on the caller's + stack and must not rely on a fresh thread-local environment per tick. + - Provider semantic declarations now support explicitly bounded Core byte aliases, including exact 32-byte identifiers. Byte aliases participate in bounded record graphs and generate canonical lawpack type definitions. diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index 53717d85..d4062075 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -344,6 +344,10 @@ impl EngineBuilder { /// - Workers are capped at `NUM_SHARDS` (256) internally /// - Values less than 1 are treated as 1 /// - Use explicit worker counts in benchmarks to compare parallelism levels + /// - With one worker (including when the cap leaves one), rules execute on + /// the thread that calls commit, on its stack; no thread is spawned. With + /// two or more, each worker runs on its own scoped thread. Native rules + /// must not rely on a fresh thread-local environment per tick. #[must_use] pub fn workers(mut self, n: usize) -> Self { self.worker_count = n.max(1); diff --git a/crates/warp-core/src/parallel/exec.rs b/crates/warp-core/src/parallel/exec.rs index 28f8619d..33c5b629 100644 --- a/crates/warp-core/src/parallel/exec.rs +++ b/crates/warp-core/src/parallel/exec.rs @@ -1028,6 +1028,18 @@ pub fn build_work_units( /// This is the **only** spawn site for cross-warp execution. Workers claim /// units atomically and execute items serially within each unit. /// +/// # Threads +/// +/// With `workers >= 2`, each worker runs on its own scoped thread. With +/// `workers == 1` (the engine default) no thread is spawned: the single +/// worker runs on the caller's thread and stack, and returns before this +/// function does. Rule executors therefore cannot rely on a fresh thread or +/// thread-local environment per tick. Thread-local values a rule sets persist +/// on the caller's thread across ticks (as they already do per worker thread +/// with two or more workers), and a deeply recursive rule uses the caller's +/// stack rather than a spawned thread's default stack. Results, their order, +/// and panic payloads are the same either way. +/// /// # Footprint Enforcement (cfg-gated) /// /// When enforcement is active, the worker loop: @@ -1060,7 +1072,10 @@ pub fn build_work_units( /// /// # Panics /// -/// Panics if `workers == 0` or if any worker thread panics. +/// Panics if `workers == 0`, or if a worker panics outside the per-item +/// `catch_unwind` (for example, the enforcement invariant on guards). With one +/// worker the panic unwinds directly from the caller's thread; with more it is +/// re-raised from the join with the same payload. pub fn execute_work_queue<'state, F>( units: &[WorkUnit], workers: usize, @@ -1080,7 +1095,8 @@ where let next_unit = AtomicUsize::new(0); // One worker gains nothing from a thread: it would run the same loop - // while this thread waits to join it. Run it here instead. + // while this thread waits to join it. Run it here instead. Rules then run + // on the caller's thread and stack (see "Threads" above). if workers == 1 { return vec![drain_units(units, &next_unit, &resolve_store)]; } @@ -1852,4 +1868,44 @@ mod tests { (ParallelExecutionPolicy::DYNAMIC_PER_SHARD, worker_hint(4),) ); } + + /// A missing store surfaces identically whether the single worker runs + /// inline on the caller's thread or workers run on scoped threads: exactly + /// one `MissingStore` naming the unresolvable warp, and every other worker + /// an empty success. The engine maps that to `EngineError::UnknownWarp`. + #[test] + fn missing_store_propagates_identically_for_every_worker_count() { + use super::{execute_work_queue, WorkUnit, WorkerResult}; + use crate::make_warp_id; + + let missing = make_warp_id("exec-test/missing-store"); + let units = [WorkUnit { + warp_id: missing, + items: Vec::new(), + #[cfg(any(debug_assertions, feature = "footprint_enforce_release"))] + #[cfg(not(feature = "unsafe_graph"))] + guards: Vec::new(), + }]; + + for workers in [1_usize, 2, 4, 8] { + let results = execute_work_queue(&units, workers, |_| None); + assert_eq!(results.len(), workers, "one result per worker"); + let missing_stores: Vec<_> = results + .iter() + .filter_map(|result| match result { + WorkerResult::MissingStore(warp_id) => Some(*warp_id), + _ => None, + }) + .collect(); + assert_eq!(missing_stores, vec![missing], "{workers} workers"); + let empty_successes = results + .iter() + .filter(|result| { + matches!(result, WorkerResult::Success(execution) + if execution.delta.is_empty() && execution.evidence.is_empty()) + }) + .count(); + assert_eq!(empty_successes, workers - 1, "{workers} workers"); + } + } } diff --git a/crates/warp-core/tests/parallel_engine_worker_invariance.rs b/crates/warp-core/tests/parallel_engine_worker_invariance.rs index e8e87bb3..dd0b1f0c 100644 --- a/crates/warp-core/tests/parallel_engine_worker_invariance.rs +++ b/crates/warp-core/tests/parallel_engine_worker_invariance.rs @@ -589,3 +589,102 @@ fn multiwarp_worker_count_invariance_stress() { } } } + +/// A rule panic surfaces identically whether the single worker runs inline on +/// the caller's thread (`workers(1)`) or workers run on scoped threads. +/// +/// One of twenty independent scopes panics in its executor. For every worker +/// count the commit must unwind with the executor's own payload, commit +/// nothing, retain no evidence, keep the poisoned item's evidence as a +/// failure artifact, and leave the state root unchanged. +#[test] +fn rule_panic_propagates_identically_for_every_worker_count() { + use std::panic::{catch_unwind, AssertUnwindSafe}; + use warp_core::{ApplyResult, EngineBuilder, NodeRecord}; + + const RULE_NAME: &str = "t16p/touch"; + const PAYLOAD: &str = "t16p: executor failure on the poisoned scope"; + + let node_ty = warp_core::make_type_id("t16p/node"); + let mut base_store = warp_core::GraphStore::default(); + let root = warp_core::make_node_id("t16p/root"); + base_store.insert_node(root, NodeRecord { ty: node_ty }); + let mut scopes = vec![root]; + for i in 1..20 { + let scope = warp_core::make_node_id(&format!("t16p/node{i}")); + base_store.insert_node(scope, NodeRecord { ty: node_ty }); + scopes.push(scope); + } + let poisoned = warp_core::make_node_id("t16p/node7"); + + let make_rule = || { + let mut rule = make_touch_rule!(RULE_NAME, "t16p/marker", b"touched-t16p"); + rule.executor = warp_core::RuleExecutor::observed( + |view: &mut warp_core::ExecutionGraphView<'_, '_>, scope, delta| { + if *scope == warp_core::make_node_id("t16p/node7") { + panic!("t16p: executor failure on the poisoned scope"); + } + let key = warp_core::AttachmentKey::node_alpha(warp_core::NodeKey { + warp_id: view.warp_id(), + local_id: *scope, + }); + let payload = warp_core::AtomPayload::new( + warp_core::make_type_id("t16p/marker"), + bytes::Bytes::from_static(b"touched-t16p"), + ); + delta.push(warp_core::WarpOp::SetAttachment { + key, + value: Some(warp_core::AttachmentValue::Atom(payload)), + }); + }, + ); + rule + }; + + let run = |workers: usize| -> warp_core::Hash { + let mut engine = EngineBuilder::new(base_store.clone(), root) + .workers(workers) + .build(); + engine.register_rule(make_rule()).expect("register rule"); + let before = engine.snapshot().state_root; + let tx = engine.begin(); + for scope in &scopes { + assert!(matches!( + engine.apply(tx, RULE_NAME, scope), + Ok(ApplyResult::Applied) + )); + } + + let Err(payload) = catch_unwind(AssertUnwindSafe(|| engine.commit(tx))) else { + panic!("commit must unwind with {workers} workers"); + }; + assert_eq!( + payload.downcast_ref::<&str>().copied(), + Some(PAYLOAD), + "panic payload changed with {workers} workers" + ); + assert!( + engine.last_execution_footprints().is_empty(), + "a failed tick retains no evidence ({workers} workers)" + ); + assert!( + engine + .last_aborted_execution_footprints() + .iter() + .any(|record| record.scope() == poisoned), + "the poisoned item's evidence is kept as a failure artifact ({workers} workers)" + ); + let after = engine.snapshot().state_root; + assert_eq!(before, after, "state changed with {workers} workers"); + after + }; + + let baseline = run(1); + for &workers in WORKER_COUNTS { + assert_eq!( + run(workers), + baseline, + "outcome differs for {workers} workers" + ); + } +} From c82a23cdd563c2ec042720067b68a7a7be87bf18 Mon Sep 17 00:00:00 2001 From: James Ross Date: Wed, 21 Oct 2026 04:56:47 -0700 Subject: [PATCH 3/3] Use assert! for the poisoned-scope panic in the witness clippy::manual_assert rejected an if-then panic! in the new engine witness. assert! with a literal message keeps the same &str payload. Co-Authored-By: Claude Opus 5.5 --- .../warp-core/tests/parallel_engine_worker_invariance.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/crates/warp-core/tests/parallel_engine_worker_invariance.rs b/crates/warp-core/tests/parallel_engine_worker_invariance.rs index dd0b1f0c..5732dc98 100644 --- a/crates/warp-core/tests/parallel_engine_worker_invariance.rs +++ b/crates/warp-core/tests/parallel_engine_worker_invariance.rs @@ -621,9 +621,10 @@ fn rule_panic_propagates_identically_for_every_worker_count() { let mut rule = make_touch_rule!(RULE_NAME, "t16p/marker", b"touched-t16p"); rule.executor = warp_core::RuleExecutor::observed( |view: &mut warp_core::ExecutionGraphView<'_, '_>, scope, delta| { - if *scope == warp_core::make_node_id("t16p/node7") { - panic!("t16p: executor failure on the poisoned scope"); - } + assert!( + *scope != warp_core::make_node_id("t16p/node7"), + "t16p: executor failure on the poisoned scope" + ); let key = warp_core::AttachmentKey::node_alpha(warp_core::NodeKey { warp_id: view.warp_id(), local_id: *scope,