timely/dataflow/operators/vec/
flow_controlled.rs

1//! Methods to construct flow-controlled sources.
2
3use crate::order::TotalOrder;
4use crate::progress::timestamp::Timestamp;
5use crate::dataflow::operators::generic::operator::source;
6use crate::dataflow::operators::probe::Handle;
7use crate::dataflow::{StreamVec, Scope};
8
9/// Output of the input reading function for iterator_source.
10pub struct IteratorSourceInput<T: Clone, D: 'static, DI: IntoIterator<Item=D>, I: IntoIterator<Item=(T, DI)>> {
11    /// Lower bound on timestamps that can be emitted by this input in the future.
12    pub lower_bound: T,
13    /// Any `T: IntoIterator` of new input data in the form (time, data): time must be
14    /// monotonically increasing.
15    pub data: I,
16    /// A timestamp that represents the frontier that the probe should have
17    /// reached before the function is invoked again to ingest additional input.
18    pub target: T,
19}
20
21/// Construct a source that repeatedly calls the provided function to ingest input.
22///
23/// The function can return `None` to signal the end of the input.
24/// Otherwise, it should return a [`IteratorSourceInput`], where:
25/// * `lower_bound` is a lower bound on timestamps that can be emitted by this input in the future,
26///   `Default::default()` can be used if this isn't needed (the source will assume that
27///   the timestamps in `data` are monotonically increasing and will release capabilities
28///   accordingly);
29/// * `data` is any `T: IntoIterator` of new input data in the form (time, data): time must be
30///   monotonically increasing;
31/// * `target` is a timestamp that represents the frontier that the probe should have
32///   reached before the function is invoked again to ingest additional input.
33///   The function will receive the current lower bound of timestamps that can be inserted,
34///   `lower_bound`.
35///
36/// # Example
37/// ```rust
38/// use timely::dataflow::operators::vec::flow_controlled::{iterator_source, IteratorSourceInput};
39/// use timely::dataflow::operators::{probe, Probe, Inspect};
40///
41/// timely::execute_from_args(std::env::args(), |worker| {
42///     let mut input = (0u64..100000).peekable();
43///     worker.dataflow(|scope| {
44///         let mut probe_handle = probe::Handle::new();
45///         let probe_handle_2 = probe_handle.clone();
46///
47///         let mut next_t: u64 = 0;
48///         iterator_source(
49///             scope,
50///             "Source",
51///             move |prev_t| {
52///                 if let Some(first_x) = input.peek().cloned() {
53///                     next_t = first_x / 100 * 100;
54///                     Some(IteratorSourceInput {
55///                         lower_bound: Default::default(),
56///                         data: vec![
57///                             (next_t,
58///                              input.by_ref().take(10).map(|x| (/* "timestamp" */ x, x)).collect::<Vec<_>>())],
59///                         target: *prev_t,
60///                     })
61///                 } else {
62///                     None
63///                 }
64///             },
65///             probe_handle_2)
66///         .inspect_time(|t, d| eprintln!("@ {:?}: {:?}", t, d))
67///         .probe_with(&mut probe_handle);
68///     });
69/// }).unwrap();
70/// ```
71pub fn iterator_source<
72    'scope,
73    T: Timestamp,
74    D: 'static,
75    DI: IntoIterator<Item=D>,
76    I: IntoIterator<Item=(T, DI)>,
77    F: FnMut(&T)->Option<IteratorSourceInput<T, D, DI, I>>+'static>(
78        scope: Scope<'scope, T>,
79        name: &str,
80        mut input_f: F,
81        probe: Handle<T>,
82        ) -> StreamVec<'scope, T, D> where T: TotalOrder {
83
84    let mut target = T::minimum();
85    source(scope, name, |cap, info| {
86        let mut cap = Some(cap);
87        let activator = scope.activator_for(info.address);
88        move |output| {
89            cap = cap.take().and_then(|mut cap| {
90                loop {
91                    if !probe.less_than(&target) {
92                        if let Some(IteratorSourceInput {
93                             lower_bound,
94                             data,
95                             target: new_target,
96                         }) = input_f(cap.time()) {
97                            target = new_target;
98                            let mut has_data = false;
99                            for (t, ds) in data.into_iter() {
100                                cap = if cap.time() != &t { cap.delayed(&t) } else { cap };
101                                let mut session = output.session(&cap);
102                                session.give_iterator(ds.into_iter());
103                                has_data = true;
104                            }
105
106                            cap = if cap.time().less_than(&lower_bound) { cap.delayed(&lower_bound) } else { cap };
107                            if !has_data {
108                                break Some(cap);
109                            }
110                        } else {
111                            break None;
112                        }
113                    } else {
114                        break Some(cap);
115                    }
116                }
117            });
118
119            if cap.is_some() {
120                activator.activate();
121            }
122        }
123    })
124}
timely/dataflow/operators/vec/flow_controlled.rs

timely/dataflow/operators/vec/
flow_controlled.rs
