diff --git a/communication/src/allocator/generic.rs b/communication/src/allocator/generic.rs index a3f68b12e..36048c8de 100644 --- a/communication/src/allocator/generic.rs +++ b/communication/src/allocator/generic.rs @@ -9,6 +9,7 @@ use std::cell::RefCell; use crate::allocator::thread::ThreadBuilder; use crate::allocator::{Allocate, AllocateBuilder, Exchangeable, Thread, Process, ProcessBuilder}; use crate::allocator::zero_copy::allocator::{TcpBuilder, TcpAllocator}; +use crate::allocator::simulation::SimAllocator; use crate::{Push, Pull}; @@ -21,6 +22,8 @@ pub enum Allocator { Process(Process), /// Inter-process allocator (TCP-based, with a `Process` as its intra-process inner). Tcp(TcpAllocator), + /// Single-threaded simulation allocator, with driver-controlled message delivery. + Sim(SimAllocator), } impl Allocator { @@ -30,6 +33,7 @@ impl Allocator { Allocator::Thread(t) => t.index(), Allocator::Process(p) => p.index(), Allocator::Tcp(z) => z.index(), + Allocator::Sim(s) => s.index(), } } /// The number of workers. @@ -38,6 +42,7 @@ impl Allocator { Allocator::Thread(t) => t.peers(), Allocator::Process(p) => p.peers(), Allocator::Tcp(z) => z.peers(), + Allocator::Sim(s) => s.peers(), } } /// Constructs several send endpoints and one receive endpoint. @@ -46,6 +51,7 @@ impl Allocator { Allocator::Thread(t) => t.allocate(identifier), Allocator::Process(p) => p.allocate(identifier), Allocator::Tcp(z) => z.allocate(identifier), + Allocator::Sim(s) => s.allocate(identifier), } } /// Constructs several send endpoints and one receive endpoint. @@ -54,6 +60,7 @@ impl Allocator { Allocator::Thread(t) => t.broadcast(identifier), Allocator::Process(p) => p.broadcast(identifier), Allocator::Tcp(z) => z.broadcast(identifier), + Allocator::Sim(s) => s.broadcast(identifier), } } /// Perform work before scheduling operators. @@ -62,6 +69,7 @@ impl Allocator { Allocator::Thread(t) => t.receive(), Allocator::Process(p) => p.receive(), Allocator::Tcp(z) => z.receive(), + Allocator::Sim(s) => s.receive(), } } /// Perform work after scheduling operators. @@ -70,6 +78,7 @@ impl Allocator { Allocator::Thread(t) => t.release(), Allocator::Process(p) => p.release(), Allocator::Tcp(z) => z.release(), + Allocator::Sim(s) => s.release(), } } /// Provides access to the shared event queue. @@ -78,6 +87,7 @@ impl Allocator { Allocator::Thread(ref t) => t.events(), Allocator::Process(ref p) => p.events(), Allocator::Tcp(ref z) => z.events(), + Allocator::Sim(ref s) => s.events(), } } @@ -87,6 +97,7 @@ impl Allocator { Allocator::Thread(t) => t.await_events(duration), Allocator::Process(p) => p.await_events(duration), Allocator::Tcp(z) => z.await_events(duration), + Allocator::Sim(s) => s.await_events(duration), } } @@ -119,6 +130,7 @@ impl Allocate for Allocator { Allocator::Thread(t) => t.await_events(_duration), Allocator::Process(p) => p.await_events(_duration), Allocator::Tcp(z) => z.await_events(_duration), + Allocator::Sim(s) => s.await_events(_duration), } } } diff --git a/communication/src/allocator/mod.rs b/communication/src/allocator/mod.rs index 8c7cf6893..2075f9500 100644 --- a/communication/src/allocator/mod.rs +++ b/communication/src/allocator/mod.rs @@ -13,6 +13,7 @@ pub mod generic; pub mod canary; pub mod counters; +pub mod simulation; pub mod zero_copy; diff --git a/communication/src/allocator/simulation.rs b/communication/src/allocator/simulation.rs new file mode 100644 index 000000000..8fef2b8fc --- /dev/null +++ b/communication/src/allocator/simulation.rs @@ -0,0 +1,319 @@ +//! An allocator for deterministic simulation of multi-worker computation. +//! +//! All workers live on one thread, and all inter-worker communication is held in +//! per-`(source, target)` FIFO streams whose delivery is controlled by a [`SimNet`] +//! handle. A message becomes visible to its recipient only once the driver releases +//! it, so the complete execution is a pure function of the sequence of "step worker w" +//! and "release k messages from s to t" decisions the driver makes. +//! +//! The allocator mirrors [`ProcessAllocator`](crate::allocator::zero_copy::allocator_process::ProcessAllocator): +//! messages are serialized through the same `Bytesable` framing, so simulated runs also +//! exercise the byte-oriented data path. Unlike `ProcessAllocator`, the streams are plain +//! `Rc>` state rather than `MergeQueue`s: there is no concurrency to defend +//! against, and no thread to awaken. + +use std::rc::Rc; +use std::cell::RefCell; +use std::collections::{VecDeque, HashMap, hash_map::Entry}; + +use timely_bytes::arc::Bytes; + +use crate::networking::MessageHeader; + +use crate::{Allocate, Push, Pull}; +use crate::allocator::Exchangeable; +use crate::allocator::canary::Canary; +use crate::allocator::zero_copy::bytes_exchange::{BytesPush, SendEndpoint}; +use crate::allocator::zero_copy::bytes_slab::BytesRefill; +use crate::allocator::zero_copy::push_pull::{Pusher, Puller}; + +/// A directed stream of framed messages from one worker to another. +/// +/// Bytes pushed by the sender accumulate in `pending`, one framed message per entry, +/// and move to `released` only when the simulation driver says so. The receiving +/// allocator drains `released` in its `receive()`. +struct SimStream { + pending: VecDeque, + released: VecDeque, +} + +/// A driver's handle to all inter-worker streams of a simulated computation. +/// +/// Constructed by [`SimNet::new_vector`], which also produces one [`SimAllocator`] +/// per worker. The handle controls delivery: nothing a worker sends is visible to +/// its recipient until the driver releases it through this handle. Per-stream FIFO +/// order is maintained; the driver chooses only the interleaving across streams, +/// which is exactly the schedule space a real transport (TCP) can produce. +pub struct SimNet { + /// Indexed by `[source][target]`. + streams: Vec>>>, +} + +impl SimNet { + /// Constructs the network and one allocator per worker. + /// + /// The allocators contain `Rc` state shared with the returned handle, and so must + /// all be used from the thread that calls this method. + pub fn new_vector(peers: usize, refill: BytesRefill) -> (SimNet, Vec) { + + let streams: Vec> = (0 .. peers) + .map(|_source| (0 .. peers) + .map(|_target| Rc::new(RefCell::new(SimStream { + pending: VecDeque::new(), + released: VecDeque::new(), + }))) + .collect()) + .collect(); + + let allocators = (0 .. peers) + .map(|index| { + let sends = (0 .. peers) + .map(|target| { + let sender = SimSender { + stream: Rc::clone(&streams[index][target]), + immediate: index == target, + }; + Rc::new(RefCell::new(SendEndpoint::new(sender, refill.clone()))) + }) + .collect(); + let recvs = (0 .. peers) + .map(|source| Rc::clone(&streams[source][index])) + .collect(); + SimAllocator { + index, + peers, + events: Rc::new(RefCell::new(Vec::new())), + canaries: Rc::new(RefCell::new(Vec::new())), + channel_id_bound: None, + staged: Vec::new(), + sends, + recvs, + to_local: HashMap::new(), + } + }) + .collect(); + + (SimNet { streams }, allocators) + } + + /// The number of simulated workers. + pub fn peers(&self) -> usize { self.streams.len() } + + /// The number of undelivered messages on the stream from `source` to `target`. + pub fn pending(&self, source: usize, target: usize) -> usize { + self.streams[source][target].borrow().pending.len() + } + + /// Releases up to `count` pending messages from `source` to `target`, in FIFO order. + /// + /// Returns the number of messages released, which may be less than `count` if fewer + /// are pending. This clamping keeps any decision sequence valid, which matters when + /// shrinking failing schedules. + pub fn release(&mut self, source: usize, target: usize, count: usize) -> usize { + let mut stream = self.streams[source][target].borrow_mut(); + let stream = &mut *stream; + let count = std::cmp::min(count, stream.pending.len()); + stream.released.extend(stream.pending.drain(.. count)); + count + } + + /// Fault injection: moves the front pending message from `source` to `target` + /// to the back of the pending queue, deferring it behind everything currently + /// pending. Combined with ordinary releases this reaches arbitrary per-stream + /// permutations, violating FIFO. No-op with fewer than two messages pending; + /// returns the number deferred. + /// + /// This produces deliveries no real transport can produce. It exists so that + /// test oracles can be probed against actual protocol violations, not to + /// explore legal behavior. + pub fn hold_back(&mut self, source: usize, target: usize) -> usize { + let mut stream = self.streams[source][target].borrow_mut(); + let stream = &mut *stream; + if stream.pending.len() >= 2 { + let first = stream.pending.pop_front().expect("len checked"); + stream.pending.push_back(first); + 1 + } + else { 0 } + } + + /// Fault injection: discards the front pending message from `source` to + /// `target`, modeling message loss. Returns the number discarded. + /// + /// No real (TCP-backed) deployment loses messages without tearing down the + /// computation; this exists to confirm that the test oracles detect loss. + pub fn drop_front(&mut self, source: usize, target: usize) -> usize { + let mut stream = self.streams[source][target].borrow_mut(); + if stream.pending.pop_front().is_some() { 1 } else { 0 } + } + + /// Releases all pending messages on all streams, and returns the number released. + pub fn release_all(&mut self) -> usize { + let mut total = 0; + for source in 0 .. self.peers() { + for target in 0 .. self.peers() { + total += self.release(source, target, usize::MAX); + } + } + total + } +} + +/// The sending half of a stream: splits pushed byte blocks into framed messages. +/// +/// Splitting at push time gives the driver per-message delivery granularity, even +/// though `SendEndpoint` may hand over blocks containing several messages. +struct SimSender { + stream: Rc>, + /// Set for a worker's stream to itself, whose messages skip `pending`. + /// + /// Real allocators surface self-sent messages at the next `receive()`; gating + /// them would admit schedules no actual deployment can produce. + immediate: bool, +} + +impl BytesPush for SimSender { + fn extend>(&mut self, iter: I) { + let mut stream = self.stream.borrow_mut(); + for mut bytes in iter { + while !bytes.is_empty() { + let header = MessageHeader::try_read(&bytes[..]) + .expect("simulation: sent bytes must contain whole framed messages"); + let message = bytes.extract_to(header.required_bytes()); + if self.immediate { + stream.released.push_back(message); + } else { + stream.pending.push_back(message); + } + } + } + } +} + +/// A serializing allocator whose message delivery is driven by a [`SimNet`]. +pub struct SimAllocator { + + index: usize, // number out of peers + peers: usize, // number of peer allocators. + + events: Rc>>, + + canaries: Rc>>, + + channel_id_bound: Option, + + // sending, receiving, and responding to binary buffers. + staged: Vec, + sends: Vec>>>, // sends[x] -> goes to worker x. + recvs: Vec>>, // recvs[x] <- from worker x. + to_local: HashMap>>>, // to worker-local typed pullers. +} + +impl Allocate for SimAllocator { + fn index(&self) -> usize { self.index } + fn peers(&self) -> usize { self.peers } + fn allocate(&mut self, identifier: usize) -> (Vec>>, Box>) { + + // Assume and enforce in-order identifier allocation. + if let Some(bound) = self.channel_id_bound { + assert!(bound < identifier); + } + self.channel_id_bound = Some(identifier); + + let mut pushes = Vec::>>::with_capacity(self.peers()); + + for target_index in 0 .. self.peers() { + + // message header template. + let header = MessageHeader { + channel: identifier, + source: self.index, + target_lower: target_index, + target_upper: target_index+1, + length: 0, + seqno: 0, + }; + + pushes.push(Box::new(Pusher::new(header, Rc::clone(&self.sends[target_index])))); + } + + let channel = Rc::clone(self.to_local.entry(identifier).or_default()); + + use crate::allocator::counters::Puller as CountPuller; + let canary = Canary::new(identifier, Rc::clone(&self.canaries)); + let puller = Box::new(CountPuller::new(Puller::new(channel, canary), identifier, Rc::clone(self.events()))); + + (pushes, puller) + } + + // Surface released binary buffers as typed channel messages. + fn receive(&mut self) { + + // Check for channels whose `Puller` has been dropped. + let mut canaries = self.canaries.borrow_mut(); + for dropped_channel in canaries.drain(..) { + let _dropped = + self.to_local + .remove(&dropped_channel) + .expect("non-existent channel dropped"); + } + std::mem::drop(canaries); + + let mut events = self.events.borrow_mut(); + + for recv in self.recvs.iter() { + self.staged.extend(recv.borrow_mut().released.drain(..)); + } + + for mut bytes in self.staged.drain(..) { + + // We expect that `bytes` contains an integral number of messages. + // No splitting occurs across allocations. + while !bytes.is_empty() { + + if let Some(header) = MessageHeader::try_read(&bytes[..]) { + + // Get the header and payload, ditch the header. + let mut peel = bytes.extract_to(header.required_bytes()); + let _ = peel.extract_to(header.header_bytes()); + + // Increment message count for channel. + // Safe to do this even if the channel has been dropped. + events.push(header.channel); + + // Ensure that a queue exists. + match self.to_local.entry(header.channel) { + Entry::Vacant(entry) => { + // We may receive data before allocating, and shouldn't block. + if self.channel_id_bound.map(|b| b < header.channel).unwrap_or(true) { + entry.insert(Rc::new(RefCell::new(VecDeque::new()))) + .borrow_mut() + .push_back(peel); + } + } + Entry::Occupied(mut entry) => { + entry.get_mut().borrow_mut().push_back(peel); + } + } + } + else { + panic!("simulation: failed to read full header!"); + } + } + } + } + + // Publish un-full binary buffers into the pending streams. + fn release(&mut self) { + for send in self.sends.iter_mut() { + send.borrow_mut().publish(); + } + } + + fn events(&self) -> &Rc>> { + &self.events + } + fn await_events(&self, _duration: Option) { + // Never parks: progress is driven by the simulation driver, not by other threads. + } +} 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..a2c057906 --- /dev/null +++ b/timely/src/simulate.rs @@ -0,0 +1,143 @@ +//! Deterministic simulation of multi-worker timely computations. +//! +//! A [`Simulation`] hosts several workers on the calling thread, communicating through +//! streams whose delivery is controlled by the caller. Each worker is deterministic given +//! the messages delivered to it, and delivery is constrained only by per-stream 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::sync::Arc; + +use crate::WorkerConfig; +use crate::communication::Allocator; +use crate::communication::allocator::simulation::SimNet; +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 { + net: SimNet, + 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 (net, allocators) = SimNet::new_vector(peers, refill); + let workers = allocators + .into_iter() + .map(|allocator| Worker::new(WorkerConfig::default(), Allocator::Sim(allocator), None)) + .collect(); + Simulation { net, 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 number of undelivered messages from `source` to `target`. + pub fn pending(&self, source: usize, target: usize) -> usize { + self.net.pending(source, target) + } + + /// 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.net.release(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.net.release(source, target, count) + } + + /// Fault injection: defers the next message from `source` to `target` behind all + /// currently pending messages, violating per-stream FIFO. See [`SimNet::hold_back`]. + pub fn inject_hold_back(&mut self, source: usize, target: usize) -> usize { + self.net.hold_back(source, target) + } + + /// Fault injection: discards the next message from `source` to `target`, modeling + /// message loss. See [`SimNet::drop_front`]. + pub fn inject_drop(&mut self, source: usize, target: usize) -> usize { + self.net.drop_front(source, target) + } + + /// 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 = self.net.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..fee4094a6 --- /dev/null +++ b/timely/tests/grind.rs @@ -0,0 +1,336 @@ +//! 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. +//! +//! The `oracle_detects_fifo_violation` test validates the oracle pipeline by injecting +//! genuine protocol violations (swapped in-stream deliveries) and confirming detection. + +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); +} + +/// Runs a fault-injecting schedule; returns (injections performed, violation detected). +fn faulty_run(peers: usize, wseed: u64, sseed: u64, fault: impl Fn(&mut Simulation, usize, usize) -> usize) -> (usize, bool) { + let mut injected = 0; + let outcome = std::panic::catch_unwind(AssertUnwindSafe(|| { + let mut sim = Simulation::new(peers); + let results = build_chaos(&mut sim, peers, wseed); + let mut rng = SplitMix64(sseed); + for _ in 0 .. 3_000 { + match rng.next() % 4 { + 0 | 1 => sim.apply(Decision::Step(rng.below(peers))), + 2 => { + let (s, t) = (rng.below(peers), rng.below(peers)); + sim.apply(Decision::Deliver { source: s, target: t, count: 1 + rng.below(4) }); + } + _ => { injected += fault(&mut sim, rng.below(peers), rng.below(peers)); } + } + } + assert!(sim.drain(50_000), "simulation failed to quiesce"); + let mut log = Rc::try_unwrap(results).expect("operators should be dropped").into_inner(); + log.sort(); + assert_eq!(log, expected(peers, wseed), "conservation violated"); + })); + (injected, outcome.is_err()) +} + +/// Validates the oracle pipeline: injected message loss must be detected (by the +/// quiescence or conservation oracle) in every run where a message was dropped. +#[test] +fn oracle_detects_message_loss() { + + // Quiet the expected panics; this test is its own binary and its concurrent + // tests do not intentionally panic. + let prior = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + + let mut detected = 0; + let mut effective = 0; + for sseed in 0 .. 32 { + let (injected, caught) = faulty_run(3, 0, sseed, |sim, s, t| sim.inject_drop(s, t)); + if injected > 0 { + effective += 1; + if caught { detected += 1; } + } + } + + std::panic::set_hook(prior); + + println!("message loss detected in {}/{} affected runs", detected, effective); + assert!(effective > 10, "fault injection ineffective; nothing was tested"); + assert_eq!(detected, effective, "oracles missed injected message loss"); +} + +/// Probes protocol sensitivity to per-stream FIFO violations: defers messages behind +/// their successors and reports how often any oracle notices. This is a measurement, +/// not an assertion: progress accumulation is commutative and batches remain intact, +/// so reordering may be genuinely tolerated. Run manually with `--ignored --nocapture`. +#[test] +#[ignore] +fn probe_fifo_sensitivity() { + + let prior = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + + let mut detected = 0; + let mut effective = 0; + let attempts = std::env::var("GRIND_RUNS").ok().and_then(|s| s.parse().ok()).unwrap_or(1_000); + for sseed in 0 .. attempts { + let (injected, caught) = faulty_run(2 + (sseed as usize) % 3, sseed, !sseed, |sim, s, t| sim.inject_hold_back(s, t)); + if injected > 0 { + effective += 1; + if caught { detected += 1; } + } + } + + std::panic::set_hook(prior); + println!("fifo violations detected in {}/{} affected runs", detected, effective); +} 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]); +}