differential_dataflow/algorithms/graphs/
bfs.rs

1//! Breadth-first distance labeling.
2
3use std::hash::Hash;
4
5use timely::progress::Timestamp;
6
7use crate::{VecCollection, ExchangeData};
8use crate::operators::*;
9use crate::lattice::Lattice;
10
11/// Returns pairs (node, dist) indicating distance of each node from a root.
12pub fn bfs<'scope, T, N>(edges: VecCollection<'scope, T, (N,N)>, roots: VecCollection<'scope, T, N>) -> VecCollection<'scope, T, (N,u32)>
13where
14    T: Timestamp + Lattice,
15    N: ExchangeData+Hash,
16{
17    let edges = edges.arrange_by_key();
18    bfs_arranged(edges, roots)
19}
20
21use crate::trace::{BatchCursor, Cursor, Navigable, TraceReader};
22use crate::operators::arrange::Arranged;
23
24/// Returns pairs (node, dist) indicating distance of each node from a root.
25pub fn bfs_arranged<'scope, N, Tr>(edges: Arranged<'scope, Tr>, roots: VecCollection<'scope, Tr::Time, N>) -> VecCollection<'scope, Tr::Time, (N, u32)>
26where
27    N: ExchangeData+Hash,
28    Tr: TraceReader<Batch: Navigable>+Clone+'static,
29    for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>,
30{
31    // initialize roots as reaching themselves at distance 0
32    let nodes = roots.map(|x| (x, 0));
33
34    // repeatedly update minimal distances each node can be reached from each root
35    nodes.clone().iterate(|scope, inner| {
36
37        let edges = edges.enter(scope);
38        let nodes = nodes.enter(scope);
39
40        inner.join_core(edges, |_k,l,d| Some((d.clone(), l+1)))
41             .concat(nodes)
42             .reduce(|_, s, t| t.push((*s[0].0, 1)))
43     })
44}
differential_dataflow/algorithms/graphs/bfs.rs

differential_dataflow/algorithms/graphs/
bfs.rs
