timely/dataflow/operators/vec/
unordered_input.rs

1//! Create new `Streams` connected to external inputs.
2
3use crate::container::CapacityContainerBuilder;
4use crate::dataflow::operators::{ActivateCapability};
5use crate::dataflow::operators::core::{UnorderedInput as UnorderedInputCore, UnorderedHandle as UnorderedHandleCore};
6use crate::dataflow::{StreamVec, Scope};
7
8/// Create a new `StreamVec` and `Handle` through which to supply input.
9pub trait UnorderedInput<G: Scope> {
10    /// Create a new capability-based `StreamVec` and `Handle` through which to supply input. This
11    /// input supports multiple open epochs (timestamps) at the same time.
12    ///
13    /// The `new_unordered_input` method returns `((Handle, Capability), Stream)` where the `StreamVec` can be used
14    /// immediately for timely dataflow construction, `Handle` and `Capability` are later used to introduce
15    /// data into the timely dataflow computation.
16    ///
17    /// The `Capability` returned is for the default value of the timestamp type in use. The
18    /// capability can be dropped to inform the system that the input has advanced beyond the
19    /// capability's timestamp. To retain the ability to send, a new capability at a later timestamp
20    /// should be obtained first, via the `delayed` function for `Capability`.
21    ///
22    /// To communicate the end-of-input drop all available capabilities.
23    ///
24    /// # Examples
25    ///
26    /// ```
27    /// use std::sync::{Arc, Mutex};
28    ///
29    /// use timely::*;
30    /// use timely::dataflow::operators::*;
31    /// use timely::dataflow::operators::vec::UnorderedInput;
32    /// use timely::dataflow::operators::capture::Extract;
33    ///
34    /// // get send and recv endpoints, wrap send to share
35    /// let (send, recv) = ::std::sync::mpsc::channel();
36    /// let send = Arc::new(Mutex::new(send));
37    ///
38    /// timely::execute(Config::thread(), move |worker| {
39    ///
40    ///     // this is only to validate the output.
41    ///     let send = send.lock().unwrap().clone();
42    ///
43    ///     // create and capture the unordered input.
44    ///     let (mut input, mut cap) = worker.dataflow::<usize,_,_>(|scope| {
45    ///         let (input, stream) = scope.new_unordered_input();
46    ///         stream.capture_into(send);
47    ///         input
48    ///     });
49    ///
50    ///     // feed values 0..10 at times 0..10.
51    ///     for round in 0..10 {
52    ///         input.activate().session(&cap).give(round);
53    ///         cap = cap.delayed(&(round + 1));
54    ///         worker.step();
55    ///     }
56    /// }).unwrap();
57    ///
58    /// let extract = recv.extract();
59    /// for i in 0..10 {
60    ///     assert_eq!(extract[i], (i, vec![i]));
61    /// }
62    /// ```
63    fn new_unordered_input<D: 'static>(&mut self) -> ((UnorderedHandle<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamVec<G, D>);
64}
65
66
67impl<G: Scope> UnorderedInput<G> for G {
68    fn new_unordered_input<D: 'static>(&mut self) -> ((UnorderedHandle<G::Timestamp, D>, ActivateCapability<G::Timestamp>), StreamVec<G, D>) {
69        UnorderedInputCore::new_unordered_input(self)
70    }
71}
72
73/// An unordered handle specialized to vectors.
74pub type UnorderedHandle<T, D> = UnorderedHandleCore<T, CapacityContainerBuilder<Vec<D>>>;
timely/dataflow/operators/vec/unordered_input.rs

timely/dataflow/operators/vec/
unordered_input.rs
