From 86e9007667019ed982cbd816b3e6f4695fc9ea8f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 4 Sep 2026 09:53:24 -0400 Subject: [PATCH 1/2] communication: a receive-side gate for deterministic simulation A `Gate` sits on the receive side of the serializing allocators, `ProcessAllocator` and `TcpAllocator`. It physically receives all bytes. Built open, the default, bytes pass through as before. Built holding, it splits received bytes into framed messages, files them by source worker, and surfaces a message only when released. Held messages raise no events, so operators see only what has been logically delivered. Per-source FIFO order is preserved; the holder chooses only the interleaving across sources. Messages a worker sends to itself are never held, matching real allocators. Whether a gate holds is a builder setting, `ProcessBuilder::holding`, fixed when the allocator is built. Building several `Bytes` process allocators sequentially on one thread deadlocked, because each `build` waits on its peers. `ProcessBuilder::build_all` fulfills every obligation before completing any allocator. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01LEpSYjkqw2zfhGSMAXNXyD --- communication/src/allocator/mod.rs | 27 +++++ .../src/allocator/zero_copy/allocator.rs | 18 ++- .../allocator/zero_copy/allocator_process.rs | 55 +++++++-- communication/src/allocator/zero_copy/gate.rs | 108 ++++++++++++++++++ communication/src/allocator/zero_copy/mod.rs | 1 + 5 files changed, 193 insertions(+), 16 deletions(-) create mode 100644 communication/src/allocator/zero_copy/gate.rs diff --git a/communication/src/allocator/mod.rs b/communication/src/allocator/mod.rs index 8c7cf6893..72c1831f8 100644 --- a/communication/src/allocator/mod.rs +++ b/communication/src/allocator/mod.rs @@ -151,6 +151,33 @@ impl ProcessBuilder { } } + /// Sets whether the built allocator's receive gate holds messages until released. + /// + /// Only `Bytes` allocators have a gate; the setting is ignored for `Typed` allocators. + pub fn holding(self, holding: bool) -> Self { + match self { + ProcessBuilder::Typed(t) => ProcessBuilder::Typed(t), + ProcessBuilder::Bytes(b) => ProcessBuilder::Bytes(b.holding(holding)), + } + } + + /// Builds a vector of peers on the calling thread. + /// + /// `Bytes` builders block in `build` until their peers are built, and so must be + /// built together when they share a thread; `Typed` builders build independently. + pub fn build_all(builders: Vec) -> Vec { + if builders.iter().all(|builder| matches!(builder, ProcessBuilder::Bytes(_))) { + let bytes = builders.into_iter().map(|builder| match builder { + ProcessBuilder::Bytes(b) => b, + ProcessBuilder::Typed(_) => unreachable!("checked above"), + }).collect(); + BytesProcessBuilder::build_all(bytes).into_iter().map(Process::Bytes).collect() + } + else { + builders.into_iter().map(|builder| builder.build()).collect() + } + } + /// Constructs a vector of regular (mpsc-based, "Typed") intra-process builders. pub fn new_typed_vector(peers: usize, refill: BytesRefill, spill: Option) -> Vec { ::new_vector(peers, refill, spill) diff --git a/communication/src/allocator/zero_copy/allocator.rs b/communication/src/allocator/zero_copy/allocator.rs index 59ce805a1..a7aa536be 100644 --- a/communication/src/allocator/zero_copy/allocator.rs +++ b/communication/src/allocator/zero_copy/allocator.rs @@ -13,7 +13,8 @@ use crate::allocator::{Process, ProcessBuilder, Exchangeable}; use crate::allocator::canary::Canary; use crate::allocator::zero_copy::bytes_slab::BytesRefill; use crate::allocator::zero_copy::spill::SpillPolicyFn; -use super::bytes_exchange::{BytesPull, SendEndpoint, MergeQueue}; +use super::bytes_exchange::{SendEndpoint, MergeQueue}; +use super::gate::Gate; use super::push_pull::{Pusher, PullerInner}; /// Builds an instance of a TcpAllocator. @@ -124,7 +125,7 @@ impl TcpBuilder { channel_id_bound: None, staged: Vec::new(), sends, - recvs, + gate: Rc::new(RefCell::new(Gate::new(self.index, self.peers, recvs, false))), to_local: HashMap::new(), } } @@ -145,10 +146,17 @@ pub struct TcpAllocator { // sending, receiving, and responding to binary buffers. sends: Vec>>>, // sends[x] -> goes to process x. - recvs: Vec, // recvs[x] <- from process x. + gate: Rc>, // receives from all processes. to_local: HashMap>>>, // to worker-local typed pullers. } +impl TcpAllocator { + /// The gate through which this allocator receives messages from other processes. + /// + /// Messages from workers in the same process pass through the inner allocator instead. + pub fn gate(&self) -> Rc> { Rc::clone(&self.gate) } +} + impl Allocate for TcpAllocator { fn index(&self) -> usize { self.index } fn peers(&self) -> usize { self.peers } @@ -264,9 +272,7 @@ impl Allocate for TcpAllocator { self.inner.receive(); - for recv in self.recvs.iter_mut() { - recv.drain_into(&mut self.staged); - } + self.gate.borrow_mut().receive(&mut self.staged); let mut events = self.inner.events().borrow_mut(); diff --git a/communication/src/allocator/zero_copy/allocator_process.rs b/communication/src/allocator/zero_copy/allocator_process.rs index 72a1e94b8..6a1688392 100644 --- a/communication/src/allocator/zero_copy/allocator_process.rs +++ b/communication/src/allocator/zero_copy/allocator_process.rs @@ -14,7 +14,8 @@ use crate::allocator::{AllocateBuilder, Exchangeable, PeerBuilder}; use crate::allocator::canary::Canary; use crate::allocator::zero_copy::bytes_slab::BytesRefill; use crate::allocator::zero_copy::spill::SpillPolicyFn; -use super::bytes_exchange::{BytesPull, SendEndpoint, MergeQueue}; +use super::bytes_exchange::{SendEndpoint, MergeQueue}; +use super::gate::Gate; use super::push_pull::{Pusher, Puller}; @@ -31,6 +32,7 @@ pub struct ProcessBuilder { pullers: Vec>, // for pulling bytes from other workers. refill: BytesRefill, spill: Option, // optional spill factory for recv queues. + holding: bool, // whether the receive gate holds messages. } impl PeerBuilder for ProcessBuilder { @@ -55,6 +57,7 @@ impl PeerBuilder for ProcessBuilder { pullers, refill: refill.clone(), spill: spill.clone(), + holding: false, } ) .collect() @@ -62,12 +65,36 @@ impl PeerBuilder for ProcessBuilder { } impl ProcessBuilder { + /// Sets whether the built allocator's receive gate holds messages until released. + pub fn holding(mut self, holding: bool) -> Self { + self.holding = holding; + self + } + /// Builds a `ProcessAllocator`, instantiating `Rc>` elements. - pub fn build(self) -> ProcessAllocator { + /// + /// Each builder must be built on its own thread, or all together with + /// [`build_all`](Self::build_all): `build` blocks until every peer has + /// fulfilled its obligations, which a peer does only when it is itself built. + pub fn build(mut self) -> ProcessAllocator { + let recvs = self.fulfill(); + self.complete(recvs) + } + + /// Builds a vector of peers on the calling thread. + /// + /// All obligations are fulfilled before any allocator is completed, so no + /// builder blocks waiting on a peer that has not yet been built. + pub fn build_all(mut builders: Vec) -> Vec { + let recvs: Vec<_> = builders.iter_mut().map(|builder| builder.fulfill()).collect(); + builders.into_iter().zip(recvs).map(|(builder, recvs)| builder.complete(recvs)).collect() + } - // Fulfill puller obligations. + /// Fulfills puller obligations: creates this allocator's receive queues on the + /// calling thread, sending the writer halves to peers and returning the readers. + fn fulfill(&mut self) -> Vec { let mut recvs = Vec::with_capacity(self.peers); - for puller in self.pullers.into_iter() { + for puller in std::mem::take(&mut self.pullers) { let buzzer = crate::buzzer::Buzzer::default(); let (writer, reader) = match self.spill.as_ref() { Some(build_fn) => { @@ -81,8 +108,11 @@ impl ProcessBuilder { recvs.push(reader); puller.send(writer).expect("Failed to send MergeQueue"); } + recvs + } - // Extract pusher commitments. + /// Extracts pusher commitments, blocking until each peer has fulfilled them. + fn complete(self, recvs: Vec) -> ProcessAllocator { let mut sends = Vec::with_capacity(self.peers); for pusher in self.pushers.into_iter() { let queue = pusher.recv().expect("Failed to receive MergeQueue"); @@ -98,7 +128,7 @@ impl ProcessBuilder { channel_id_bound: None, staged: Vec::new(), sends, - recvs, + gate: Rc::new(RefCell::new(Gate::new(self.index, self.peers, recvs, self.holding))), to_local: HashMap::new(), } } @@ -128,10 +158,17 @@ pub struct ProcessAllocator { // sending, receiving, and responding to binary buffers. staged: Vec, sends: Vec>>>, // sends[x] -> goes to thread x. - recvs: Vec, // recvs[x] <- from thread x. + gate: Rc>, // receives from all threads. to_local: HashMap>>>, // to worker-local typed pullers. } +impl ProcessAllocator { + /// The gate through which this allocator receives messages from its peers. + /// + /// A gate built holding lets a driver control message delivery, for deterministic simulation. + pub fn gate(&self) -> Rc> { Rc::clone(&self.gate) } +} + impl Allocate for ProcessAllocator { fn index(&self) -> usize { self.index } fn peers(&self) -> usize { self.peers } @@ -191,9 +228,7 @@ impl Allocate for ProcessAllocator { let mut events = self.events.borrow_mut(); - for recv in self.recvs.iter_mut() { - recv.drain_into(&mut self.staged); - } + self.gate.borrow_mut().receive(&mut self.staged); for mut bytes in self.staged.drain(..) { diff --git a/communication/src/allocator/zero_copy/gate.rs b/communication/src/allocator/zero_copy/gate.rs new file mode 100644 index 000000000..ca197f5f5 --- /dev/null +++ b/communication/src/allocator/zero_copy/gate.rs @@ -0,0 +1,108 @@ +//! Receive-side admission of framed messages. +//! +//! A [`Gate`] sits between the byte queues an allocator physically receives from and the +//! typed channels it surfaces messages into. By default it is open: bytes pass through +//! unchanged, and the gate costs one additional `Vec` move per `receive()`. +//! +//! A gate constructed holding splits received bytes into framed messages and files them by +//! source worker. A held message is physically present but logically undelivered: it is +//! not surfaced to its channel and raises no event, until the holder of the gate releases +//! it. Per-source FIFO order is preserved; the holder chooses only the interleaving +//! across sources, which is exactly the schedule space a real transport can produce. +//! This is the mechanism behind deterministic simulation of multi-worker computations. +//! +//! Messages a worker sends to itself are never held. Real allocators surface them at the +//! next `receive()`, and holding them would admit schedules no deployment can produce. + +use std::collections::VecDeque; + +use timely_bytes::arc::Bytes; + +use crate::networking::MessageHeader; +use super::bytes_exchange::{BytesPull, MergeQueue}; + +/// Admission control over bytes received from peers. +pub struct Gate { + /// The owning worker's index, whose own messages are never held. + index: usize, + /// Physical sources of bytes. Each may carry messages from several source workers. + recvs: Vec, + /// Whether received messages are held until released. + holding: bool, + /// Held messages, one framed message per entry, indexed by source worker. + held: Vec>, + /// Bytes admitted for delivery, each containing whole framed messages. + admitted: Vec, +} + +impl Gate { + /// Creates a gate over `recvs`, for worker `index` of `peers`, holding if `holding`. + pub fn new(index: usize, peers: usize, recvs: Vec, holding: bool) -> Self { + Gate { + index, + recvs, + holding, + held: (0 .. peers).map(|_| VecDeque::new()).collect(), + admitted: Vec::new(), + } + } + + /// Drains the physical sources, admitting or holding what they contain. + /// + /// The allocator calls this from `receive()`. A driver may also call it, to see + /// what has arrived without stepping the worker. + pub fn fetch(&mut self) { + if !self.holding { + for recv in self.recvs.iter_mut() { + recv.drain_into(&mut self.admitted); + } + } + else { + let mut staged = Vec::new(); + for recv in self.recvs.iter_mut() { + recv.drain_into(&mut staged); + } + for mut bytes in staged { + // Received bytes contain whole framed messages; no splitting across allocations. + while !bytes.is_empty() { + let header = MessageHeader::try_read(&bytes[..]).expect("failed to read full header!"); + let message = bytes.extract_to(header.required_bytes()); + if header.source == self.index { + self.admitted.push(message); + } + else { + self.held[header.source].push_back(message); + } + } + } + } + } + + /// Fetches, then moves all admitted bytes into `into`. + pub fn receive(&mut self, into: &mut Vec) { + self.fetch(); + into.append(&mut self.admitted); + } + + /// The number of held messages from `source`, as of the last fetch. + pub fn held(&self, source: usize) -> usize { + self.held[source].len() + } + + /// Releases up to `count` held messages from `source`, in FIFO order. + /// + /// Returns the number released, which is less than `count` if fewer are held. + /// This clamping keeps any sequence of releases valid, which matters when + /// shrinking failing schedules. + pub fn release(&mut self, source: usize, count: usize) -> usize { + let held = &mut self.held[source]; + let count = std::cmp::min(count, held.len()); + self.admitted.extend(held.drain(.. count)); + count + } + + /// Releases all held messages, and returns the number released. + pub fn release_all(&mut self) -> usize { + (0 .. self.held.len()).map(|source| self.release(source, usize::MAX)).sum() + } +} diff --git a/communication/src/allocator/zero_copy/mod.rs b/communication/src/allocator/zero_copy/mod.rs index 8d92f05d4..b30a4f2d6 100644 --- a/communication/src/allocator/zero_copy/mod.rs +++ b/communication/src/allocator/zero_copy/mod.rs @@ -10,6 +10,7 @@ pub mod bytes_slab; pub mod bytes_exchange; +pub mod gate; pub mod spill; pub mod tcp; pub mod allocator; From 331d4729bc37faf773ce16d20e4a3ead5bbfd71e Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 4 Sep 2026 09:53:24 -0400 Subject: [PATCH 2/2] timely: deterministic simulation of multi-worker computations Adds `timely::simulate`: several workers hosted on the calling thread, using ordinary `Bytes` process allocators with holding gates. An execution is a pure function of the applied sequence of `Decision`s: step a worker, or deliver up to N messages from one worker to another. Any decision sequence is valid, since deliveries clamp to what is pending, so schedule traces are recordable, replayable, and shrinkable by construction. Tests run multi-worker barriers under seeded random schedules, asserting progress safety. A property-testing grind checks frontier safety, conservation, quiescence, and same-seed determinism. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01LEpSYjkqw2zfhGSMAXNXyD --- timely/src/lib.rs | 1 + timely/src/simulate.rs | 156 +++++++++++++++++++++++ timely/tests/grind.rs | 255 +++++++++++++++++++++++++++++++++++++ timely/tests/simulation.rs | 155 ++++++++++++++++++++++ 4 files changed, 567 insertions(+) create mode 100644 timely/src/simulate.rs create mode 100644 timely/tests/grind.rs create mode 100644 timely/tests/simulation.rs diff --git a/timely/src/lib.rs b/timely/src/lib.rs index 39625f7e3..c38222803 100644 --- a/timely/src/lib.rs +++ b/timely/src/lib.rs @@ -92,6 +92,7 @@ pub mod dataflow; pub mod synchronization; pub mod execute; pub mod order; +pub mod simulate; pub mod logging; // pub mod log_events; diff --git a/timely/src/simulate.rs b/timely/src/simulate.rs new file mode 100644 index 000000000..9049f505d --- /dev/null +++ b/timely/src/simulate.rs @@ -0,0 +1,156 @@ +//! Deterministic simulation of multi-worker timely computations. +//! +//! A [`Simulation`] hosts several workers on the calling thread, using the ordinary +//! serializing intra-process allocator with each worker's receive [`Gate`] holding: every +//! message is physically received, but is logically delivered only when the caller says +//! so. Each worker is deterministic given the messages delivered to it, and delivery is +//! constrained only by per-source FIFO order, so an execution is a pure function of the +//! sequence of [`Decision`]s applied: the decision trace *is* the execution. Traces are +//! trivially recordable, replayable, and shrinkable, which makes this the substrate for +//! randomized schedule exploration ("deterministic simulation testing") of +//! progress-tracking behavior. +//! +//! # Examples +//! ```rust +//! use timely::simulate::{Decision, Simulation}; +//! use timely::dataflow::operators::{ToStream, Inspect}; +//! +//! let mut sim = Simulation::new(2); +//! for index in 0 .. 2 { +//! sim.worker_mut(index).dataflow::<(),_,_>(|scope| { +//! (0 .. 10).to_stream(scope) +//! .container::>() +//! .inspect(|x| println!("seen: {:?}", x)); +//! }); +//! } +//! // Interleave worker steps and message deliveries however the test likes ... +//! sim.apply(Decision::Step(0)); +//! sim.apply(Decision::Deliver { source: 0, target: 1, count: 1 }); +//! // ... then run to completion. +//! assert!(sim.drain(1_000)); +//! ``` + +use std::rc::Rc; +use std::cell::RefCell; +use std::sync::Arc; + +use crate::WorkerConfig; +use crate::communication::Allocator; +use crate::communication::allocator::{Process, ProcessBuilder}; +use crate::communication::allocator::zero_copy::gate::Gate; +use crate::communication::allocator::zero_copy::bytes_slab::BytesRefill; +use crate::worker::Worker; + +/// One decision in a simulation schedule. +/// +/// Any sequence of decisions is valid: stepping a worker with nothing to do and +/// delivering on an empty stream are both no-ops. This keeps randomly generated +/// and mechanically shrunk schedules well-formed by construction. +#[derive(Clone, Debug)] +pub enum Decision { + /// Run one step of the indicated worker. + Step(usize), + /// Deliver up to `count` undelivered messages from `source` to `target`, in FIFO order. + Deliver { + /// The sending worker. + source: usize, + /// The receiving worker. + target: usize, + /// The maximum number of messages to deliver. + count: usize, + }, +} + +/// Several workers on one thread, with caller-controlled message delivery. +pub struct Simulation { + /// Each worker's receive gate, holding for the duration of the simulation. + gates: Vec>>, + workers: Vec, +} + +impl Simulation { + /// Creates a simulation of `peers` workers with default worker configuration. + /// + /// The workers are constructed without a timer, so time-based reschedulings + /// (`activate_after`) degrade to immediate activation and logging is disabled; + /// nothing in a simulated execution reads the wall clock. + pub fn new(peers: usize) -> Self { + let refill = BytesRefill { + logic: Arc::new(|size| Box::new(vec![0_u8; size]) as Box+Send>), + limit: None, + }; + let mut gates = Vec::with_capacity(peers); + let mut workers = Vec::with_capacity(peers); + let builders = ProcessBuilder::new_bytes_vector(peers, refill, None) + .into_iter() + .map(|builder| builder.holding(true)) + .collect(); + for process in ProcessBuilder::build_all(builders) { + let gate = match &process { + Process::Bytes(allocator) => allocator.gate(), + _ => unreachable!("new_bytes_vector produces Bytes allocators"), + }; + gates.push(gate); + workers.push(Worker::new(WorkerConfig::default(), Allocator::Process(process), None)); + } + Simulation { gates, workers } + } + + /// The number of simulated workers. + pub fn peers(&self) -> usize { self.workers.len() } + + /// Mutable access to a worker, e.g. to install dataflows or inspect probes. + pub fn worker_mut(&mut self, index: usize) -> &mut Worker { + &mut self.workers[index] + } + + /// The gate of `target`, with everything sent to it so far fetched and held. + fn gate(&self, target: usize) -> std::cell::RefMut<'_, Gate> { + let mut gate = self.gates[target].borrow_mut(); + gate.fetch(); + gate + } + + /// The number of undelivered messages from `source` to `target`. + pub fn pending(&self, source: usize, target: usize) -> usize { + self.gate(target).held(source) + } + + /// Applies one schedule decision. + pub fn apply(&mut self, decision: Decision) { + match decision { + Decision::Step(index) => { self.workers[index].step(); } + Decision::Deliver { source, target, count } => { self.deliver(source, target, count); } + } + } + + /// Runs one step of the indicated worker, returning whether dataflows remain. + pub fn step_worker(&mut self, index: usize) -> bool { + self.workers[index].step() + } + + /// Delivers up to `count` messages from `source` to `target`; returns the number delivered. + pub fn deliver(&mut self, source: usize, target: usize, count: usize) -> usize { + self.gate(target).release(source, count) + } + + /// Runs the simulation to completion under a fair schedule: repeatedly deliver + /// everything and step every worker, until no dataflows and no messages remain. + /// + /// Returns `true` if the simulation quiesced within `bound` rounds. A `false` + /// return after a generous bound indicates a genuine liveness problem, as the + /// schedule from here on is maximally fair. + pub fn drain(&mut self, bound: usize) -> bool { + for _ in 0 .. bound { + let mut active = false; + for target in 0 .. self.peers() { + active |= self.gate(target).release_all() > 0; + } + for worker in self.workers.iter_mut() { + active |= worker.step(); + } + if !active { return true; } + } + false + } +} diff --git a/timely/tests/grind.rs b/timely/tests/grind.rs new file mode 100644 index 000000000..2c0839d2f --- /dev/null +++ b/timely/tests/grind.rs @@ -0,0 +1,255 @@ +//! Property-testing grind over simulated schedules. +//! +//! Each run builds a "chaos" dataflow on several workers — records spread over input +//! times, exchanged by value, redistributed in time by a capability-holding `delay`, +//! exchanged again — and executes it under a seeded, policy-biased schedule of worker +//! steps and message deliveries. Oracles checked on every run: +//! +//! - **Frontier safety**: an auditing operator asserts that no record arrives at a +//! time its input frontier had already passed on a previous scheduling. This is the +//! observable form of the progress-tracking safety property. +//! - **Conservation**: after a fair drain, the multiset of (time, value) pairs observed +//! across all workers equals exactly what was introduced (no loss, duplication, or +//! mistiming through exchange, serialization, and delay). +//! - **Quiescence**: the fair drain terminates within a generous bound. +//! - **Determinism**: identical seeds produce identical observation logs. + +use std::rc::Rc; +use std::cell::RefCell; +use std::panic::AssertUnwindSafe; + +use timely::simulate::{Decision, Simulation}; +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::{ToStream, Exchange}; +use timely::dataflow::operators::vec::Delay; +use timely::dataflow::operators::generic::operator::Operator; +use timely::progress::Antichain; + +/// Records per worker, distinct input times, and maximum delay, per run. +const RECORDS: u64 = 40; +const TIMES: u64 = 5; +const DELAYS: u64 = 4; + +/// A tiny deterministic RNG (SplitMix64); the harness stays free of external deps. +struct SplitMix64(u64); +impl SplitMix64 { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + fn below(&mut self, bound: usize) -> usize { + (self.next() % (bound as u64)) as usize + } +} + +/// A deterministic hash for workload choices (input times, delays, routing). +fn mix(seed: u64, value: u64, salt: u64) -> u64 { + let mut rng = SplitMix64(seed ^ value.wrapping_mul(0x9E37_79B9_7F4A_7C15) ^ salt); + rng.next() +} + +fn initial_time(wseed: u64, value: u64) -> usize { (mix(wseed, value, 0xA) % TIMES) as usize } +fn delay_by(wseed: u64, value: u64) -> usize { (mix(wseed, value, 0xB) % DELAYS) as usize } + +/// Schedule-generation policies; diversity of bias matters more than seed count. +#[derive(Clone, Copy, Debug)] +enum Policy { + /// Even mix of steps and deliveries. + Uniform, + /// Mostly steps; messages pile up undelivered. + StepHeavy, + /// Mostly deliveries; workers rarely run. + DeliverHeavy, + /// Even mix, but one stream never delivers until the drain. + StarveStream(usize, usize), + /// Long alternating phases of steps-only and deliveries-only. + Bursty, +} + +fn policy_for(rng: &mut SplitMix64, peers: usize) -> Policy { + match rng.next() % 5 { + 0 => Policy::Uniform, + 1 => Policy::StepHeavy, + 2 => Policy::DeliverHeavy, + 3 => Policy::StarveStream(rng.below(peers), rng.below(peers)), + _ => Policy::Bursty, + } +} + +/// Builds the chaos dataflow on each worker; returns the shared observation log. +fn build_chaos(sim: &mut Simulation, peers: usize, wseed: u64) -> Rc>> { + + let results = Rc::new(RefCell::new(Vec::new())); + + for index in 0 .. peers { + let results = Rc::clone(&results); + sim.worker_mut(index).dataflow::(move |scope| { + let base = (index as u64) * RECORDS; + (base .. base + RECORDS) + .to_stream(scope) + .delay(move |v, _t| initial_time(wseed, *v)) + .exchange(move |v| mix(wseed, *v, 0xC)) + .delay(move |v, t| t + delay_by(wseed, *v)) + .exchange(move |v| mix(wseed, *v, 0xD)) + .unary_frontier::>, _, _, _>( + Pipeline, + "Auditor", + move |_capability, _info| { + // The frontier as of the end of the previous scheduling: a + // promise that no future record arrives at a time it passed. + let mut previous: Antichain = Antichain::from_elem(0); + move |(input, frontier), _output| { + input.for_each_time(|time, data| { + assert!( + previous.less_equal(time.time()), + "frontier safety violated: records at {:?} after frontier {:?}", + time.time(), previous.elements(), + ); + // Delivered-but-unconsumed records hold the frontier, + // so even the current frontier may not pass their time. + assert!( + frontier.less_equal(time.time()), + "frontier safety violated: records at {:?} under frontier {:?}", + time.time(), frontier.frontier(), + ); + for datum in data.flat_map(|d| d.drain(..)) { + results.borrow_mut().push((*time.time(), datum)); + } + }); + previous = frontier.frontier().to_owned(); + } + } + ); + }); + } + + results +} + +/// The (time, value) multiset every run must observe, independent of schedule. +fn expected(peers: usize, wseed: u64) -> Vec<(usize, u64)> { + let mut expected = Vec::new(); + for value in 0 .. (peers as u64) * RECORDS { + expected.push((initial_time(wseed, value) + delay_by(wseed, value), value)); + } + expected.sort(); + expected +} + +/// Runs one seeded schedule against one seeded workload; returns the observation log. +fn chaos_run(peers: usize, wseed: u64, sseed: u64, prefix: usize) -> Vec<(usize, u64)> { + + let mut sim = Simulation::new(peers); + let results = build_chaos(&mut sim, peers, wseed); + + let mut rng = SplitMix64(sseed); + let policy = policy_for(&mut rng, peers); + + for round in 0 .. prefix { + let step = + match policy { + Policy::Uniform => rng.next() % 100 < 50, + Policy::StepHeavy => rng.next() % 100 < 90, + Policy::DeliverHeavy => rng.next() % 100 < 10, + Policy::StarveStream(_, _) => rng.next() % 100 < 50, + Policy::Bursty => (round / 100) % 2 == 0, + }; + let decision = + if step { + Decision::Step(rng.below(peers)) + } + else { + let source = rng.below(peers); + let target = rng.below(peers); + if let Policy::StarveStream(s, t) = policy { + if (source, target) == (s, t) { continue; } + } + Decision::Deliver { source, target, count: 1 + rng.below(4) } + }; + sim.apply(decision); + } + + assert!(sim.drain(50_000), "simulation failed to quiesce"); + + let mut log = Rc::try_unwrap(results).expect("operators should be dropped").into_inner(); + + // Conservation: exactly the expected records, at exactly the expected times. + let mut sorted = log.clone(); + sorted.sort(); + assert_eq!(sorted, expected(peers, wseed), "conservation violated"); + + log.sort(); // return in canonical order; arrival order is checked by same-seed runs on raw logs. + log +} + +/// As `chaos_run`, but returns the log in arrival order for determinism comparison. +fn chaos_run_raw(peers: usize, wseed: u64, sseed: u64, prefix: usize) -> Vec<(usize, u64)> { + let mut sim = Simulation::new(peers); + let results = build_chaos(&mut sim, peers, wseed); + let mut rng = SplitMix64(sseed); + let _policy = policy_for(&mut rng, peers); + for _ in 0 .. prefix { + let decision = + if rng.next() % 2 == 0 { Decision::Step(rng.below(peers)) } + else { + Decision::Deliver { source: rng.below(peers), target: rng.below(peers), count: 1 + rng.below(4) } + }; + sim.apply(decision); + } + assert!(sim.drain(50_000), "simulation failed to quiesce"); + Rc::try_unwrap(results).expect("operators should be dropped").into_inner() +} + +#[test] +fn chaos_small_grind() { + for wseed in 0 .. 8 { + for sseed in 0 .. 8 { + chaos_run(3, wseed, sseed, 2_000); + } + } +} + +#[test] +fn chaos_wide_grind() { + for wseed in 0 .. 4 { + for sseed in 0 .. 4 { + chaos_run(5, wseed, sseed, 4_000); + } + } +} + +#[test] +fn chaos_deterministic() { + for seed in 0 .. 4 { + let first = chaos_run_raw(3, seed, seed, 2_000); + let second = chaos_run_raw(3, seed, seed, 2_000); + assert_eq!(first, second, "same seed, different execution"); + } +} + +/// A larger grind for manual exploration: `GRIND_RUNS=50000 cargo test --release +/// -p timely --test grind -- --ignored --nocapture`. +#[test] +#[ignore] +fn chaos_big_grind() { + let runs: u64 = std::env::var("GRIND_RUNS").ok().and_then(|s| s.parse().ok()).unwrap_or(10_000); + let mut rng = SplitMix64(0x6E1D); + for run in 0 .. runs { + let peers = 2 + rng.below(4); + let wseed = rng.next(); + let sseed = rng.next(); + let outcome = std::panic::catch_unwind(AssertUnwindSafe(|| { + chaos_run(peers, wseed, sseed, 3_000); + })); + assert!( + outcome.is_ok(), + "run {} failed: reproduce with chaos_run({}, {:#x}, {:#x}, 3_000)", + run, peers, wseed, sseed, + ); + if run % 1_000 == 0 { println!("{}/{} runs clean", run, runs); } + } + println!("{} runs clean", runs); +} diff --git a/timely/tests/simulation.rs b/timely/tests/simulation.rs new file mode 100644 index 000000000..41006439b --- /dev/null +++ b/timely/tests/simulation.rs @@ -0,0 +1,155 @@ +//! Deterministic simulation tests: run a multi-worker barrier under seeded random +//! schedules of worker steps and message deliveries, asserting progress-tracking +//! safety (no round notified twice, rounds in order) on every explored schedule. + +use std::rc::Rc; +use std::cell::RefCell; + +use timely::simulate::{Decision, Simulation}; +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::{Feedback, ConnectLoop}; +use timely::dataflow::operators::generic::operator::Operator; +use timely::container::CapacityContainerBuilder; + +const ROUNDS: usize = 25; + +/// A tiny deterministic RNG (SplitMix64); the harness stays free of external deps. +struct SplitMix64(u64); +impl SplitMix64 { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + fn below(&mut self, bound: usize) -> usize { + (self.next() % (bound as u64)) as usize + } +} + +/// Runs a barrier to completion under a seeded random schedule prefix followed by a +/// fair drain. Returns the sequence of (worker, round) notifications, in the order +/// they occurred across all workers (meaningful: everything is on one thread). +fn barrier_run(peers: usize, seed: u64, prefix: usize) -> Vec<(usize, usize)> { + + let mut sim = Simulation::new(peers); + let log = Rc::new(RefCell::new(Vec::new())); + + for index in 0 .. peers { + let log = Rc::clone(&log); + sim.worker_mut(index).dataflow(move |scope| { + let (handle, stream) = scope.feedback::>(1); + stream.unary_notify::, _, _>( + Pipeline, + "Barrier", + vec![0], + move |_, _, notificator| { + let mut count = 0; + while let Some((cap, _cnt)) = notificator.next() { + count += 1; + let round = *cap.time(); + log.borrow_mut().push((index, round)); + if round + 1 < ROUNDS { + notificator.notify_at(cap.delayed(&(round + 1))); + } + } + // Progress safety: at most one round may be notified per scheduling. + assert!(count <= 1); + } + ) + .connect_loop(handle); + }); + } + + // A seeded random prefix of step/deliver decisions ... + let mut rng = SplitMix64(seed); + for _ in 0 .. prefix { + let decision = + if rng.next() % 2 == 0 { + Decision::Step(rng.below(peers)) + } + else { + Decision::Deliver { + source: rng.below(peers), + target: rng.below(peers), + count: 1 + rng.below(4), + } + }; + sim.apply(decision); + } + + // ... then a fair drain to completion. + assert!(sim.drain(10_000), "simulation failed to quiesce"); + + let log = Rc::try_unwrap(log).expect("operators should be dropped").into_inner(); + + // Progress safety: each worker sees every round exactly once, in order. + for worker in 0 .. peers { + let rounds: Vec = log.iter().filter(|(w, _)| *w == worker).map(|(_, r)| *r).collect(); + assert_eq!(rounds, (0 .. ROUNDS).collect::>(), "worker {}", worker); + } + + log +} + +#[test] +fn barrier_completes_under_random_schedules() { + for seed in 0 .. 32 { + barrier_run(3, seed, 2_000); + } +} + +#[test] +fn barrier_under_wide_schedules() { + for seed in 0 .. 8 { + barrier_run(6, seed, 4_000); + } +} + +#[test] +fn same_seed_same_execution() { + let first = barrier_run(4, 0xDECAF, 3_000); + let second = barrier_run(4, 0xDECAF, 3_000); + assert_eq!(first, second); +} + +#[test] +fn starved_worker_stalls_then_completes() { + // Only ever step worker 0 and deliver into it; nobody else runs until the drain. + let mut sim = Simulation::new(3); + let reached = Rc::new(RefCell::new(vec![0_usize; 3])); + for index in 0 .. 3 { + let reached = Rc::clone(&reached); + sim.worker_mut(index).dataflow(move |scope| { + let (handle, stream) = scope.feedback::>(1); + stream.unary_notify::, _, _>( + Pipeline, + "Barrier", + vec![0], + move |_, _, notificator| { + while let Some((cap, _cnt)) = notificator.next() { + let round = *cap.time(); + reached.borrow_mut()[index] = round; + if round + 1 < ROUNDS { + notificator.notify_at(cap.delayed(&(round + 1))); + } + } + } + ) + .connect_loop(handle); + }); + } + + for _ in 0 .. 1_000 { + sim.apply(Decision::Step(0)); + for source in 0 .. 3 { + sim.apply(Decision::Deliver { source, target: 0, count: usize::MAX }); + } + } + + // Worker 0 cannot pass round 0: peers have neither run nor confirmed progress. + assert_eq!(reached.borrow()[0], 0); + assert!(sim.drain(10_000), "simulation failed to quiesce"); + assert_eq!(*reached.borrow(), vec![ROUNDS - 1; 3]); +}