diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs index 9c72ffeab..e247fc44a 100644 --- a/differential-dataflow/src/operators/int_proxy/mod.rs +++ b/differential-dataflow/src/operators/int_proxy/mod.rs @@ -46,6 +46,7 @@ mod history; pub mod join; pub mod reduce; +pub mod vec_backend; /// Integer-only exchange medium: a consolidated collection of `[((hash, id), time, diff)]`. /// diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs new file mode 100644 index 000000000..426391864 --- /dev/null +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -0,0 +1,443 @@ +//! A reference [`ProxyReduceBackend`] over [`VecChunk`] storage. +//! +//! This backend exists to demonstrate the backend recipe in plain `Vec`-of-rows form, and to give +//! the proxy tactic a counterpart that can be tested and benchmarked against the cursor tactic on +//! identical storage: both can drive a reduction over the same hash-keyed +//! [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine) arrangement (see `tests/int_proxy.rs` and +//! `tests/int_proxy_bench.rs`). +//! +//! # The recipe +//! +//! The wiring arranges `coll.map(|(k, v)| (k.hashed(), (k, v)))` into a `ChunkSpine` whose key is +//! the `u64` key hash and whose value is the full `(key, val)` pair. +//! +//! Per window, the backend presents three proxy bridges — novel input, prior input, and prior output. +//! The identifiers chosen are based on ordinal position. Unfortunately, they are chosen independently +//! for the novel and prior inputs, meaning that empty input collections may be supplied to the logic +//! for evaluation, where the logic should be ignored and zero -> zero enforced. This could be fixed +//! with a more attentive identifier selection. +//! +//! The backend crawls all keys at once, rather than respecting the window setting. +//! This is a defect that should be fixed, but the backend shouldn't be used at scale. +//! The backend clones keys and values like it is being paid to waste cycles. +//! Generally, this is not a high-performance backend, but shouldn't be abysmal. + +use std::collections::{BTreeMap, HashMap}; +use std::rc::Rc; + +use timely::container::PushInto; +use timely::progress::Timestamp; + +use crate::consolidation::{consolidate, consolidate_updates_from}; +use crate::difference::Semigroup; +use crate::lattice::Lattice; +use crate::trace::chunk::ChunkBatch; +use crate::trace::chunk::vec::VecChunk; +use crate::trace::Description; + +use super::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; + +/// The batch type of a hash-keyed [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine): payload +/// `D` is `(K, V)` on the input side and `(K, W)` on the output side. +type VBatch = Rc>>; + +/// A reference [`ProxyReduceBackend`] over [`VBatch`] storage. +pub struct VecReduceBackend { + /// User supplied reduce closure. + logic: L, + /// Configuration: keys per window, to size the steps the backend performs. + window_size: usize, + + /// All active keys, either novel input or supplied as externally changed. + /// This is *not* windowed, which is a defect to fix. + keys_cache: Vec, + /// A state bit indicating that the keys cache should be rebuild (each begin). + keys_stale: bool, + + /// Resolves an input value id (a per-window ordinal) to its `(key, val)` row. + in_pool: Vec<(K, V)>, + /// Resolves an output value id to its `(key, out)` row, for the current window only. + out_pool: Vec<(K, W)>, + /// Interns `(key, out)` rows to their output id, for the current window only. + /// Lookup-only: the non-determinism of the map's iteration order is never observed. + out_ids: HashMap<(K, W), u64>, + + /// The retire's output tile descriptions, and the rows accumulated for each. + tiles: Vec>, + tile_rows: Vec>, +} + +impl VecReduceBackend { + /// A backend deferring value semantics to `logic`, covering the key space in windows. + pub fn new(logic: L) -> Self { Self::with_window(logic, 1 << 12) } + + /// A backend with an explicit window size, in keys. + pub fn with_window(logic: L, window_size: usize) -> Self { + VecReduceBackend { + logic, + window_size: window_size.max(1), + keys_cache: Vec::new(), + keys_stale: true, + in_pool: Vec::new(), + out_pool: Vec::new(), + out_ids: HashMap::new(), + tiles: Vec::new(), + tile_rows: Vec::new(), + } + } +} + +impl ProxyReduceBackend, VBatch<(K, W), T, R>> + for VecReduceBackend +where + K: Ord + Clone + std::hash::Hash + 'static, + V: Ord + Clone + 'static, + W: Ord + Clone + std::hash::Hash + 'static, + T: Lattice + Timestamp, + R: Semigroup + Ord + Clone + 'static, + L: FnMut(&K, &[(V, R)], &mut Vec<(W, R)>, &mut Vec<(W, R)>), +{ + type RIn = R; + type ROut = R; + + fn begin(&mut self, tiles: &[Description]) { + self.keys_stale = true; + self.tiles = tiles.to_vec(); + self.tile_rows = (0..tiles.len()).map(|_| Vec::new()).collect(); + } + + #[inline(never)] + fn next_window( + &mut self, + instance: &ReduceInstance<'_, VBatch<(K, V), T, R>, VBatch<(K, W), T, R>>, + changed: &[u64], + from: &mut Option, + window: &mut ReduceWindow, + ) { + let Some(start) = *from else { return }; + + // If the first window: form a list of all active keys, novel or changed. + // TODO: this is wasteful; the in-order chunk keys could be merged instead. + if self.keys_stale { + self.keys_stale = false; + self.keys_cache.clear(); + for batch in instance.input_batches.iter() { + for chunk in batch.chunks.iter() { + self.keys_cache.extend(chunk.as_slice().iter().map(|r| r.0.0)); + } + } + self.keys_cache.sort_unstable(); + self.keys_cache.dedup(); + if !changed.is_empty() { + let mut merged = Vec::with_capacity(self.keys_cache.len() + changed.len()); + let (mut a, mut b) = (0usize, 0usize); + while a < self.keys_cache.len() || b < changed.len() { + let key = match (self.keys_cache.get(a), changed.get(b)) { + (Some(x), Some(y)) => *x.min(y), + (Some(x), None) => *x, + (None, Some(y)) => *y, + (None, None) => unreachable!("loop condition ensures one is present"), + }; + if self.keys_cache.get(a) == Some(&key) { a += 1; } + if changed.get(b) == Some(&key) { b += 1; } + merged.push(key); + } + self.keys_cache = merged; + } + } + + // Determine the range of active keys to process in this window. + let lo = self.keys_cache.partition_point(|k| *k < start); + if lo == self.keys_cache.len() { + *from = None; + return; + } + let hi = (lo + self.window_size).min(self.keys_cache.len()); + *from = if hi == self.keys_cache.len() { None } else { Some(self.keys_cache[hi]) }; + let keys = &self.keys_cache[lo..hi]; + + // The two input runs, presented apart on ordinal ids from a shared pool. Ids ascend with + // the walk, so each bridge emerges sorted by `((hash, id), time)` with nothing to sort or + // consolidate; a value in both runs gets two ids, reconciled by value in the corrections. + self.in_pool.clear(); + let pool = &mut self.in_pool; + let mut last: Option = None; + merged_run(instance.source_batches, keys, |hash, data, time, diff| { + if last != Some(hash) || pool.last() != Some(data) { + pool.push(data.clone()); + last = Some(hash); + } + window.history.push(((hash, (pool.len() - 1) as u64), time.clone(), diff.clone())); + }); + let mut last: Option = None; + merged_run(instance.input_batches, keys, |hash, data, time, diff| { + if last != Some(hash) || pool.last() != Some(data) { + pool.push(data.clone()); + last = Some(hash); + } + window.novel.push(((hash, (pool.len() - 1) as u64), time.clone(), diff.clone())); + }); + + // The output history, interned into the id namespace corrections mint into. + self.out_ids.clear(); + self.out_pool.clear(); + let (out_pool, out_ids) = (&mut self.out_pool, &mut self.out_ids); + merged_run(instance.output_batches, keys, |hash, data, time, diff| { + let id = *out_ids.entry(data.clone()).or_insert_with(|| { + out_pool.push(data.clone()); + (out_pool.len() - 1) as u64 + }); + window.output.push(((hash, id), time.clone(), diff.clone())); + }); + } + + #[inline(never)] + fn reduce_corrections( + &mut self, + keys: &[u64], + in_ends: &[usize], + input: &[(u64, R)], + out_ends: &[usize], + output: &[(u64, R)], + ) -> (Vec<(u64, R)>, Vec) { + let mut corr: Vec<(u64, R)> = Vec::new(); + let mut corr_ends: Vec = Vec::with_capacity(keys.len()); + let (mut is, mut os) = (0usize, 0usize); + let mut updates: Vec<(W, R)> = Vec::new(); + let mut input_vals: Vec<(V, R)> = Vec::new(); + let mut current: Vec<(W, R)> = Vec::new(); + for i in 0..keys.len() { + let (ie, oe) = (in_ends[i], out_ends[i]); + // No-collision fast path: when the whole bracket is one real key, the per-key grouping + // below can be skipped. Testing only the bracket's ENDPOINTS would be wrong: neither id + // space is key-ordered across a bracket. Input ids are ordinals minted history-run + // first and then novel, so the order is `history by key, then novel by key`; output ids + // are interned, with corrections appended after the presentation. Either can read + // `[A, B, A]`, whose endpoints agree while its interior does not. So resolve every id + // and require them all to agree — a linear pass over data the fast path is about to + // clone anyway, against the general path's per-key maps. + let single_key: Option = { + let mut only: Option<&K> = None; + let mut agree = true; + for (vid, _) in &input[is..ie] { + let k = &self.in_pool[*vid as usize].0; + match only { + None => only = Some(k), + Some(prev) if prev == k => {} + Some(_) => { agree = false; break } + } + } + if agree { + for (vid, _) in &output[os..oe] { + let k = &self.out_pool[*vid as usize].0; + match only { + None => only = Some(k), + Some(prev) if prev == k => {} + Some(_) => { agree = false; break } + } + } + } + if agree { only.cloned() } else { None } + }; + if let Some(key) = single_key { + input_vals.clear(); + input_vals.extend(input[is..ie].iter().map(|(vid, d)| (self.in_pool[*vid as usize].1.clone(), d.clone()))); + consolidate(&mut input_vals); + current.clear(); + current.extend(output[os..oe].iter().map(|(vid, d)| (self.out_pool[*vid as usize].1.clone(), d.clone()))); + consolidate(&mut current); + updates.clear(); + (self.logic)(&key, &input_vals, &mut current, &mut updates); + consolidate(&mut updates); + for (w, d) in updates.drain(..) { + let key_w = (key.clone(), w); + let id = *self.out_ids.entry(key_w.clone()).or_insert_with(|| { + self.out_pool.push(key_w); + (self.out_pool.len() - 1) as u64 + }); + corr.push((id, d)); + } + } else { + let mut ins: BTreeMap> = BTreeMap::new(); + for (vid, d) in &input[is..ie] { + let (k, v) = &self.in_pool[*vid as usize]; + ins.entry(k.clone()).or_default().push((v.clone(), d.clone())); + } + let mut outs: BTreeMap> = BTreeMap::new(); + for (vid, d) in &output[os..oe] { + let (k, w) = &self.out_pool[*vid as usize]; + outs.entry(k.clone()).or_default().push((w.clone(), d.clone())); + } + let mut real_keys: Vec = ins.keys().chain(outs.keys()).cloned().collect(); + real_keys.sort(); + real_keys.dedup(); + for key in real_keys { + let mut ivals = ins.remove(&key).unwrap_or_default(); + consolidate(&mut ivals); + let mut cur = outs.remove(&key).unwrap_or_default(); + consolidate(&mut cur); + updates.clear(); + (self.logic)(&key, &ivals, &mut cur, &mut updates); + consolidate(&mut updates); + for (w, d) in updates.drain(..) { + let key_w = (key.clone(), w); + let id = *self.out_ids.entry(key_w.clone()).or_insert_with(|| { + self.out_pool.push(key_w); + (self.out_pool.len() - 1) as u64 + }); + corr.push((id, d)); + } + } + } + corr_ends.push(corr.len()); + is = ie; + os = oe; + } + (corr, corr_ends) + } + + #[inline(never)] + fn emit(&mut self, tile: usize, records: &[((u64, u64), T, R)]) { + let mark = self.tile_rows[tile].len(); + for ((h, vid), t, d) in records { + let row = self.out_pool[*vid as usize].clone(); + self.tile_rows[tile].push(((*h, row), t.clone(), d.clone())); + } + consolidate_updates_from(&mut self.tile_rows[tile], mark); + } + + #[inline(never)] + fn finish(&mut self) -> Vec> { + self.in_pool.clear(); + self.out_pool.clear(); + self.out_ids.clear(); + let tiles = std::mem::take(&mut self.tiles); + let tile_rows = std::mem::take(&mut self.tile_rows); + tiles + .into_iter() + .zip(tile_rows) + .map(|(desc, rows)| { + let mut chunks: Vec> = Vec::default(); + let mut iter = rows.into_iter(); + while iter.len() > 0 { + let mut chunk = VecChunk::default(); + for update in (&mut iter).take( as crate::trace::chunk::Chunk>::TARGET) { + chunk.push_into(update); + } + chunks.push(chunk); + } + Rc::new(ChunkBatch::new(chunks, desc)) + }) + .collect() + } +} + +/// Walks `batches` restricted to the ascending `keys`, emitting each record as +/// `(hash, &payload, &time, &diff)` in `(hash, payload, time)` order. +/// +/// Per hash bracket, the batches' contiguous runs are gathered and sorted by `(payload, time)`, so +/// equal payloads meet across batches and each payload's times arrive in order (batch intervals +/// are disjoint, so cross-batch times need ordering but never summing). The walk seeks to the +/// first requested key and stops after the last, so a bounded window pays for its own range. +#[inline(never)] +fn merged_run( + batches: &[VBatch], + keys: &[u64], + mut sink: impl FnMut(u64, &D, &T, &R), +) where + D: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Semigroup + Ord + Clone + 'static, +{ + if keys.is_empty() { + return; + } + let n = batches.len(); + let (mut ci, mut oi) = (vec![0usize; n], vec![0usize; n]); + let mut cur: Vec<&[((u64, D), T, R)]> = vec![&[]; n]; + // Seek each batch to the first record with hash at or above the window's first key. + let k0 = keys[0]; + for b in 0..n { + let chunks = &batches[b].chunks; + ci[b] = chunks.partition_point(|c| c.as_slice().last().is_some_and(|r| r.0.0 < k0)); + cur[b] = chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); + oi[b] = cur[b].partition_point(|r| r.0.0 < k0); + } + let mut scratch: Vec<(&D, &T, &R)> = Vec::new(); + let mut ki = 0usize; + loop { + // Least hash among the batch heads. + let mut minh: Option = None; + for b in 0..n { + if let Some(r) = cur[b].get(oi[b]) { + if minh.is_none_or(|m| r.0.0 < m) { + minh = Some(r.0.0); + } + } + } + let Some(h) = minh else { break }; + while ki < keys.len() && keys[ki] < h { + ki += 1; + } + if ki >= keys.len() { + break; + } + if keys[ki] != h { + // `h` is not wanted. Seeking every batch to the next key that IS wanted costs a binary + // search per batch and lands at or above it, so at most one unwanted hash is visited + // per wanted key. Walking `h`'s records instead would make the whole pass linear in the + // accumulated history rather than in what is asked for — which on an iterative + // computation, where a retire asks for a handful of scattered keys, is the difference + // between presenting the window and re-reading the trace. When the key set is dense + // this branch is simply never taken, so there is no threshold to tune. + let next = keys[ki]; + for b in 0..n { + loop { + let Some(chunk) = batches[b].chunks.get(ci[b]) else { cur[b] = &[]; break }; + let slice = chunk.as_slice(); + if slice.last().is_some_and(|r| r.0.0 < next) { + ci[b] += 1; + oi[b] = 0; + continue; + } + cur[b] = slice; + oi[b] += slice[oi[b]..].partition_point(|r| r.0.0 < next); + // Chunks are non-empty (`ChunkBatch::new` asserts it) and the guard above + // skipped any whose last key is below `next`, so this chunk holds a record at + // or above `next` and the search cannot land at the end. The branch below is + // unreachable; it is kept so a violated invariant degrades to a slower walk + // rather than to a batch that silently reads as drained. + debug_assert!(oi[b] < slice.len(), "a chunk kept by the skip guard must hold a record at or above the sought key"); + if oi[b] >= slice.len() { + ci[b] += 1; + oi[b] = 0; + continue; + } + break; + } + } + continue; + } + scratch.clear(); + for b in 0..n { + while let Some(r) = cur[b].get(oi[b]) { + if r.0.0 != h { + break; + } + scratch.push((&r.0.1, &r.1, &r.2)); + oi[b] += 1; + if oi[b] >= cur[b].len() { + ci[b] += 1; + oi[b] = 0; + cur[b] = batches[b].chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); + } + } + } + { + scratch.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1))); + for (d, t, r) in scratch.drain(..) { + sink(h, d, t, r); + } + } + } +} diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs index 1a6b9aff8..18dc05d78 100644 --- a/differential-dataflow/src/trace/chunk/vec.rs +++ b/differential-dataflow/src/trace/chunk/vec.rs @@ -37,6 +37,11 @@ const TARGET: usize = 8192; /// A sorted, consolidated run of `((key, val), time, diff)`, shared via `Rc`. pub struct VecChunk(Rc>); +impl VecChunk { + /// The chunk's records as a sorted, consolidated slice. + pub fn as_slice(&self) -> &[((K, V), T, R)] { &self.0 } +} + impl Clone for VecChunk { fn clone(&self) -> Self { VecChunk(Rc::clone(&self.0)) } } diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs new file mode 100644 index 000000000..37836c677 --- /dev/null +++ b/differential-dataflow/tests/int_proxy.rs @@ -0,0 +1,343 @@ +//! Tests for the proxy reduce tactic over the [`VecReduceBackend`] reference backend. +//! +//! The backend reads a HASH-KEYED arrangement (`VecChunk`, key = hash(K)), so +//! the caller maps `(k, v) -> (hash(k), (k, v))` before arranging. Every dataflow test compares +//! against the cursor-tactic reduce on the same input; run in debug so the tactic's +//! `debug_assert_sorted_bridge` and window-contract assertions are live. + +use std::rc::Rc; +use std::sync::{Arc, Mutex}; + +use timely::container::PushInto; +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::ToStream; +use timely::order::Product; +use timely::progress::{Antichain, Timestamp}; + +use differential_dataflow::consolidation::consolidate_updates; +use differential_dataflow::difference::Semigroup; +use differential_dataflow::hashable::Hashable; +use differential_dataflow::lattice::Lattice; +use differential_dataflow::operators::arrange::arrangement::arrange_core; +use differential_dataflow::operators::int_proxy::reduce::ProxyReduceTactic; +use differential_dataflow::operators::int_proxy::vec_backend::VecReduceBackend; +use differential_dataflow::operators::iterate::Iterate; +use differential_dataflow::operators::reduce::{reduce_with_tactic, ReduceTactic}; +use differential_dataflow::trace::chunk::vec::{ + ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine, VecChunk, +}; +use differential_dataflow::trace::chunk::ChunkBatch; +use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::implementations::ContainerChunker; +use differential_dataflow::trace::{Description, Navigable}; +use differential_dataflow::AsCollection; + +type Batch = Rc>>; + +/// Read a `u64`-keyed batch, dropping the hash key; returns `(value, time, diff)`. +fn hread(batches: &[Batch]) -> Vec<(KV, T, R)> +where + KV: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Ord + Semigroup + 'static, +{ + let mut out = Vec::new(); + for b in batches { + for chunk in &b.chunks { + let mut c = chunk.cursor(); + while c.key_valid(chunk) { + while c.val_valid(chunk) { + let kv = c.val(chunk).clone(); + c.map_times(chunk, |t, d| out.push((kv.clone(), t.clone(), d.clone()))); + c.step_val(chunk); + } + c.step_key(chunk); + } + } + } + consolidate_updates(&mut out); + out +} + +/// Build a HASH-KEYED input batch from `((K, V), T, R)` rows: `((hash(K), (K, V)), T, R)`. +fn hbatch(rows: Vec<((K, V), T, R)>, lower: T, upper: T) -> Batch +where + K: Hashable + Ord + Clone + 'static, + K::Output: Into, + V: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Ord + Semigroup + 'static, +{ + let mut hrows: Vec<((u64, (K, V)), T, R)> = + rows.into_iter().map(|((k, v), t, r)| ((k.hashed().into(), (k, v)), t, r)).collect(); + consolidate_updates(&mut hrows); + let mut chunk = VecChunk::default(); + for u in hrows { + chunk.push_into(u); + } + let desc = Description::new(Antichain::from_elem(lower), Antichain::from_elem(upper), Antichain::from_elem(T::minimum())); + Rc::new(ChunkBatch::new(vec![chunk], desc)) +} + +fn max_logic(_k: &u64, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>) { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { + updates.push((*w, -*d)); + } +} + +#[test] +fn reduce_one_retire() { + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(max_logic)); + let input = hbatch::(vec![((7, 3), 0, 1), ((7, 5), 0, 1), ((7, 4), 0, 1), ((9, 2), 0, 1)], 0, 1); + let (produced, frontier) = tactic.retire( + vec![], vec![], vec![input], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + assert!(frontier.is_empty()); + let out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + // Output values are `(key, max)`: (7,5) and (9,2). + assert_eq!(out, vec![((7u64, 5u64), 0u64, 1i64), ((9, 2), 0, 1)]); +} + +#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)] +struct Collide(u64); +/// Every `Collide` hashes alike, so they share a `key_hash` and exercise the collision paths. +/// Written as `Hash` rather than `Hashable` because the latter has a blanket impl for `T: Hash`, +/// and the backend's id interning needs `Hash` too. +impl std::hash::Hash for Collide { + fn hash(&self, state: &mut H) { state.write_u64(0); } +} + +#[test] +fn reduce_collision_correct() { + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new( + |_k: &Collide, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { updates.push((*w, -*d)); } + }, + )); + let input = hbatch::(vec![((Collide(1), 9), 0, 1), ((Collide(1), 4), 0, 1), ((Collide(2), 9), 0, 1)], 0, 1); + let (produced, _f) = tactic.retire( + vec![], vec![], vec![input], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + let out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + assert_eq!(out, vec![((Collide(1), 9u64), 0u64, 1i64), ((Collide(2), 9), 0, 1)], "collision must not merge keys"); +} + +/// Run the proxy reduce (with the given backend window size) and the cursor reduce over the same +/// updates; both outputs must agree. +fn proxy_matches_mainline(updates: Vec<((u64, u64), u64, i64)>, window: usize) { + let t_out = Arc::new(Mutex::new(Vec::<((u64, u64), u64, i64)>::new())); + let r_out = Arc::new(Mutex::new(Vec::<((u64, u64), u64, i64)>::new())); + let (ts, rs) = (t_out.clone(), r_out.clone()); + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let coll = updates.clone().to_stream(scope).as_collection(); + let hashed = coll.clone().map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, ContainerChunker>, VChunkBatcher, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange"); + reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, window))) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + .inspect(move |(d, t, r)| ts.lock().unwrap().push((*d, *t, *r))); + coll.reduce(|_k, input, output| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { output.push((*m, 1)); } + }) + .inspect(move |(d, t, r)| rs.lock().unwrap().push((*d, *t, *r))); + }); + }); + let (mut got, mut want) = (t_out.lock().unwrap().clone(), r_out.lock().unwrap().clone()); + got.sort(); + want.sort(); + assert_eq!(got, want, "proxy-reduce must match the cursor reduce"); +} + +#[test] +fn reduce_dataflow_matches_mainline() { + proxy_matches_mainline( + vec![((7, 3), 0, 1), ((7, 5), 0, 1), ((7, 4), 0, 1), ((9, 2), 0, 1), ((7, 5), 1, -1)], + usize::MAX, + ); +} + +#[test] +fn reduce_multiwindow_matches_mainline() { + // One key per window: every window-loop edge (the `from` cursor, the window-key merge, and the + // in-window `changed` consumption) runs many times, under the harness's contract asserts. + let mut updates = Vec::new(); + for k in 0..64u64 { + updates.push(((k, k), 0u64, 1i64)); + updates.push(((k, 100 + k), 0, 1)); + updates.push(((k, 100 + k), 1, -1)); + } + proxy_matches_mainline(updates, 1); +} + +#[test] +fn reduce_multimoment_matches_mainline() { + // Four distinct times in ONE batch: multi-moment keys exercise the round loop, including + // produced-corrections feedback (the max changes at t1, reverts at t2, changes at t3). + proxy_matches_mainline( + vec![ + ((7, 3), 0, 1), ((7, 5), 1, 1), ((7, 5), 2, -1), ((7, 9), 3, 1), + ((9, 2), 0, 1), ((9, 1), 2, 1), ((9, 2), 3, -1), + ], + usize::MAX, + ); +} + +#[test] +fn reduce_string_values_matches_mainline() { + let t_out = Arc::new(Mutex::new(Vec::<((u64, String), u64, i64)>::new())); + let r_out = Arc::new(Mutex::new(Vec::<((u64, String), u64, i64)>::new())); + let (ts, rs) = (t_out.clone(), r_out.clone()); + let updates: Vec<((u64, String), u64, i64)> = vec![ + ((7, "b".into()), 0, 1), ((7, "d".into()), 0, 1), ((7, "c".into()), 0, 1), ((9, "a".into()), 0, 1), + ((7, "d".into()), 1, -1), ((9, "z".into()), 1, 1), + ]; + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let coll = updates.clone().to_stream(scope).as_collection(); + let hashed = coll.clone().map(|(k, v): (u64, String)| (k.hashed(), (k, v))); + let arr = arrange_core::, ContainerChunker>, VChunkBatcher, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange"); + reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduceStr", ProxyReduceTactic::new(VecReduceBackend::with_window( + |_k: &u64, input: &[(String, i64)], current: &mut Vec<(String, i64)>, updates: &mut Vec<(String, i64)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { updates.push((m, 1)); } + for (w, d) in current.iter() { updates.push((w.clone(), -*d)); } + }, + 1, + ))) + .as_collection(|_h, kw: &(u64, String)| (kw.0, kw.1.clone())) + .inspect(move |(d, t, r)| ts.lock().unwrap().push((d.clone(), *t, *r))); + coll.reduce(|_k, input, output| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| (*v).clone()).max() { output.push((m, 1)); } + }) + .inspect(move |(d, t, r)| rs.lock().unwrap().push((d.clone(), *t, *r))); + }); + }); + let (mut got, mut want) = (t_out.lock().unwrap().clone(), r_out.lock().unwrap().clone()); + got.sort(); + want.sort(); + assert_eq!(got, want, "proxy-reduce must match the cursor reduce for String values"); +} + +#[test] +fn reduce_inside_iterate() { + let out = Arc::new(Mutex::new(Vec::<((u64, u64), u64, i64)>::new())); + let os = out.clone(); + let updates: Vec<((u64, u64), u64, i64)> = + vec![((7, 3), 0, 1), ((7, 5), 0, 1), ((7, 4), 0, 1), ((9, 2), 0, 1), ((9, 8), 0, 1)]; + + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let input = updates.clone().to_stream(scope).as_collection(); + let result = input.iterate(|_scope, inner| { + let hashed = inner.map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, VChunkBatcher, i64>, VChunkBuilder, i64>, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter"); + reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "IterReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + }); + result.inspect(move |(d, t, r)| os.lock().unwrap().push((*d, *t, *r))); + }); + }); + + let mut got = out.lock().unwrap().clone(); + got.sort(); + assert_eq!(got, vec![((7u64, 5u64), 0u64, 1i64), ((9, 8), 0, 1)], "max-per-key fixpoint"); +} + +/// Two colliding keys, but the collision spans the HISTORY/NOVEL boundary: retire 1 inserts both +/// keys, retire 2 touches only the lower one. Input ids are ordinals minted history-first, so the +/// bracket reads `[C1, C2, C1]` by id and its endpoints agree even though its interior does not. +#[test] +fn reduce_collision_across_retires() { + let logic = |_k: &Collide, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { updates.push((*w, -*d)); } + }; + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(logic)); + + // Retire 1: both colliding keys arrive. + let b0 = hbatch::(vec![((Collide(1), 5), 0, 1), ((Collide(2), 7), 0, 1)], 0, 1); + let (p0, _f) = tactic.retire( + vec![], vec![], vec![b0.clone()], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + let out0: Vec<_> = p0.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + assert_eq!(out0, vec![((Collide(1), 5u64), 0u64, 1i64), ((Collide(2), 7), 0, 1)], "retire 1"); + + // Retire 2: a novel update to the LOWER key only, so the id order is [C1(hist), C2(hist), C1(novel)]. + let out_batches: Vec<_> = { + let mut t2 = ProxyReduceTactic::new(VecReduceBackend::new(logic)); + let b = hbatch::(vec![((Collide(1), 5), 0, 1), ((Collide(2), 7), 0, 1)], 0, 1); + let (p, _) = t2.retire(vec![], vec![], vec![b], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64)); + p.into_iter().map(|(_t, b)| b).collect() + }; + let b1 = hbatch::(vec![((Collide(1), 9), 1, 1)], 1, 2); + let (p1, _f) = tactic.retire( + vec![b0], out_batches, vec![b1], + &Antichain::from_elem(1u64), &Antichain::from_elem(2u64), &Antichain::from_elem(1u64), + ); + let out1: Vec<_> = p1.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + // C1's max rises 5 -> 9; C2 is untouched and must NOT be disturbed. + assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 9), 1, 1)], "retire 2 must not disturb C2"); +} + +/// A reduction that emits only for `Collide(1)`, so `Collide(2)` has input but never any output. +fn only_first(k: &Collide, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>) { + if k.0 == 1 { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + } + for (w, d) in current.iter() { updates.push((*w, -*d)); } +} + +/// Input ids are ordinals minted history-first, so the id order across a hash bracket is +/// `[history by key, then novel by key]` — NOT globally key-sorted. Here that makes the bracket +/// read `[C1, C2, C1]` by id while its endpoints agree, and the output bracket holds only `C1` +/// (C2 never emits), so the endpoint test concludes the bracket is a single key. +#[test] +fn reduce_collision_fastpath_endpoints() { + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(only_first)); + let b0 = hbatch::(vec![((Collide(1), 5), 0, 1), ((Collide(2), 900), 0, 1)], 0, 1); + let (p0, _) = tactic.retire( + vec![], vec![], vec![b0.clone()], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + let outs: Vec<_> = p0.into_iter().map(|(_t, b)| b).collect(); + assert_eq!(outs.iter().flat_map(|b| hread(std::slice::from_ref(b))).collect::>(), + vec![((Collide(1), 5u64), 0u64, 1i64)], "retire 1: only C1 emits"); + + let b1 = hbatch::(vec![((Collide(1), 6), 1, 1)], 1, 2); + let (p1, _) = tactic.retire( + vec![b0], outs, vec![b1], + &Antichain::from_elem(1u64), &Antichain::from_elem(2u64), &Antichain::from_elem(1u64), + ); + let out1: Vec<_> = p1.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + // C1's values are {5, 6}: the max becomes 6. C2's 900 belongs to a different real key. + assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 6), 1, 1)], + "C2's value must not enter C1's reduction"); +} + +#[test] +fn reduce_cancelling_keys_matches_mainline() { + // Every key's input cancels completely by time 2, so from then on it has no records in any of + // the three presentations while still being a key the retire must consider — its stale output + // has to be retracted. With one key per window this is also the case where a window's key list + // and the `changed` set disagree. + let mut updates = Vec::new(); + for k in 0..16u64 { + updates.push(((k, 10 + k), 0u64, 1i64)); + updates.push(((k, 20 + k), 1, 1)); + updates.push(((k, 10 + k), 2, -1)); + updates.push(((k, 20 + k), 2, -1)); + } + proxy_matches_mainline(updates, 1); +} diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs new file mode 100644 index 000000000..6872c0ed4 --- /dev/null +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -0,0 +1,458 @@ +//! Benchmarks the cursor tactic against the proxy tactic over IDENTICAL storage. +//! +//! Both tactics drive a reduction over the same hash-keyed `ChunkSpine` arrangement — the cursor +//! tactic through [`Arranged::reduce_core`], the proxy tactic through `reduce_with_tactic` with the +//! [`VecReduceBackend`] — so their difference is the tactic (and its backend boundary), not the +//! storage. The inherent `reduce` (its own `OrdVal`-style arrangement) is carried as a familiar +//! reference point. +//! +//! Three workloads cover the regimes that have historically mattered: +//! * `churn` — bounded per-key history, one distinct time per flush: the streaming steady state. +//! * `multimoment` — several distinct times per flush: the batched/throughput regime. +//! * `propagate` — label propagation inside `iterate`: `Product` times and carried interesting +//! times across retires, the shape on which visiting non-due keys was quadratic (PR #824). +//! +//! Run with: +//! ```text +//! cargo test --release --test int_proxy_bench -- --ignored --nocapture +//! ``` + +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::probe::Handle as ProbeHandle; +use timely::dataflow::operators::Probe; +use timely::order::Product; + +use differential_dataflow::hashable::Hashable; +use differential_dataflow::input::InputSession; +use differential_dataflow::operators::arrange::arrangement::arrange_core; +use differential_dataflow::operators::int_proxy::reduce::ProxyReduceTactic; +use differential_dataflow::operators::int_proxy::vec_backend::VecReduceBackend; +use differential_dataflow::operators::iterate::Iterate; +use differential_dataflow::operators::reduce::reduce_with_tactic; +use differential_dataflow::trace::chunk::vec::{ + ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine, + VecChunk, VecChunkCursor, +}; +use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::implementations::ContainerChunker; + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +enum Mode { + /// The inherent `reduce`: cursor tactic over its own conventional arrangement. + Mainline, + /// The cursor tactic over the hash-keyed chunk arrangement. + CursorSame, + /// The proxy tactic + `VecReduceBackend` over the hash-keyed chunk arrangement. + Proxy, +} + +/// Arrange `(k, v) -> (hash(k), (k, v))` into a hash-keyed `ChunkSpine`. +macro_rules! harrange { + ($coll:expr, $t:ty, $name:expr) => {{ + let hashed = $coll.map(|(k, v)| (k.hashed(), (k, v))); + arrange_core::< + Pipeline, + Vec<((u64, (u64, u64)), $t, isize)>, + ContainerChunker>, + VChunkBatcher, + VChunkBuilder, + VChunkSpine, + >(hashed.inner, Pipeline, $name) + }}; +} + +/// The cursor tactic over a hash-keyed chunk arrangement, with `$fold` computing the desired +/// output value from the accumulated `(key, val)` pairs. +macro_rules! cursor_reduce { + ($arr:expr, $t:ty, $name:expr, $fold:expr) => { + $arr.reduce_core::< + _, + VChunkBuilder, + VChunkSpine, + as Cursor>::KeyContainer, + _, + >( + $name, + |_h, input, current, updates| { + if let Some(kw) = $fold(input.iter().filter(|(_, d)| *d > 0).map(|(kv, _)| **kv)) { + updates.push((kw, 1)); + } + for (w, d) in current.iter() { + updates.push((w.clone(), -*d)); + } + }, + |chunk, key, list| { + use timely::container::PushInto; + *chunk = Default::default(); + for (v, t, d) in list.drain(..) { + chunk.push_into(((*key, v), t, d)); + } + }, + ) + }; +} + +/// The proxy tactic + backend over a hash-keyed chunk arrangement, same `$fold`. +macro_rules! proxy_reduce { + ($arr:expr, $t:ty, $name:expr, $fold:expr) => { + reduce_with_tactic::<_, VChunkSpine, _>( + $arr, + $name, + ProxyReduceTactic::new(VecReduceBackend::new( + |k: &u64, input: &[(u64, isize)], current: &mut Vec<(u64, isize)>, updates: &mut Vec<(u64, isize)>| { + if let Some((_, w)) = $fold(input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| (*k, *v))) { + updates.push((w, 1)); + } + for (w, d) in current.iter() { + updates.push((*w, -*d)); + } + }, + )), + ) + }; +} + +/// A `String`-valued churn: the same shape as `churn`, but every value clone allocates. The +/// `u64` workloads cannot show what the proxy boundary costs to marshal owned data, because +/// cloning a `u64` is free. +fn wide_keys() -> u64 { sized("WIDE_KEYS", 20_000) } +fn wide_len() -> usize { sized("WIDE_LEN", 48) as usize } + +fn wide_value(k: u64, t: u64, len: usize) -> String { + let mut s = String::with_capacity(len); + s.push_str(&format!("{k:016x}{t:016x}")); + while s.len() < len { s.push('x'); } + s +} + +fn run_wide(mode: Mode) -> f64 { + let keys = wide_keys(); + let len = wide_len(); + timely::execute_directly(move |worker| { + let mut input: InputSession = InputSession::new(); + let probe = worker.dataflow::(|scope| { + let coll = input.to_collection(scope); + let mut ph = ProbeHandle::new(); + match mode { + Mode::Mainline => { + coll.reduce(|_k, i: &[(&String, isize)], o: &mut Vec<(String, isize)>| { + if let Some(m) = i.iter().filter(|(_, d)| *d > 0).map(|(v, _)| (*v).clone()).max() { + o.push((m, 1)); + } + }).inner.probe_with(&mut ph); + } + Mode::CursorSame => { + let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, + ContainerChunker>, + VChunkBatcher, + VChunkBuilder, + VChunkSpine>(hashed.inner, Pipeline, "ArrW"); + arr.reduce_core::<_, VChunkBuilder, + VChunkSpine, + as Cursor>::KeyContainer, _>( + "CursorWide", + |_h, input, current, updates| { + if let Some(kv) = input.iter().filter(|(_, d)| *d > 0).map(|(kv, _)| (*kv).clone()).max_by(|a, b| a.1.cmp(&b.1)) { + updates.push((kv, 1)); + } + for (w, d) in current.iter() { updates.push((w.clone(), -*d)); } + }, + |chunk, key, list| { + use timely::container::PushInto; + *chunk = Default::default(); + for (v, t, d) in list.drain(..) { chunk.push_into(((*key, v), t, d)); } + }, + ).stream.probe_with(&mut ph); + } + Mode::Proxy => { + let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, + ContainerChunker>, + VChunkBatcher, + VChunkBuilder, + VChunkSpine>(hashed.inner, Pipeline, "ArrW"); + reduce_with_tactic::<_, VChunkSpine, _>( + arr, "ProxyWide", + ProxyReduceTactic::new(VecReduceBackend::new( + |_k: &u64, input: &[(String, isize)], current: &mut Vec<(String, isize)>, updates: &mut Vec<(String, isize)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { updates.push((w.clone(), -*d)); } + }, + )), + ).stream.probe_with(&mut ph); + } + } + ph + }); + input.advance_to(0); + for k in 0..keys { input.insert((k, wide_value(k, 0, len))); } + input.advance_to(1); + input.flush(); + while probe.less_than(&1) { worker.step(); } + let mut times = Vec::new(); + let warm = warmup(); + for r in 0..rounds() { + let t = 1 + r; + input.advance_to(t); + for k in 0..keys { + if t > 1 { input.remove((k, wide_value(k, t - 1, len))); } + input.insert((k, wide_value(k, t, len))); + } + input.advance_to(t + 1); + input.flush(); + let start = Instant::now(); + while probe.less_than(&(t + 1)) { worker.step(); } + if r >= warm { times.push(start.elapsed().as_micros()); } + } + times.iter().sum::() as f64 / times.len() as f64 + }) +} + +fn max_fold(iter: impl Iterator) -> Option<(u64, u64)> { + iter.max_by_key(|kv| kv.1) +} + +fn min_fold(iter: impl Iterator) -> Option<(u64, u64)> { + iter.min_by_key(|kv| kv.1) +} + +/// Workload sizes, overridable so the suite can be run at a scale where costs that are invisible +/// at a few megabytes show up. `churn_keys()=4000000 ROUNDS=6 WARMUP=2` moves gigabytes per round. +fn sized(var: &str, default: u64) -> u64 { + std::env::var(var).ok().and_then(|v| v.parse().ok()).unwrap_or(default) +} +fn churn_keys() -> u64 { sized("CHURN_KEYS", 50_000) } +fn mm_keys() -> u64 { sized("MM_KEYS", 10_000) } +fn mm_sub() -> u64 { sized("MM_SUB", 4) } +fn rounds() -> u64 { sized("ROUNDS", 16) } +fn warmup() -> u64 { sized("WARMUP", 6) } + +/// Bounded churn, one distinct time per flush; returns averaged microseconds per round. +fn run_churn(mode: Mode) -> f64 { + run_flat(mode, churn_keys(), 1) +} + +/// Bounded churn, `mm_sub()` distinct times per flush. +fn run_multimoment(mode: Mode) -> f64 { + run_flat(mode, mm_keys(), mm_sub()) +} + +fn run_flat(mode: Mode, keys: u64, sub: u64) -> f64 { + timely::execute_directly(move |worker| { + let mut input: InputSession = InputSession::new(); + let probe = worker.dataflow::(|scope| { + let coll = input.to_collection(scope); + let mut ph = ProbeHandle::new(); + match mode { + Mode::Mainline => { + coll.reduce(|_k, i: &[(&u64, isize)], o: &mut Vec<(u64, isize)>| { + if let Some(m) = i.iter().filter(|(_, d)| *d > 0).map(|(v, _)| **v).max() { + o.push((m, 1)); + } + }) + .inner + .probe_with(&mut ph); + } + Mode::CursorSame => { + let arr = harrange!(coll, u64, "Arrange"); + cursor_reduce!(arr, u64, "CursorReduce", max_fold).stream.probe_with(&mut ph); + } + Mode::Proxy => { + let arr = harrange!(coll, u64, "Arrange"); + proxy_reduce!(arr, u64, "ProxyReduce", max_fold).stream.probe_with(&mut ph); + } + } + ph + }); + input.advance_to(0); + for k in 0..keys { + input.insert((k, 0)); + input.insert((k, 1)); + } + input.advance_to(1); + input.flush(); + while probe.less_than(&1) { + worker.step(); + } + let mut times = Vec::new(); + for r in 0..rounds() { + let base = 1 + r * sub; + for s in 0..sub { + let t = base + s; + input.advance_to(t); + for k in 0..keys { + if t > 1 { + input.remove((k, 1000 + t - 1)); + } + input.insert((k, 1000 + t)); + } + } + input.advance_to(base + sub); + input.flush(); + let start = Instant::now(); + while probe.less_than(&(base + sub)) { + worker.step(); + } + if r >= warmup() { + times.push(start.elapsed().as_micros()); + } + } + times.iter().sum::() as f64 / times.len() as f64 + }) +} + +const PROP_NODES: u64 = 2_000; +const PROP_EDGES: usize = 4_000; +const PROP_CHURN: usize = 50; +const PROP_ROUNDS: u64 = 40; +const PROP_WARMUP: u64 = 5; + +fn edge(i: usize) -> (u64, u64) { + let n = PROP_NODES; + ((i as u64).wrapping_mul(2654435761) % n, (i as u64).wrapping_mul(40503).wrapping_add(7) % n) +} + +/// Label propagation (min over self and in-neighbors) inside `iterate`, with edge churn: the +/// carried-interesting-times shape. Returns averaged microseconds per round and a checksum of the +/// produced label updates, which must agree across modes. +fn run_propagate(mode: Mode) -> (f64, i64) { + let check = Arc::new(Mutex::new(0i64)); + let sum = check.clone(); + let avg = timely::execute_directly(move |worker| { + let mut nodes: InputSession = InputSession::new(); + let mut edges: InputSession = InputSession::new(); + let probe = worker.dataflow::(|scope| { + let nodes = nodes.to_collection(scope); + let edges = edges.to_collection(scope); + let mut ph = ProbeHandle::new(); + let labels = nodes.clone().iterate(move |_scope, inner| { + let scope = inner.scope(); + let nodes_in = nodes.enter(scope); + let edges_in = edges.enter(scope); + let prop = edges_in.join_map(inner, |_src, dst, lbl| (*dst, *lbl)).concat(nodes_in); + match mode { + Mode::Mainline => prop.reduce(|_k, i: &[(&u64, isize)], o: &mut Vec<(u64, isize)>| { + if let Some(m) = i.iter().filter(|(_, d)| *d > 0).map(|(v, _)| **v).min() { + o.push((m, 1)); + } + }), + Mode::CursorSame => { + let arr = harrange!(prop, Product, "ArrProp"); + cursor_reduce!(arr, Product, "CursorProp", min_fold) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + } + Mode::Proxy => { + let arr = harrange!(prop, Product, "ArrProp"); + proxy_reduce!(arr, Product, "ProxyProp", min_fold) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + } + } + }); + labels + .inspect(move |((_, lbl), _, d)| { + *sum.lock().unwrap() += (*lbl as i64) * (*d as i64); + }) + .inner + .probe_with(&mut ph); + ph + }); + nodes.advance_to(0); + edges.advance_to(0); + for k in 0..PROP_NODES { + nodes.insert((k, k)); + } + for i in 0..PROP_EDGES { + let (s, d) = edge(i); + edges.insert((s, d)); + } + nodes.advance_to(1); + nodes.flush(); + edges.advance_to(1); + edges.flush(); + while probe.less_than(&1) { + worker.step(); + } + let mut times = Vec::new(); + for r in 0..PROP_ROUNDS { + let t = r + 1; + // Perturb a low-numbered label each round: low labels spread widely under `min`, so + // withdrawing one forces its reach to relabel — real iterate work, with interesting + // times carried across retires. + let m = r % 50; + if r > 0 { + let p = (r - 1) % 50; + nodes.insert((p, p)); + } + nodes.remove((m, m)); + for c in 0..PROP_CHURN { + let dead = (r as usize) * PROP_CHURN + c; + let (s, d) = edge(dead); + edges.remove((s, d)); + let (s, d) = edge(PROP_EDGES + dead); + edges.insert((s, d)); + } + nodes.advance_to(t + 1); + nodes.flush(); + edges.advance_to(t + 1); + edges.flush(); + let start = Instant::now(); + while probe.less_than(&(t + 1)) { + worker.step(); + } + if r >= PROP_WARMUP { + times.push(start.elapsed().as_micros()); + } + } + times.iter().sum::() as f64 / times.len() as f64 + }); + let total = *check.lock().unwrap(); + (avg, total) +} + +fn report(name: &str, mainline: f64, cursor: f64, proxy: f64) { + eprintln!(" {name}:"); + eprintln!(" inherent reduce (own storage) : {:9.0}us/round", mainline); + eprintln!(" cursor tactic (chunk storage) : {:9.0}us/round", cursor); + eprintln!(" proxy tactic (chunk storage) : {:9.0}us/round ({:.2}x cursor-same, {:.2}x inherent)", proxy, proxy / cursor, proxy / mainline); +} + +#[test] +#[ignore] +fn bench_reduce_tactics() { + eprintln!(); + // `PROG=churn MODE=proxy` runs one cell of the matrix, which is what a large run wants. + if let (Some(prog), Some(mode)) = (std::env::var("PROG").ok(), match std::env::var("MODE").ok().as_deref() { + Some("proxy") => Some(Mode::Proxy), + Some("cursor") => Some(Mode::CursorSame), + Some("mainline") => Some(Mode::Mainline), + _ => None, + }) { + let us = match prog.as_str() { + "churn" => run_churn(mode), + "multimoment" => run_multimoment(mode), + "propagate" => run_propagate(mode).0, + "wide" => run_wide(mode), + other => panic!("unknown PROG {other:?}"), + }; + eprintln!(" {prog} {:?}: {us:.0}us/round", mode); + return; + } + for (name, run) in [ + (format!("churn ({} keys, 1 time/flush)", churn_keys()), run_churn as fn(Mode) -> f64), + (format!("multimoment ({} keys, {} times/flush)", mm_keys(), mm_sub()), run_multimoment), + ] { + report(&name, run(Mode::Mainline), run(Mode::CursorSame), run(Mode::Proxy)); + } + let (mainline, sum_m) = run_propagate(Mode::Mainline); + let (cursor, sum_c) = run_propagate(Mode::CursorSame); + let (proxy, sum_p) = run_propagate(Mode::Proxy); + assert_eq!(sum_m, sum_c, "cursor-tactic propagate output must match the inherent reduce"); + assert_eq!(sum_m, sum_p, "proxy-tactic propagate output must match the inherent reduce"); + report("propagate (2k nodes, 4k edges, iterate)", mainline, cursor, proxy); +}