differential_dataflow/algorithms/graphs/
sequential.rs

1//! Sequential (non-concurrent) graph algorithms.
2
3use std::hash::Hash;
4
5use timely::dataflow::*;
6
7use crate::{Collection, ExchangeData};
8use crate::lattice::Lattice;
9use crate::operators::*;
10use crate::hashable::Hashable;
11
12fn _color<G, N>(edges: &Collection<G, (N,N)>) -> Collection<G,(N,Option<u32>)>
13where
14    G: Scope,
15    G::Timestamp: Lattice+Ord,
16    N: ExchangeData+Hash,
17{
18    // need some bogus initial values.
19    let start = edges.map(|(x,_y)| (x,u32::max_value()))
20                     .distinct();
21
22    // repeatedly apply color-picking logic.
23    sequence(&start, edges, |_node, vals| {
24
25        // look for the first absent positive integer.
26        // start at 1 in case we ever use NonZero<u32>.
27
28        (1u32 ..)
29            .find(|&i| vals.get(i as usize - 1).map(|x| *x.0) != Some(i))
30            .unwrap()
31    })
32}
33
34/// Applies `logic` to nodes sequentially, in order of node identifiers.
35///
36/// The `logic` function updates a node's state as a function of its
37/// neighbor states. It will only be called on complete input.
38///
39/// Internally, this method performs a fixed-point computation in which
40/// a node "fires" once all of its neighbors with lower identifier have
41/// fired, and we apply `logic` to the new state of lower neighbors and
42/// the old state (input) of higher neighbors.
43pub fn sequence<G, N, V, F>(
44    state: &Collection<G, (N,V)>,
45    edges: &Collection<G, (N,N)>,
46    logic: F) -> Collection<G, (N,Option<V>)>
47where
48    G: Scope,
49    G::Timestamp: Lattice+Hash+Ord,
50    N: ExchangeData+Hashable,
51    V: ExchangeData,
52    F: Fn(&N, &[(&V, isize)])->V+'static
53{
54
55    let _timer = ::std::time::Instant::now();
56
57    // start iteration with None messages for all.
58    state
59        .map(|(node, _state)| (node, None))
60        .iterate(|new_state| {
61            // immutable content: edges and initial state.
62            let edges = edges.enter(&new_state.scope());
63            let old_state = state.enter(&new_state.scope());
64                                 // .map(|x| (x.0, Some(x.1)));
65
66            // break edges into forward and reverse directions.
67            let forward = edges.filter(|edge| edge.0 < edge.1);
68            let reverse = edges.filter(|edge| edge.0 > edge.1);
69
70            // new state goes along forward edges, old state along reverse edges
71            let new_messages = new_state.join_map(&forward, |_k,v,d| (d.clone(),v.clone()));
72
73            let incomplete = new_messages.filter(|x| x.1.is_none()).map(|x| x.0).distinct();
74            let new_messages = new_messages.filter(|x| x.1.is_some()).map(|x| (x.0, x.1.unwrap()));
75
76            let old_messages = old_state.join_map(&reverse, |_k,v,d| (d.clone(),v.clone()));
77
78            let messages = new_messages.concat(&old_messages).antijoin(&incomplete);
79
80            // // determine who has incoming `None` messages, and suppress all of them.
81            // let incomplete = new_messages.filter(|x| x.1.is_none()).map(|x| x.0).distinct();
82
83            // merge messages; suppress computation if not all inputs available yet.
84            messages
85                // .concat(&old_messages)  // /-- possibly too clever: None if any inputs None.
86                // .antijoin(&incomplete)
87                .reduce(move |k, vs, t| t.push((Some(logic(k,vs)),1)))
88                .concat(&incomplete.map(|x| (x, None)))
89        })
90}
differential_dataflow/algorithms/graphs/sequential.rs

differential_dataflow/algorithms/graphs/
sequential.rs
