diff --git a/.gitignore b/.gitignore index 3dfb331f474..3d7ecc32d44 100644 --- a/.gitignore +++ b/.gitignore @@ -101,4 +101,8 @@ playwright/.cache/ # Devcontainers - the lock file should be committed when more people use them .devcontainer/devcontainer-lock.json js-packages/support-bundle-triage/ -.roc-check/ \ No newline at end of file +.roc-check/ + +# Local dev configs +.envrc +.direnv diff --git a/crates/dbsp/Cargo.toml b/crates/dbsp/Cargo.toml index 8af54c76821..283d0f18d41 100644 --- a/crates/dbsp/Cargo.toml +++ b/crates/dbsp/Cargo.toml @@ -167,6 +167,18 @@ harness = false name = "filter_predictor" harness = false +[[bench]] +name = "transitive_closure_acyclic" +harness = false + +[[bench]] +name = "transitive_closure_cyclic" +harness = false + +[[bench]] +name = "graph_coloring" +harness = false + [[example]] name = "orgchart" diff --git a/crates/dbsp/benches/graph_coloring.rs b/crates/dbsp/benches/graph_coloring.rs new file mode 100644 index 00000000000..f724bdf801b --- /dev/null +++ b/crates/dbsp/benches/graph_coloring.rs @@ -0,0 +1,362 @@ +//! Benchmark comparing two graph 2-coloring (bipartiteness) approaches: +//! +//! - **recursive**: `Circuit::recursive_dynamic`, which splices an implicit +//! `distinct` onto every recursive output stream. +//! See [`build_recursive_variant`]. +//! - **iterate**: `Circuit::iterate` with a manual feedback loop that carries +//! only the per-iteration frontier and deduplicates with an explicit +//! `distinct`. See [`build_iterate_variant`]. +//! +//! Unlike the transitive-closure benchmarks, both variants perform the same +//! deduplication work (implicit vs. explicit `distinct`). Hence, we should see +//! comparable run times. The coloring propagates from a single seed node +//! along directed edges: a red node paints its out-neighbors blue and vice +//! versa. On a strongly connected graph with odd cycles every reachable node +//! ends up both red and blue, so both variants converge to the same colorings +//! (the benchmark asserts this before measuring). +//! +//! Run with, e.g.: +//! `cargo bench --bench graph_coloring` + +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use dbsp::{ + Circuit, Consensus, DBSPHandle, NestedCircuit, NumEntries, OrdIndexedZSet, OrdZSet, + OutputHandle, Runtime, Stream, ZSetHandle, ZWeight, mimalloc::MiMalloc, operator::Z1, + trace::BatchReaderFactories, typed_batch::SpineSnapshot, utils::Tup2, +}; +use rand::{Rng, SeedableRng}; +use rand_chacha::ChaCha8Rng; +use std::cell::Cell; + +#[global_allocator] +static ALLOC: MiMalloc = MiMalloc; + +type NodeId = u64; +type Edge = Tup2; + +struct BipartiteGraphCircuit { + handle: DBSPHandle, + init_input: ZSetHandle, + edges_input: ZSetHandle, + red_output: OutputHandle>>, + blue_output: OutputHandle>>, +} + +/// Number of DBSP worker threads used for every circuit in this benchmark. +const WORKERS: usize = 4; + +/// Deterministic seed so every run sees the same graphs. +const SEED: u64 = 0x0BADC0DE_DEADBEEF; + +/// Out-edges emitted per node by [`generate_graph`]. +const OUT_DEGREE: usize = 3; + +/// A single-transaction workload: the seed node(s) plus all edges of one graph. +#[derive(Clone)] +struct Workload { + init: Vec>, + edges: Vec>, +} + +/// Generates a random directed graph plus a single seed node (node `0`, painted +/// red) as one insertion batch. +/// +/// Each node emits `out_degree` edges to distinct random targets (excluding +/// itself). With `out_degree >= 2` and a few hundred nodes the graph is +/// strongly connected with overwhelming probability and contains odd cycles, so +/// the coloring reaches every node in both colors. That makes the two variants +/// produce identical red/blue sets regardless of the exact structure. +fn generate_graph(nodes: NodeId, out_degree: usize) -> Workload { + let mut rng = ChaCha8Rng::seed_from_u64(SEED); + // At most `nodes - 1` distinct targets exist per node (self is excluded). + let fanout = out_degree.min(nodes.saturating_sub(1) as usize); + let mut edges = Vec::with_capacity(nodes as usize * fanout); + + for from in 0..nodes { + let mut targets = Vec::with_capacity(fanout); + while targets.len() < fanout { + let to = rng.gen_range(0..nodes); + if to != from && !targets.contains(&to) { + targets.push(to); + } + } + for to in targets { + edges.push(Tup2(Tup2(from, to), 1)); + } + } + + // Seed the discovery by marking node `0` red. + let init = vec![Tup2(0, 1)]; + + Workload { init, edges } +} + +/// Feeds a whole graph into a freshly built circuit in one transaction and +/// returns the number of `(red, blue)` color marks discovered. +/// +/// The circuit (and its worker threads) is dropped on return, so the caller +/// must supply a fresh circuit for each measured iteration. +fn drive_once(mut circuit: BipartiteGraphCircuit, mut workload: Workload) -> (usize, usize) { + circuit.init_input.append(&mut workload.init); + circuit.edges_input.append(&mut workload.edges); + circuit + .handle + .transaction() + .expect("transaction should succeed"); + let red = circuit + .red_output + .concat() + .consolidate() + .num_entries_shallow(); + let blue = circuit + .blue_output + .concat() + .consolidate() + .num_entries_shallow(); + (red, blue) +} + +/// Cross-checks that both variants compute the same coloring on the given graph, +/// returning the number of `(red, blue)` color marks (used both as a sanity check +/// and to label the benchmark with the output cardinality). +fn assert_variants_agree(workload: &Workload) -> (usize, usize) { + let recursive = build_recursive_variant().expect("build recursive circuit"); + let iterate = build_iterate_variant().expect("build iterate circuit"); + + let mut recursive_circuit = recursive.handle; + let mut iterate_circuit = iterate.handle; + + recursive.init_input.append(&mut workload.init.clone()); + recursive.edges_input.append(&mut workload.edges.clone()); + iterate.init_input.append(&mut workload.init.clone()); + iterate.edges_input.append(&mut workload.edges.clone()); + recursive_circuit.transaction().unwrap(); + iterate_circuit.transaction().unwrap(); + + let recursive_red = recursive.red_output.concat().consolidate(); + let recursive_blue = recursive.blue_output.concat().consolidate(); + let iterate_red = iterate.red_output.concat().consolidate(); + let iterate_blue = iterate.blue_output.concat().consolidate(); + + assert_eq!( + recursive_red, iterate_red, + "recursive and iterate variants must agree on the red coloring" + ); + assert_eq!( + recursive_blue, iterate_blue, + "recursive and iterate variants must agree on the blue coloring" + ); + + ( + recursive_red.num_entries_shallow(), + recursive_blue.num_entries_shallow(), + ) +} + +/// The graph sizes (node counts) we benchmark. +const GRAPH_SIZES: &[NodeId] = &[500, 1000, 2000]; + +fn graph_coloring_benches(c: &mut Criterion) { + let mut group = c.benchmark_group("graph-coloring"); + // Building a circuit spins up worker threads, so keep the sample count + // modest to bound the wall-clock time of the whole benchmark. + group.sample_size(30); + + for &nodes in GRAPH_SIZES { + let workload = generate_graph(nodes, OUT_DEGREE); + let (red_marks, blue_marks) = assert_variants_agree(&workload); + let shape = format!("n{nodes}"); + let edges = workload.edges.len(); + println!("graph {shape}: {edges} edges, {red_marks} red / {blue_marks} blue color marks",); + + // Report throughput in edges processed per second, which scales with the + // graph size and the join work both variants perform. + group.throughput(Throughput::Elements(edges as u64)); + + group.bench_with_input( + BenchmarkId::new("recursive", &shape), + &workload, + |b, workload| { + b.iter_batched( + || { + ( + build_recursive_variant().expect("build recursive circuit"), + workload.clone(), + ) + }, + |(circuit, workload)| drive_once(circuit, workload), + BatchSize::PerIteration, + ); + }, + ); + + group.bench_with_input( + BenchmarkId::new("iterate", &shape), + &workload, + |b, workload| { + b.iter_batched( + || { + ( + build_iterate_variant().expect("build iterate circuit"), + workload.clone(), + ) + }, + |(circuit, workload)| drive_once(circuit, workload), + BatchSize::PerIteration, + ); + }, + ); + } + + group.finish(); +} + +criterion_group!(benches, graph_coloring_benches); +criterion_main!(benches); + +/// Helper function to advance the frontier of red and blue nodes by exploring +/// one more hop in the graph along `edges`. +fn hop( + frontier: &Stream>, + edges: &Stream>, +) -> Stream> { + frontier + .map_index(|node| (*node, *node)) + .join(edges, |_from, _node, to| *to) +} + +/// Computes the red/blue graph coloring using DBSP's +/// [`recursive` API](dbsp::circuit::ChildCircuit::recursive), +/// which deduplicates every recursive output stream implicitly. +fn build_recursive_variant() -> anyhow::Result { + let (handle, ((init_input, edges_input), (red_output, blue_output))) = + Runtime::init_circuit(WORKERS, move |root_circuit| { + let (edges, edges_input) = root_circuit.add_input_zset::(); + let (init, init_input) = root_circuit.add_input_zset::(); + + let (red_output, blue_output) = root_circuit.recursive( + |child_circuit, + (red, blue): ( + Stream>, + Stream>, + )| { + let edges = edges.delta0(child_circuit); + let init = init.delta0(child_circuit); + let edges_indexed = edges.map_index(|Tup2(from, to)| (*from, *to)); + + let new_red = init.plus(&hop(&blue, &edges_indexed)); + let new_blue = hop(&red, &edges_indexed); + + Ok((new_red, new_blue)) + }, + )?; + + let red_output = red_output.accumulate_output(); + let blue_output = blue_output.accumulate_output(); + + Ok(((init_input, edges_input), (red_output, blue_output))) + })?; + + Ok(BipartiteGraphCircuit { + handle, + init_input, + edges_input, + red_output, + blue_output, + }) +} + +/// Upper bound on nested iterations before the iterate variant gives up waiting +/// for a fixed point. +const MAX_ITERATIONS: usize = 4096; + +/// Computes the red/blue graph coloring using DBSP's lower level +/// [`iterate` API](dbsp::circuit::Circuit::iterate). +/// +/// It uses a hybrid termination check: Either stop once the subcircuit +/// reaches a fixed point, or after [`MAX_ITERATIONS`] steps if it never does. +fn build_iterate_variant() -> anyhow::Result { + let (handle, (init_input, edges_input, red_output, blue_output)) = + Runtime::init_circuit(WORKERS, move |root_circuit| { + let (init, init_input) = root_circuit.add_input_zset::(); + let (edges, edges_input) = root_circuit.add_input_zset::(); + + let (red_output, blue_output) = root_circuit.iterate(|child_circuit| { + let edges = edges.delta0(child_circuit); + let init = init.delta0(child_circuit); + let edges_indexed = edges.map_index(|Tup2(from, to)| (*from, *to)); + + // Feedback carries only the frontier (the delta from the last step). + let (red, red_feedback) = + child_circuit.add_feedback(Z1::new(OrdZSet::::default())); + let (blue, blue_feedback) = + child_circuit.add_feedback(Z1::new(OrdZSet::::default())); + + // Extend each frontier by one hop and deduplicate with `distinct`, + // which caps every node's weight at 1 and stops nodes discovered + // earlier from being re-emitted. + let new_red = init.plus(&hop(&blue, &edges_indexed)).distinct(); + let new_blue = hop(&red, &edges_indexed).distinct(); + + red_feedback.connect(&new_red); + blue_feedback.connect(&new_blue); + + // Accumulate every frontier across the nested clock cycles into the + // full colorings and export them to the parent. + let red = Stream::dyn_integrate_trace( + &new_red.inner(), + &BatchReaderFactories::new::(), + ); + let blue = Stream::dyn_integrate_trace( + &new_blue.inner(), + &BatchReaderFactories::new::(), + ); + + let child_circuit = child_circuit.clone(); + let consensus = Consensus::new("iterate"); + let step = Cell::new(0usize); + + let termination_check = async move || { + let step_count = step.get() + 1; + step.set(step_count); + + // The step bound is data-independent, hence identical on every worker; + // the fixed-point check is per-worker. + let fixedpoint = child_circuit.is_fixedpoint(0); + let iteration_ceiling = step_count >= MAX_ITERATIONS; + let done = consensus.check(fixedpoint || iteration_ceiling).await?; + if done { + // Reset the step counter for the next transaction. + step.set(0); + } + + Ok(done) + }; + + Ok((termination_check, (red.export(), blue.export()))) + })?; + + let red_output = Stream::dyn_consolidate( + &red_output, + &BatchReaderFactories::new::(), + ) + .typed::>() + .accumulate_output(); + let blue_output = Stream::dyn_consolidate( + &blue_output, + &BatchReaderFactories::new::(), + ) + .typed::>() + .accumulate_output(); + + Ok((init_input, edges_input, red_output, blue_output)) + })?; + + Ok(BipartiteGraphCircuit { + handle, + init_input, + edges_input, + red_output, + blue_output, + }) +} diff --git a/crates/dbsp/benches/transitive_closure_acyclic.rs b/crates/dbsp/benches/transitive_closure_acyclic.rs new file mode 100644 index 00000000000..807575a27af --- /dev/null +++ b/crates/dbsp/benches/transitive_closure_acyclic.rs @@ -0,0 +1,348 @@ +//! Benchmark comparing two transitive-closure computation approaches on acyclic +//! graphs similar to `examples/tutorial/tutorial10.rs`: +//! +//! - **recursive**: `Circuit::recursive`, which applies an implicit `distinct` +//! on every nested clock tick. See [`build_recursive_variant`]. +//! - **iterate**: `Circuit::iterate` with a manual feedback loop that only +//! carries the per-iteration frontier and therefore skips the `distinct`. +//! See [`build_iterate_variant`]. +//! +//! The two variants are only semantically equivalent when there is at most one +//! path between any pair of nodes (see [`generate_forest`] for why), so the +//! benchmark drives random forests (and no general DAGs) and asserts the +//! variants agree before measuring. In that regime the recursive variant's +//! `distinct` removes nothing yet still costs, so the iterate variant is +//! expected to win. This mirrors `UNION` vs `UNION ALL` in recursive SQL CTEs. +//! +//! Run with, e.g.: +//! `cargo bench --bench transitive_closure_acyclic` + +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use dbsp::{ + Circuit, Consensus, DBSPHandle, NestedCircuit, NumEntries, OrdIndexedZSet, OutputHandle, + Runtime, Stream, ZSetHandle, ZWeight, + mimalloc::MiMalloc, + operator::Z1, + trace::BatchReaderFactories, + typed_batch::SpineSnapshot, + utils::{Tup2, Tup3, Tup4}, +}; +use rand::{Rng, SeedableRng}; +use rand_chacha::ChaCha8Rng; +use std::cell::Cell; + +#[global_allocator] +static ALLOC: MiMalloc = MiMalloc; + +type NodeId = u64; +type Weight = u64; +type CumWeight = u64; +type Hopcnt = u64; +type WeightedEdge = Tup3; +type Path = Tup2; +type TransClosurePath = Tup4; +type Accumulator = Stream>; + +struct TransClosureCircuit { + handle: DBSPHandle, + edges_input: ZSetHandle, + closure_output: OutputHandle>>, +} + +/// Number of DBSP worker threads used for every circuit in this benchmark. +const WORKERS: usize = 4; + +/// Deterministic seed so every run sees the same graphs. +const SEED: u64 = 0x0BADC0DE_DEADBEEF; + +/// Largest edge weight generated by [`generate_forest`]. +const MAX_EDGE_WEIGHT: Weight = 10; + +/// A single-transaction workload: all edges of one acyclic graph. +type Workload = Vec>; + +/// Generates a random weighted *forest* (each non-root node has exactly one +/// parent) as a single insertion batch. +/// +/// Why a forest rather than an arbitrary DAG? The two variants are only +/// semantically equivalent when there is **at most one path between any pair of +/// nodes**. When several distinct paths connect the same `(start, end)` and +/// happen to share the same cumulative weight and hop count, the recursive +/// variant's `distinct` collapses them into a single tuple, whereas the iterate +/// variant (UNION ALL semantics) keeps the multiplicity, so the two disagree on +/// a general DAG. A forest guarantees a unique path per reachable pair: +/// the `distinct` becomes pure overhead that removes nothing, and both variants +/// produce identical transitive closures. +/// +/// Node `child` (for `child` in `1..nodes`) draws a uniformly random parent from +/// `0..child`, so edges always point from a lower id to a higher one and no cycle +/// can form. Node `0` is the single root. A bushy tree still exercises wide +/// fan-out joins because low-id nodes accumulate many children. +fn generate_forest(nodes: NodeId) -> Workload { + let mut rng = ChaCha8Rng::seed_from_u64(SEED); + let mut edges = Vec::with_capacity(nodes.saturating_sub(1) as usize); + + for child in 1..nodes { + let parent = rng.gen_range(0..child); + let weight = rng.gen_range(1..=MAX_EDGE_WEIGHT); + edges.push(Tup2(Tup3(parent, child, weight), 1)); + } + + edges +} + +/// Feeds a whole graph into a freshly built circuit in one transaction and +/// returns the number of paths in the resulting transitive closure. +/// +/// The circuit (and its worker threads) is dropped on return, so the caller +/// must supply a fresh circuit for each measured iteration. +fn drive_once(mut circuit: TransClosureCircuit, mut workload: Workload) -> usize { + circuit.edges_input.append(&mut workload); + circuit + .handle + .transaction() + .expect("transaction should succeed"); + circuit + .closure_output + .concat() + .consolidate() + .num_entries_shallow() +} + +/// Cross-checks that both variants compute the same closure on the given graph, +/// returning the shared closure size (used both as a sanity check and to label +/// the benchmark with the output cardinality). +fn assert_variants_agree(workload: &Workload) -> usize { + let recursive = build_recursive_variant().expect("build recursive circuit"); + let iterate = build_iterate_variant().expect("build iterate circuit"); + + let mut recursive_circuit = recursive.handle; + let mut iterate_circuit = iterate.handle; + + recursive.edges_input.append(&mut workload.clone()); + iterate.edges_input.append(&mut workload.clone()); + recursive_circuit.transaction().unwrap(); + iterate_circuit.transaction().unwrap(); + + let recursive_closure = recursive.closure_output.concat().consolidate(); + let iterate_closure = iterate.closure_output.concat().consolidate(); + + assert_eq!( + recursive_closure, iterate_closure, + "recursive and iterate variants must agree on acyclic input" + ); + + recursive_closure.num_entries_shallow() +} + +/// The forest sizes (in terms of node counts) we benchmark. +const GRAPH_SIZES: &[NodeId] = &[500, 1000, 2000]; + +fn transitive_closure_benches(c: &mut Criterion) { + let mut group = c.benchmark_group("transitive-closure-acyclic"); + // Building a circuit spins up worker threads, so keep the sample count + // modest to bound the wall-clock time of the whole benchmark. + group.sample_size(25); + + for &nodes in GRAPH_SIZES { + let workload = generate_forest(nodes); + let closure_size = assert_variants_agree(&workload); + let shape = format!("n{nodes}"); + println!( + "graph {shape}: {} edges, closure of {closure_size} paths", + workload.len() + ); + + // Report throughput in paths of the computed closure per second so the + // two variants are compared on the same unit of useful work. + group.throughput(Throughput::Elements(closure_size as u64)); + + group.bench_with_input( + BenchmarkId::new("recursive", &shape), + &workload, + |b, workload| { + b.iter_batched( + || { + ( + build_recursive_variant().expect("build recursive circuit"), + workload.clone(), + ) + }, + |(circuit, workload)| drive_once(circuit, workload), + BatchSize::PerIteration, + ); + }, + ); + + group.bench_with_input( + BenchmarkId::new("iterate", &shape), + &workload, + |b, workload| { + b.iter_batched( + || { + ( + build_iterate_variant().expect("build iterate circuit"), + workload.clone(), + ) + }, + |(circuit, workload)| drive_once(circuit, workload), + BatchSize::PerIteration, + ); + }, + ); + } + + group.finish(); +} + +criterion_group!(benches, transitive_closure_benches); +criterion_main!(benches); + +/// Computes the transitive closure of an acyclic graph using DBSP's +/// [`recursive` API](dbsp::circuit::ChildCircuit::recursive). +fn build_recursive_variant() -> anyhow::Result { + let (handle, (edges_input, closure_output)) = + Runtime::init_circuit(WORKERS, move |root_circuit| { + let (edges, edges_input) = root_circuit.add_input_zset::(); + + let len_1 = edges.map_index(|Tup3(from, to, weight)| { + (Tup2(*from, *to), Tup4(*from, *to, *weight, 1)) + }); + + let closure = root_circuit.recursive(|child_circuit, closure: Accumulator| { + let edges = edges.delta0(child_circuit); + let len_1 = len_1.delta0(child_circuit); + + let closure = len_1 + .plus( + &closure + .map_index( + |(Tup2(_start, _end), Tup4(start, end, cum_weight, hopcnt))| { + (*end, Tup4(*start, *end, *cum_weight, *hopcnt)) + }, + ) + .join_index( + &edges.map_index(|Tup3(from, to, weight)| { + (*from, Tup3(*from, *to, *weight)) + }), + |_end_from, + Tup4(start, _end, cum_weight, hopcnt), + Tup3(_from, to, weight)| { + Some(( + Tup2(*start, *to), + Tup4(*start, *to, cum_weight + weight, hopcnt + 1), + )) + }, + ), + // + ) + // We explicitly mark the stream as distinct to avoid the + // unnecessary implicit distinct. + .mark_distinct(); + + Ok(closure) + })?; + + Ok((edges_input, closure.accumulate_output())) + })?; + + Ok(TransClosureCircuit { + handle, + edges_input, + closure_output, + }) +} + +/// Upper bound on nested iterations before the iterate variant gives up waiting +/// for a fixed point. +const MAX_ITERATIONS: usize = 4096; + +/// Computes the transitive closure of an acyclic graph using DBSP's lower level +/// [`iterate` API](dbsp::circuit::Circuit::iterate). +/// +/// It uses a hybrid termination check: Either stop once the subcircuit +/// reaches a fixed point, or after [`MAX_ITERATIONS`] steps if it never does. +fn build_iterate_variant() -> anyhow::Result { + let (handle, (edges_input, closure_output)) = + Runtime::init_circuit(WORKERS, move |root_circuit| { + let (edges, edges_input) = root_circuit.add_input_zset::(); + + let len_1 = edges.map_index(|Tup3(from, to, weight)| { + (Tup2(*from, *to), Tup4(*from, *to, *weight, 1)) + }); + + let closure = root_circuit.iterate(|child_circuit| { + let edges = edges.delta0(child_circuit); + let len_1 = len_1.delta0(child_circuit); + + let (closure_frontier, closure_frontier_feedback) = child_circuit + .add_feedback(Z1::new(OrdIndexedZSet::::default())); + + let new_paths = len_1.plus( + &closure_frontier + .map_index( + |(Tup2(_start, _end), Tup4(start, end, cum_weight, hopcnt))| { + (*end, Tup4(*start, *end, *cum_weight, *hopcnt)) + }, + ) + .join_index( + &edges.map_index(|Tup3(from, to, weight)| { + (*from, Tup3(*from, *to, *weight)) + }), + |_end_from, + Tup4(start, _end, cum_weight, hopcnt), + Tup3(_from, to, weight)| { + Some(( + Tup2(*start, *to), + Tup4(*start, *to, cum_weight + weight, hopcnt + 1), + )) + }, + ), + ); + + closure_frontier_feedback.connect(&new_paths); + + let closure = Stream::dyn_integrate_trace( + &new_paths.inner(), + &BatchReaderFactories::new::(), + ); + + let child_circuit = child_circuit.clone(); + let consensus = Consensus::new("iterate"); + let step = Cell::new(0usize); + let termination_check = async move || { + let step_count = step.get() + 1; + step.set(step_count); + + // The step bound is data-independent, hence identical on every worker; + // the fixed-point check is per-worker. + let fixedpoint = child_circuit.is_fixedpoint(0); + let iteration_ceiling = step_count >= MAX_ITERATIONS; + let done = consensus.check(fixedpoint || iteration_ceiling).await?; + if done { + // Reset the step counter for the next transaction. + step.set(0); + } + + Ok(done) + }; + + Ok((termination_check, closure.export())) + })?; + + let closure = Stream::dyn_consolidate( + &closure, + &BatchReaderFactories::new::(), + ) + .typed::>() + .accumulate_output(); + + Ok((edges_input, closure)) + })?; + + Ok(TransClosureCircuit { + handle, + edges_input, + closure_output, + }) +} diff --git a/crates/dbsp/benches/transitive_closure_cyclic.rs b/crates/dbsp/benches/transitive_closure_cyclic.rs new file mode 100644 index 00000000000..73625e9f053 --- /dev/null +++ b/crates/dbsp/benches/transitive_closure_cyclic.rs @@ -0,0 +1,373 @@ +//! Benchmark comparing two transitive-closure computation approaches on cyclic +//! graphs similar to `examples/tutorial/tutorial11.rs`: +//! +//! - **recursive**: `Circuit::recursive` (which applies an implicit `distinct` +//! on every nested clock tick) followed by a `Min` aggregation that both keeps +//! the shortest path per pair and guarantees termination on cycles. +//! See [`build_recursive_variant`]. +//! - **iterate**: `Circuit::iterate` with a manual feedback loop that carries +//! only the per-iteration frontier and relies solely on the same `Min` +//! aggregation for termination and does not implicitly add a `distinct`. +//! See [`build_iterate_variant`]. +//! +//! On cyclic graphs the two approaches from the sister benchmark +//! `transitive_closure_acyclic` would not terminate, so both variants aggregate +//! with `Min` to collapse every `(start, end)` pair to its shortest path. +//! That aggregation makes the two variants produce identical closures +//! regardless of graph shape (the benchmark asserts this before measuring). +//! The recursive variant deduplicates twice, once via the hidden `distinct` and +//! once via `Min`, whereas the iterate variant deduplicates only via `Min`. +//! This benchmark asserts that the additional `distinct` of the recursive +//! variant is essentially a no-op because it operates on top of already +//! distinct data. +//! +//! Run with, e.g.: +//! `cargo bench --bench transitive_closure_cyclic` + +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use dbsp::{ + Circuit, Consensus, DBSPHandle, NestedCircuit, NumEntries, OrdIndexedZSet, OutputHandle, + Runtime, Stream, ZSetHandle, ZWeight, + mimalloc::MiMalloc, + operator::{Min, Z1}, + trace::BatchReaderFactories, + typed_batch::SpineSnapshot, + utils::{Tup2, Tup3, Tup4}, +}; +use rand::{Rng, SeedableRng}; +use rand_chacha::ChaCha8Rng; +use std::cell::Cell; + +#[global_allocator] +static ALLOC: MiMalloc = MiMalloc; + +type NodeId = u64; +type Weight = u64; +type CumWeight = u64; +type Hopcnt = u64; +type WeightedEdge = Tup3; +type Path = Tup2; +type TransClosurePath = Tup4; +type Accumulator = Stream>; + +struct TransClosureCircuit { + handle: DBSPHandle, + edges_input: ZSetHandle, + closure_output: OutputHandle>>, +} + +/// Number of DBSP worker threads used for every circuit in this benchmark. +const WORKERS: usize = 4; + +/// Deterministic seed so every run sees the same graphs. +const SEED: u64 = 0x0BADC0DE_DEADBEEF; + +/// Largest edge weight generated by [`generate_cyclic_graph`]. +const MAX_EDGE_WEIGHT: Weight = 10; + +/// Out-edges emitted per node by [`generate_cyclic_graph`]. +const OUT_DEGREE: usize = 3; + +/// A single-transaction workload: all edges of one cyclic graph. +type Workload = Vec>; + +/// Generates a random weighted *cyclic* graph as a single insertion batch. +/// +/// Each node emits `out_degree` edges to distinct random targets drawn from the +/// whole node range (excluding itself). Unlike the acyclic benchmark, targets +/// may point back to lower ids, so cycles form freely; with `out_degree >= 2` +/// and a few hundred nodes the graph is strongly connected with overwhelming +/// probability, yielding a near-complete transitive closure. +/// +/// Both variants rely on the `Min` aggregation for termination: Because `Min` +/// collapses every `(start, end)` pair to a single shortest path, the two +/// variants produce identical closures no matter how many distinct paths +/// connect a pair of nodes. +fn generate_cyclic_graph(nodes: NodeId, out_degree: usize) -> Workload { + let mut rng = ChaCha8Rng::seed_from_u64(SEED); + // At most `nodes - 1` distinct targets exist per node (self is excluded). + let fanout = out_degree.min(nodes.saturating_sub(1) as usize); + let mut edges = Vec::with_capacity(nodes as usize * fanout); + + for from in 0..nodes { + let mut targets = Vec::with_capacity(fanout); + while targets.len() < fanout { + let to = rng.gen_range(0..nodes); + if to != from && !targets.contains(&to) { + targets.push(to); + } + } + for to in targets { + let weight = rng.gen_range(1..=MAX_EDGE_WEIGHT); + edges.push(Tup2(Tup3(from, to, weight), 1)); + } + } + + edges +} + +/// Feeds a whole graph into a freshly built circuit in one transaction and +/// returns the number of paths in the resulting transitive closure. +/// +/// The circuit (and its worker threads) is dropped on return, so the caller +/// must supply a fresh circuit for each measured iteration. +fn drive_once(mut circuit: TransClosureCircuit, mut workload: Workload) -> usize { + circuit.edges_input.append(&mut workload); + circuit + .handle + .transaction() + .expect("transaction should succeed"); + circuit + .closure_output + .concat() + .consolidate() + .num_entries_shallow() +} + +/// Cross-checks that both variants compute the same closure on the given graph, +/// returning the shared closure size (used both as a sanity check and to label +/// the benchmark with the output cardinality). +fn assert_variants_agree(workload: &Workload) -> usize { + let recursive = build_recursive_variant().expect("build recursive circuit"); + let iterate = build_iterate_variant().expect("build iterate circuit"); + + let mut recursive_circuit = recursive.handle; + let mut iterate_circuit = iterate.handle; + + recursive.edges_input.append(&mut workload.clone()); + iterate.edges_input.append(&mut workload.clone()); + recursive_circuit.transaction().unwrap(); + iterate_circuit.transaction().unwrap(); + + let recursive_closure = recursive.closure_output.concat().consolidate(); + let iterate_closure = iterate.closure_output.concat().consolidate(); + + assert_eq!( + recursive_closure, iterate_closure, + "recursive and iterate variants must agree on cyclic input" + ); + + recursive_closure.num_entries_shallow() +} + +/// The forest sizes (in terms of node counts) we benchmark. The transitive +/// closure of a strongly connected graph is roughly `nodes^2`, +/// so these stay modest. +const GRAPH_SIZES: &[NodeId] = &[100, 200, 400]; + +fn transitive_closure_benches(c: &mut Criterion) { + let mut group = c.benchmark_group("transitive-closure-cyclic"); + // Building a circuit spins up worker threads, so keep the sample count + // modest to bound the wall-clock time of the whole benchmark. + group.sample_size(20); + + for &nodes in GRAPH_SIZES { + let workload = generate_cyclic_graph(nodes, OUT_DEGREE); + let closure_size = assert_variants_agree(&workload); + let shape = format!("n{nodes}"); + println!( + "graph {shape}: {} edges, closure of {closure_size} paths", + workload.len() + ); + + // Report throughput in paths of the computed closure per second so the + // two variants are compared on the same unit of useful work. + group.throughput(Throughput::Elements(closure_size as u64)); + + group.bench_with_input( + BenchmarkId::new("recursive", &shape), + &workload, + |b, workload| { + b.iter_batched( + || { + ( + build_recursive_variant().expect("build recursive circuit"), + workload.clone(), + ) + }, + |(circuit, workload)| drive_once(circuit, workload), + BatchSize::PerIteration, + ); + }, + ); + + group.bench_with_input( + BenchmarkId::new("iterate", &shape), + &workload, + |b, workload| { + b.iter_batched( + || { + ( + build_iterate_variant().expect("build iterate circuit"), + workload.clone(), + ) + }, + |(circuit, workload)| drive_once(circuit, workload), + BatchSize::PerIteration, + ); + }, + ); + } + + group.finish(); +} + +criterion_group!(benches, transitive_closure_benches); +criterion_main!(benches); + +/// Computes the transitive closure of a cyclic graph using DBSP's +/// [`recursive` API](dbsp::circuit::ChildCircuit::recursive) plus a `Min` +/// aggregation. +fn build_recursive_variant() -> anyhow::Result { + let (handle, (edges_input, closure_output)) = + Runtime::init_circuit(WORKERS, move |root_circuit| { + let (edges, edges_input) = root_circuit.add_input_zset::(); + + let len_1 = edges.map_index(|Tup3(from, to, weight)| { + (Tup2(*from, *to), Tup4(*from, *to, *weight, 1)) + }); + + let closure = root_circuit.recursive(|child_circuit, closure: Accumulator| { + let edges = edges.delta0(child_circuit); + let len_1 = len_1.delta0(child_circuit); + + let closure = len_1 + .plus( + &closure + .map_index( + |(Tup2(_start, _end), Tup4(start, end, cum_weight, hopcnt))| { + (*end, Tup4(*start, *end, *cum_weight, *hopcnt)) + }, + ) + .join_index( + &edges.map_index(|Tup3(from, to, weight)| { + (*from, Tup3(*from, *to, *weight)) + }), + |_end_from, + Tup4(start, _end, cum_weight, hopcnt), + Tup3(_from, to, weight)| { + Some(( + Tup2(*start, *to), + Tup4(*start, *to, cum_weight + weight, hopcnt + 1), + )) + }, + ), + ) + // The `Min` aggregation keeps the shortest path per pair and, + // crucially, guarantees termination on cyclic graphs. + .aggregate(Min) + // We explicitly mark the stream as distinct to avoid the + // implicit distinct because the aggregation step already + // ensures distinctness. + .mark_distinct(); + + Ok(closure) + })?; + + Ok((edges_input, closure.accumulate_output())) + })?; + + Ok(TransClosureCircuit { + handle, + edges_input, + closure_output, + }) +} + +/// Upper bound on nested iterations before the iterate variant gives up waiting +/// for a fixed point. +const MAX_ITERATIONS: usize = 4096; + +/// Computes the transitive closure of a cyclic graph using DBSP's lower level +/// [`iterate` API](dbsp::circuit::Circuit::iterate) plus a `Min` aggregation. +/// +/// It uses a hybrid termination check: Either stop once the subcircuit +/// reaches a fixed point, or after [`MAX_ITERATIONS`] steps if it never does. +fn build_iterate_variant() -> anyhow::Result { + let (handle, (edges_input, closure_output)) = + Runtime::init_circuit(WORKERS, move |root_circuit| { + let (edges, edges_input) = root_circuit.add_input_zset::(); + + let len_1 = edges.map_index(|Tup3(from, to, weight)| { + (Tup2(*from, *to), Tup4(*from, *to, *weight, 1)) + }); + + let closure = root_circuit.iterate(|child_circuit| { + let edges = edges.delta0(child_circuit); + let len_1 = len_1.delta0(child_circuit); + + let (closure_frontier, closure_frontier_feedback) = child_circuit + .add_feedback(Z1::new(OrdIndexedZSet::::default())); + + let new_paths = len_1 + .plus( + &closure_frontier + .map_index( + |(Tup2(_start, _end), Tup4(start, end, cum_weight, hopcnt))| { + (*end, Tup4(*start, *end, *cum_weight, *hopcnt)) + }, + ) + .join_index( + &edges.map_index(|Tup3(from, to, weight)| { + (*from, Tup3(*from, *to, *weight)) + }), + |_end_from, + Tup4(start, _end, cum_weight, hopcnt), + Tup3(_from, to, weight)| { + Some(( + Tup2(*start, *to), + Tup4(*start, *to, cum_weight + weight, hopcnt + 1), + )) + }, + ), + ) + // No `distinct` here (not even behind the scenes); the `Min` + // aggregation alone deduplicates and stops the iteration. + .aggregate(Min); + + closure_frontier_feedback.connect(&new_paths); + + let closure = Stream::dyn_integrate_trace( + &new_paths.inner(), + &BatchReaderFactories::new::(), + ); + + let child_circuit = child_circuit.clone(); + let consensus = Consensus::new("iterate"); + let step = Cell::new(0usize); + let termination_check = async move || { + let step_count = step.get() + 1; + step.set(step_count); + + // The step bound is data-independent, hence identical on every worker; + // the fixed-point check is per-worker. + let fixedpoint = child_circuit.is_fixedpoint(0); + let iteration_ceiling = step_count >= MAX_ITERATIONS; + let done = consensus.check(fixedpoint || iteration_ceiling).await?; + if done { + // Reset the step counter for the next transaction. + step.set(0); + } + + Ok(done) + }; + + Ok((termination_check, closure.export())) + })?; + + let closure = Stream::dyn_consolidate( + &closure, + &BatchReaderFactories::new::(), + ) + .typed::>() + .accumulate_output(); + + Ok((edges_input, closure)) + })?; + + Ok(TransClosureCircuit { + handle, + edges_input, + closure_output, + }) +} diff --git a/crates/dbsp/src/circuit.rs b/crates/dbsp/src/circuit.rs index dc5c53ac0a0..0abde9ad185 100644 --- a/crates/dbsp/src/circuit.rs +++ b/crates/dbsp/src/circuit.rs @@ -47,8 +47,8 @@ pub use dbsp_handle::{ splitter_output_first_chunk_size, }; pub use runtime::{ - Error as RuntimeError, LocalStore, LocalStoreMarker, Runtime, RuntimeHandle, WeakRuntime, - WorkerLocation, WorkerLocations, + Consensus, Error as RuntimeError, LocalStore, LocalStoreMarker, Runtime, RuntimeHandle, + WeakRuntime, WorkerLocation, WorkerLocations, }; pub use schedule::Error as SchedulerError; diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs index 8129d910204..b17ea080051 100644 --- a/crates/dbsp/src/circuit/circuit_builder.rs +++ b/crates/dbsp/src/circuit/circuit_builder.rs @@ -2762,6 +2762,22 @@ pub trait Circuit: CircuitBase + Clone + WithClock { F: FnOnce(&mut IterativeCircuit) -> Result, S: Scheduler + 'static; + /// Returns `true` if every operator in scope `scope` has reached a fixed + /// point, i.e., a state in which its output stays constant as long as its + /// input does. + /// + /// This is the same condition [`fixedpoint`](Self::fixedpoint) uses to stop + /// iterating. Unlike an emptiness test on an output stream, it accounts for + /// state still buffered inside operators (for example, results a join has + /// precomputed for future nested timestamps). Custom iteration built on + /// [`iterate`](Self::iterate) can call this from its termination check to + /// detect convergence, typically combining the per-worker results across + /// workers via [`Consensus`]. See the documentation of [`Scope`] for the + /// meaning of the parameter but passing `0` refers to the current circuit. + fn is_fixedpoint(&self, scope: Scope) -> bool { + self.check_fixedpoint(scope) + } + /// Make the contents of `parent_stream` available in the nested circuit /// via an [`ImportOperator`]. /// diff --git a/crates/dbsp/src/circuit/runtime.rs b/crates/dbsp/src/circuit/runtime.rs index aa520c315b4..906c2369e85 100644 --- a/crates/dbsp/src/circuit/runtime.rs +++ b/crates/dbsp/src/circuit/runtime.rs @@ -1354,9 +1354,27 @@ impl Runtime { } } -/// A synchronization primitive that allows multiple threads within a runtime to agree -/// when a condition is satisfied. -pub(crate) struct Consensus(Broadcast); +/// A synchronization primitive that allows multiple threads within a runtime to +/// agree when a condition is satisfied. +/// +/// Each worker submits a local vote via [`check`](Self::check); the call +/// returns `true` only when every worker in the runtime votes `true`. It is +/// the building block for combining per-worker termination decisions (for +/// example, per-worker fixed-point status) into a single, runtime-wide +/// decision when driving a nested circuit built on +/// [`Circuit::iterate`](super::circuit_builder::Circuit::iterate). +/// +/// # Safety +/// +/// Callers must respect the following invariants: +/// +/// 1. Every worker must call [`check`](Self::check) the same number of times +/// and in the same order relative to other ongoing `Consensus` in the same +/// subcircuit. A worker that skips a call to [`check`](Self::check) will +/// stop its peers from making progress. +/// 2. [`Consensus`] must be constructed once per logical decision and reused +/// across clock ticks. +pub struct Consensus(Broadcast); impl Consensus { pub fn new(name: impl Display) -> Self { diff --git a/crates/dbsp/src/lib.rs b/crates/dbsp/src/lib.rs index c541fe65bc3..24d1bc06411 100644 --- a/crates/dbsp/src/lib.rs +++ b/crates/dbsp/src/lib.rs @@ -100,8 +100,8 @@ pub use crate::time::Timestamp; pub use algebra::{DynZWeight, ZWeight}; pub use circuit::{ - ChildCircuit, Circuit, CircuitHandle, DBSPHandle, NestedCircuit, RootCircuit, Runtime, - RuntimeError, SchedulerError, Stream, WeakRuntime, + ChildCircuit, Circuit, CircuitHandle, Consensus, DBSPHandle, NestedCircuit, RootCircuit, + Runtime, RuntimeError, SchedulerError, Stream, WeakRuntime, }; #[cfg(not(feature = "backend-mode"))] pub use operator::FilterMap; diff --git a/crates/dbsp/src/tutorial.rs b/crates/dbsp/src/tutorial.rs index a4caafd4f24..d87b81a0bfa 100644 --- a/crates/dbsp/src/tutorial.rs +++ b/crates/dbsp/src/tutorial.rs @@ -2021,8 +2021,8 @@ //! We have two execution contexts: a [root circuit](`RootCircuit`) //! and a [child circuit](`crate::NestedCircuit`). //! The root circuit is the one that is built by the parameter to the -//! [`Runtime::init_circuit`] function. The child circuit is defined by the parameter to -//! the [`ChildCircuit<()>::recursive`](`crate::ChildCircuit::recursive`) function. +//! [`Runtime::init_circuit`] function. The child circuit is defined by the +//! parameter to the [`ChildCircuit::recursive`] function. //! We also make use of the [`delta0`](`crate::operator::Delta0`) operator to //! import streams from a parent circuit into a child circuit. //! Finally, we pick up the [incremental computation](#incremental-computation) @@ -2169,9 +2169,8 @@ //! //! To fix this issue, we have to change the code to stop iterating once the //! shortest path for each pair of nodes has been discovered. One approach to -//! achieve this is to group by each pair and use the -//! [`Min`](`crate::operator::dynamic::aggregate::Min`) aggregation operator -//! to only retain the shortest path for each pair. +//! achieve this is to group by each pair and use the [`aggregate::Min`] +//! aggregation operator to only retain the shortest path for each pair. //! Aggregation requires to index the stream, so there are more code changes //! required than shown here. You can find the full code in `tutorial11.rs` but //! the important changes take place within the child circuit: @@ -2292,7 +2291,7 @@ //! The examples on the transitive closure above demonstrate how to express //! self-recursive queries. DBSP also supports _mutually_-recursive queries. //! As this is a little bit more involved than what would fit in here, we defer -//! the interested reader to the example given in `tutorial12` which +//! the interested reader to the example given in `tutorial12.rs` which //! demonstrates mutual recursion in the domain of static program analysis. //! //! # Next steps @@ -2315,8 +2314,8 @@ //! let (mut circuit, (/*handles*/)) = Runtime::init_circuit(4, build_circuit)?; //! ``` use crate::{ - CircuitHandle, IndexedZSet, OrdPartitionedIndexedZSet, OutputHandle, RootCircuit, Runtime, - Stream, ZSet, ZSetHandle, - operator::{Aggregator, Max}, + ChildCircuit, CircuitHandle, IndexedZSet, OrdPartitionedIndexedZSet, OutputHandle, RootCircuit, + Runtime, Stream, ZSet, ZSetHandle, + operator::{Aggregator, Max, dynamic::aggregate}, utils::{Tup0, Tup1, Tup10}, };