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; 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]); +}