differential_dataflow/algorithms/graphs/
bijkstra.rs1use std::hash::Hash;
4
5use timely::order::Product;
6use timely::dataflow::*;
7
8use crate::{Collection, ExchangeData};
9use crate::operators::*;
10use crate::lattice::Lattice;
11use crate::operators::iterate::Variable;
12
13pub fn bidijkstra<G, N>(edges: &Collection<G, (N,N)>, goals: &Collection<G, (N,N)>) -> Collection<G, ((N,N), u32)>
24where
25 G: Scope<Timestamp: Lattice+Ord>,
26 N: ExchangeData+Hash,
27{
28 use crate::operators::arrange::arrangement::ArrangeByKey;
29 let forward = edges.arrange_by_key();
30 let reverse = edges.map(|(x,y)| (y,x)).arrange_by_key();
31 bidijkstra_arranged(&forward, &reverse, goals)
32}
33
34use crate::trace::TraceReader;
35use crate::operators::arrange::Arranged;
36
37pub fn bidijkstra_arranged<G, N, Tr>(
39 forward: &Arranged<G, Tr>,
40 reverse: &Arranged<G, Tr>,
41 goals: &Collection<G, (N,N)>
42) -> Collection<G, ((N,N), u32)>
43where
44 G: Scope<Timestamp=Tr::Time>,
45 N: ExchangeData+Hash,
46 Tr: for<'a> TraceReader<Key<'a>=&'a N, Val<'a>=&'a N, Diff=isize>+Clone+'static,
47{
48 forward
49 .stream
50 .scope().iterative::<u64,_,_>(|inner| {
51
52 let forward_edges = forward.enter(inner);
53 let reverse_edges = reverse.enter(inner);
54
55 let forward = Variable::new_from(goals.map(|(x,_)| (x.clone(),(x.clone(),0))).enter(inner), Product::new(Default::default(), 1));
61 let reverse = Variable::new_from(goals.map(|(_,y)| (y.clone(),(y.clone(),0))).enter(inner), Product::new(Default::default(), 1));
62
63 forward.map(|_| ()).consolidate().inspect(|x| println!("forward: {:?}", x));
64 reverse.map(|_| ()).consolidate().inspect(|x| println!("reverse: {:?}", x));
65
66 let goals = goals.enter(inner);
67 let reached =
75 forward
76 .join_map(&reverse, |_, (src,d1), (dst,d2)| ((src.clone(), dst.clone()), *d1 + *d2))
77 .reduce(|_key, s, t| t.push((*s[0].0, 1)))
78 .semijoin(&goals);
79
80 let active =
81 reached
82 .negate()
83 .map(|(srcdst,_)| srcdst)
84 .concat(&goals)
85 .consolidate();
86
87 let forward_active = active.map(|(x,_y)| x).distinct();
89 let forward_next =
90 forward
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.map(|_| ()).consolidate().inspect(|x| println!("forward_next: {:?}", x));
101
102 forward.set(&forward_next);
103
104 let reverse_active = active.map(|(_x,y)| y).distinct();
106 let reverse_next =
107 reverse
108 .map(|(med, (rev, dist))| (rev, (med, dist)))
109 .semijoin(&reverse_active)
110 .map(|(rev, (med, dist))| (med, (rev, dist)))
111 .join_core(&reverse_edges, |_med, (rev, dist), next| Some((next.clone(), (rev.clone(), *dist+1))))
112 .concat(&reverse)
113 .map(|(next, (rev, dist))| ((next, rev), dist))
114 .reduce(|_key, s, t| t.push((*s[0].0, 1)))
115 .map(|((next,rev), dist)| (next, (rev, dist)));
116
117 reverse_next.map(|_| ()).consolidate().inspect(|x| println!("reverse_next: {:?}", x));
118
119 reverse.set(&reverse_next);
120
121 reached.leave()
122 })
123}