differential_dataflow/algorithms/graphs/
scc.rs

1//! Strongly connected component structure.
2
3use std::mem;
4use std::hash::Hash;
5
6use timely::progress::Timestamp;
7
8use crate::{VecCollection, ExchangeData};
9use crate::lattice::Lattice;
10use crate::difference::{Abelian, Multiply};
11
12use super::propagate::propagate_at;
13
14/// Returns the subset of edges in the same strongly connected component.
15pub fn strongly_connected<'scope, T, N, R>(graph: VecCollection<'scope, T, (N,N), R>) -> VecCollection<'scope, T, (N,N), R>
16where
17    T: Timestamp + Lattice + Hash,
18    N: ExchangeData + Hash,
19    R: ExchangeData + Abelian,
20    R: Multiply<R, Output=R>,
21    R: From<i8>
22{
23    strongly_connected_at(graph, |_| 0)
24}
25
26/// Returns the subset of edges in the same strongly connected component.
27///
28/// This variant introduces node labels in rounds indicated by `logic`, as in `propagate_at`:
29/// small labels can complete their propagation before larger labels are introduced, which can
30/// substantially reduce the total work performed.
31pub fn strongly_connected_at<'scope, T, N, R, F>(graph: VecCollection<'scope, T, (N,N), R>, logic: F) -> VecCollection<'scope, T, (N,N), R>
32where
33    T: Timestamp + Lattice + Hash,
34    N: ExchangeData + Hash,
35    R: ExchangeData + Abelian,
36    R: Multiply<R, Output=R>,
37    R: From<i8>,
38    F: Fn(&N)->u64+Clone+'static,
39{
40    use timely::order::Product;
41    let outer = graph.scope();
42    outer.scoped::<Product<_, usize>,_,_>("StronglyConnected", |scope| {
43        // Bring in edges and transposed edges.
44        let edges = graph.enter(scope);
45        let trans = edges.clone().map_in_place(|x| mem::swap(&mut x.0, &mut x.1));
46        // Create a new variable that will be intra-scc edges.
47        use crate::operators::iterate::Variable;
48        let (variable, inner) = Variable::new_from(edges.clone(), Product::new(Default::default(), 1));
49
50        let result = trim_edges(trim_edges(inner, edges, logic.clone()), trans, logic);
51        variable.set(result.clone());
52        result.leave(outer)
53    })
54}
55
56fn trim_edges<'scope, T, N, R, F>(cycle: VecCollection<'scope, T, (N,N), R>, edges: VecCollection<'scope, T, (N,N), R>, logic: F)
57    -> VecCollection<'scope, T, (N,N), R>
58where
59    T: Timestamp + Lattice + Hash,
60    N: ExchangeData + Hash,
61    R: ExchangeData + Abelian,
62    R: Multiply<R, Output=R>,
63    R: From<i8>,
64    F: Fn(&N)->u64+Clone+'static,
65{
66    let outer = edges.inner.scope();
67    outer.region_named("TrimEdges", |region| {
68        let cycle = cycle.enter_region(region);
69        let edges = edges.enter_region(region);
70
71        let nodes = edges.clone()
72                         .map_in_place(|x| x.0 = x.1.clone())
73                         .consolidate();
74
75        let labels = propagate_at(cycle, nodes, logic).arrange_by_key();
76
77        edges.arrange_by_key()
78             .join_core(labels.clone(), |e1,e2,l1| [(e2.clone(),(e1.clone(),l1.clone()))])
79             .arrange_by_key()
80             .join_core(labels, |e2,(e1,l1),l2| [((e1.clone(),e2.clone()),(l1.clone(),l2.clone()))])
81             .filter(|(_,(l1,l2))| l1 == l2)
82             .map(|((x1,x2),_)| (x2,x1))
83             .leave_region(outer)
84    })
85}
differential_dataflow/algorithms/graphs/scc.rs

differential_dataflow/algorithms/graphs/
scc.rs
