differential_dataflow/algorithms/graphs/
bijkstra.rs

1//! Bi-directional Dijkstra distance labeling.
2
3use std::hash::Hash;
4
5use timely::order::Product;
6use timely::progress::Timestamp;
7
8use crate::{VecCollection, ExchangeData};
9use crate::lattice::Lattice;
10use crate::operators::iterate::Variable;
11
12/// Returns the subset of `goals` that can reach each other in `edges`, with distance.
13///
14/// This method performs bidirectional search, from both ends of each goal in forward
15/// and reverse direction, for the sources and targets respectively. Each search can
16/// examine a fraction of the graph before meeting, and multiple searches can be managed
17/// concurrently.
18///
19/// Goals that cannot reach from the source to the target are relatively expensive, as
20/// the entire graph must be explored to confirm this. A graph connectivity pre-filter
21/// could be good insurance here.
22pub fn bidijkstra<'scope, T, N>(edges: VecCollection<'scope, T, (N,N)>, goals: VecCollection<'scope, T, (N,N)>) -> VecCollection<'scope, T, ((N,N), u32)>
23where
24    T: Timestamp + Lattice,
25    N: ExchangeData+Hash,
26{
27    let forward = edges.clone().arrange_by_key();
28    let reverse = edges.map(|(x,y)| (y,x)).arrange_by_key();
29    bidijkstra_arranged(forward, reverse, goals)
30}
31
32use crate::trace::{BatchCursor, Cursor, Navigable, TraceReader};
33use crate::operators::arrange::Arranged;
34
35/// Bi-directional Dijkstra search using arranged forward and reverse edge collections.
36pub fn bidijkstra_arranged<'scope, N, Tr>(
37    forward: Arranged<'scope, Tr>,
38    reverse: Arranged<'scope, Tr>,
39    goals: VecCollection<'scope, Tr::Time, (N,N)>
40) -> VecCollection<'scope, Tr::Time, ((N,N), u32)>
41where
42    N: ExchangeData+Hash,
43    Tr: TraceReader<Batch: Navigable>+Clone+'static,
44    for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>,
45{
46    let outer = forward.stream.scope();
47    outer.iterative::<u64,_,_>(|inner| {
48
49            let forward_edges = forward.enter(inner);
50            let reverse_edges = reverse.enter(inner);
51
52        // Our plan is to start evolving distances from both sources and destinations.
53        // The evolution from a source or destination should continue as long as there
54        // is a corresponding destination or source that has not yet been reached.
55
56        // forward and reverse (node, (root, dist))
57        let (forward_bind, forward) = Variable::new_from(goals.clone().map(|(x,_)| (x.clone(),(x.clone(),0))).enter(inner), Product::new(Default::default(), 1));
58        let (reverse_bind, reverse) = Variable::new_from(goals.clone().map(|(_,y)| (y.clone(),(y.clone(),0))).enter(inner), Product::new(Default::default(), 1));
59
60        forward.clone().map(|_| ()).consolidate().inspect(|x| println!("forward: {:?}", x));
61        reverse.clone().map(|_| ()).consolidate().inspect(|x| println!("reverse: {:?}", x));
62
63        let goals = goals.enter(inner);
64        // let edges = edges.enter(inner);
65
66        // Let's determine which (src, dst) pairs are ready to return.
67        //
68        //   done(src, dst) := forward(src, med), reverse(dst, med), goal(src, dst).
69        //
70        // This is a cyclic join, which should scare us a bunch.
71        let reached =
72        forward
73            .clone()
74            .join_map(reverse.clone(), |_, (src,d1), (dst,d2)| ((src.clone(), dst.clone()), *d1 + *d2))
75            .reduce(|_key, s, t| t.push((*s[0].0, 1)))
76            .semijoin(goals.clone());
77
78        let active =
79        reached
80            .clone()
81            .negate()
82            .map(|(srcdst,_)| srcdst)
83            .concat(goals)
84            .consolidate();
85
86        // Let's expand out forward queries that are active.
87        let forward_active = active.clone().map(|(x,_y)| x).distinct();
88        let forward_next =
89        forward
90            .clone()
91            .map(|(med, (src, dist))| (src, (med, dist)))
92            .semijoin(forward_active)
93            .map(|(src, (med, dist))| (med, (src, dist)))
94            .join_core(forward_edges, |_med, (src, dist), next| Some((next.clone(), (src.clone(), *dist+1))))
95            .concat(forward)
96            .map(|(next, (src, dist))| ((next, src), dist))
97            .reduce(|_key, s, t| t.push((*s[0].0, 1)))
98            .map(|((next, src), dist)| (next, (src, dist)));
99
100        forward_next.clone().map(|_| ()).consolidate().inspect(|x| println!("forward_next: {:?}", x));
101
102        forward_bind.set(forward_next);
103
104        // Let's expand out reverse queries that are active.
105        let reverse_active = active.map(|(_x,y)| y).distinct();
106        let reverse_next =
107        reverse
108            .clone()
109            .map(|(med, (rev, dist))| (rev, (med, dist)))
110            .semijoin(reverse_active)
111            .map(|(rev, (med, dist))| (med, (rev, dist)))
112            .join_core(reverse_edges, |_med, (rev, dist), next| Some((next.clone(), (rev.clone(), *dist+1))))
113            .concat(reverse)
114            .map(|(next, (rev, dist))| ((next, rev), dist))
115            .reduce(|_key, s, t| t.push((*s[0].0, 1)))
116            .map(|((next,rev), dist)| (next, (rev, dist)));
117
118        reverse_next.clone().map(|_| ()).consolidate().inspect(|x| println!("reverse_next: {:?}", x));
119
120        reverse_bind.set(reverse_next);
121
122        reached.leave(outer)
123    })
124}
differential_dataflow/algorithms/graphs/bijkstra.rs

differential_dataflow/algorithms/graphs/
bijkstra.rs
