differential_dataflow/
input.rs

1//! Input sessions for simplified collection updates.
2//!
3//! Although users can directly manipulate timely dataflow streams as collection inputs,
4//! the `InputSession` type can make this more efficient and less error-prone. Specifically,
5//! the type batches up updates with their logical times and ships them with coarsened
6//! timely dataflow capabilities, exposing more concurrency to the operator implementations
7//! than are evident from the logical times, which appear to execute in sequence.
8
9use timely::progress::Timestamp;
10use timely::dataflow::operators::Input as TimelyInput;
11use timely::dataflow::operators::input::Handle;
12use timely::dataflow::scopes::ScopeParent;
13
14use crate::Data;
15use crate::difference::Semigroup;
16use crate::collection::{Collection, AsCollection};
17
18/// Create a new collection and input handle to control the collection.
19pub trait Input : TimelyInput {
20    /// Create a new collection and input handle to subsequently control the collection.
21    ///
22    /// # Examples
23    ///
24    /// ```
25    /// use timely::Config;
26    /// use differential_dataflow::input::Input;
27    ///
28    /// ::timely::execute(Config::thread(), |worker| {
29    ///
30    ///     let (mut handle, probe) = worker.dataflow::<(),_,_>(|scope| {
31    ///         // create input handle and collection.
32    ///         let (handle, data) = scope.new_collection();
33    ///         let probe = data.map(|x| x * 2)
34    ///                         .inspect(|x| println!("{:?}", x))
35    ///                         .probe();
36    ///         (handle, probe)
37    ///     });
38    ///
39    ///     handle.insert(1);
40    ///     handle.insert(5);
41    ///
42    /// }).unwrap();
43    /// ```
44    fn new_collection<D, R>(&mut self) -> (InputSession<<Self as ScopeParent>::Timestamp, D, R>, Collection<Self, D, R>)
45    where D: Data, R: Semigroup+'static;
46    /// Create a new collection and input handle from initial data.
47    ///
48    /// # Examples
49    ///
50    /// ```
51    /// use timely::Config;
52    /// use differential_dataflow::input::Input;
53    ///
54    /// ::timely::execute(Config::thread(), |worker| {
55    ///
56    ///     let (mut handle, probe) = worker.dataflow::<(),_,_>(|scope| {
57    ///         // create input handle and collection.
58    ///          let (handle, data) = scope.new_collection_from(0 .. 10);
59    ///          let probe = data.map(|x| x * 2)
60    ///                          .inspect(|x| println!("{:?}", x))
61    ///                          .probe();
62    ///          (handle, probe)
63    ///     });
64    ///
65    ///     handle.insert(1);
66    ///     handle.insert(5);
67    ///
68    /// }).unwrap();
69    /// ```
70    fn new_collection_from<I>(&mut self, data: I) -> (InputSession<<Self as ScopeParent>::Timestamp, I::Item, isize>, Collection<Self, I::Item, isize>)
71    where I: IntoIterator<Item: Data> + 'static;
72    /// Create a new collection and input handle from initial data.
73    ///
74    /// # Examples
75    ///
76    /// ```
77    /// use timely::Config;
78    /// use differential_dataflow::input::Input;
79    ///
80    /// ::timely::execute(Config::thread(), |worker| {
81    ///
82    ///     let (mut handle, probe) = worker.dataflow::<(),_,_>(|scope| {
83    ///         // create input handle and collection.
84    ///         let (handle, data) = scope.new_collection_from(0 .. 10);
85    ///         let probe = data.map(|x| x * 2)
86    ///                         .inspect(|x| println!("{:?}", x))
87    ///                         .probe();
88    ///         (handle, probe)
89    ///     });
90    ///
91    ///     handle.insert(1);
92    ///     handle.insert(5);
93    ///
94    /// }).unwrap();
95    /// ```
96    fn new_collection_from_raw<D, R, I>(&mut self, data: I) -> (InputSession<<Self as ScopeParent>::Timestamp, D, R>, Collection<Self, D, R>)
97    where I: IntoIterator<Item=(D,<Self as ScopeParent>::Timestamp,R)>+'static, D: Data, R: Semigroup+'static;
98}
99
100use crate::lattice::Lattice;
101impl<G: TimelyInput> Input for G where <G as ScopeParent>::Timestamp: Lattice {
102    fn new_collection<D, R>(&mut self) -> (InputSession<<G as ScopeParent>::Timestamp, D, R>, Collection<G, D, R>)
103    where
104        D: Data, R: Semigroup+'static,
105    {
106        let (handle, stream) = self.new_input();
107        (InputSession::from(handle), stream.as_collection())
108    }
109    fn new_collection_from<I>(&mut self, data: I) -> (InputSession<<G as ScopeParent>::Timestamp, I::Item, isize>, Collection<G, I::Item, isize>)
110    where I: IntoIterator+'static, I::Item: Data {
111        self.new_collection_from_raw(data.into_iter().map(|d| (d, <G::Timestamp as timely::progress::Timestamp>::minimum(), 1)))
112    }
113    fn new_collection_from_raw<D,R,I>(&mut self, data: I) -> (InputSession<<G as ScopeParent>::Timestamp, D, R>, Collection<G, D, R>)
114    where
115        D: Data,
116        R: Semigroup+'static,
117        I: IntoIterator<Item=(D,<Self as ScopeParent>::Timestamp,R)>+'static,
118    {
119        use timely::dataflow::operators::ToStream;
120
121        let (handle, stream) = self.new_input();
122        let source = data.to_stream(self).as_collection();
123
124        (InputSession::from(handle), stream.as_collection().concat(&source))
125    }}
126
127/// An input session wrapping a single timely dataflow capability.
128///
129/// Each timely dataflow message has a corresponding capability, which is a logical time in the
130/// timely dataflow system. Differential dataflow updates can happen at a much higher rate than
131/// timely dataflow's progress tracking infrastructure supports, because the logical times are
132/// promoted to data and updates are batched together. The `InputSession` type does this batching.
133///
134/// # Examples
135///
136/// ```
137/// use timely::Config;
138/// use differential_dataflow::input::Input;
139///
140/// ::timely::execute(Config::thread(), |worker| {
141///
142///     let (mut handle, probe) = worker.dataflow(|scope| {
143///         // create input handle and collection.
144///         let (handle, data) = scope.new_collection_from(0 .. 10);
145///         let probe = data.map(|x| x * 2)
146///                         .inspect(|x| println!("{:?}", x))
147///                         .probe();
148///         (handle, probe)
149///     });
150///
151///     handle.insert(3);
152///     handle.advance_to(1);
153///     handle.insert(5);
154///     handle.advance_to(2);
155///     handle.flush();
156///
157///     while probe.less_than(handle.time()) {
158///         worker.step();
159///     }
160///
161///     handle.remove(5);
162///     handle.advance_to(3);
163///     handle.flush();
164///
165///     while probe.less_than(handle.time()) {
166///         worker.step();
167///     }
168///
169/// }).unwrap();
170/// ```
171pub struct InputSession<T: Timestamp+Clone, D: Data, R: Semigroup+'static> {
172    time: T,
173    buffer: Vec<(D, T, R)>,
174    handle: Handle<T,(D,T,R)>,
175}
176
177impl<T: Timestamp+Clone, D: Data> InputSession<T, D, isize> {
178    /// Adds an element to the collection.
179    pub fn insert(&mut self, element: D) { self.update(element, 1); }
180    /// Removes an element from the collection.
181    pub fn remove(&mut self, element: D) { self.update(element,-1); }
182}
183
184// impl<T: Timestamp+Clone, D: Data> InputSession<T, D, i64> {
185//     /// Adds an element to the collection.
186//     pub fn insert(&mut self, element: D) { self.update(element, 1); }
187//     /// Removes an element from the collection.
188//     pub fn remove(&mut self, element: D) { self.update(element,-1); }
189// }
190
191// impl<T: Timestamp+Clone, D: Data> InputSession<T, D, i32> {
192//     /// Adds an element to the collection.
193//     pub fn insert(&mut self, element: D) { self.update(element, 1); }
194//     /// Removes an element from the collection.
195//     pub fn remove(&mut self, element: D) { self.update(element,-1); }
196// }
197
198impl<T: Timestamp+Clone, D: Data, R: Semigroup+'static> InputSession<T, D, R> {
199
200    /// Introduces a handle as collection.
201    pub fn to_collection<G: TimelyInput>(&mut self, scope: &mut G) -> Collection<G, D, R>
202    where
203        G: ScopeParent<Timestamp=T>,
204    {
205        scope
206            .input_from(&mut self.handle)
207            .as_collection()
208    }
209
210    /// Allocates a new input handle.
211    pub fn new() -> Self {
212        let handle: Handle<T,_> = Handle::new();
213        InputSession {
214            time: handle.time().clone(),
215            buffer: Vec::new(),
216            handle,
217        }
218    }
219
220    /// Creates a new session from a reference to an input handle.
221    pub fn from(handle: Handle<T,(D,T,R)>) -> Self {
222        InputSession {
223            time: handle.time().clone(),
224            buffer: Vec::new(),
225            handle,
226        }
227    }
228
229    /// Adds to the weight of an element in the collection.
230    pub fn update(&mut self, element: D, change: R) {
231        if self.buffer.len() == self.buffer.capacity() {
232            if !self.buffer.is_empty() {
233                self.handle.send_batch(&mut self.buffer);
234            }
235            // TODO : This is a fairly arbitrary choice; should probably use `Context::default_size()` or such.
236            self.buffer.reserve(1024);
237        }
238        self.buffer.push((element, self.time.clone(), change));
239    }
240
241    /// Adds to the weight of an element in the collection at a future time.
242    pub fn update_at(&mut self, element: D, time: T, change: R) {
243        assert!(self.time.less_equal(&time));
244        if self.buffer.len() == self.buffer.capacity() {
245            if !self.buffer.is_empty() {
246                self.handle.send_batch(&mut self.buffer);
247            }
248            // TODO : This is a fairly arbitrary choice; should probably use `Context::default_size()` or such.
249            self.buffer.reserve(1024);
250        }
251        self.buffer.push((element, time, change));
252    }
253
254    /// Forces buffered data into the timely dataflow input, and advances its time to match that of the session.
255    ///
256    /// It is important to call `flush` before expecting timely dataflow to report progress. Until this method is
257    /// called, all updates may still be in internal buffers and not exposed to timely dataflow. Once the method is
258    /// called, all buffers are flushed and timely dataflow is advised that some logical times are no longer possible.
259    pub fn flush(&mut self) {
260        self.handle.send_batch(&mut self.buffer);
261        if self.handle.epoch().less_than(&self.time) {
262            self.handle.advance_to(self.time.clone());
263        }
264    }
265
266    /// Advances the logical time for future records.
267    ///
268    /// Importantly, this method does **not** immediately inform timely dataflow of the change. This happens only when
269    /// the session is dropped or flushed. It is not correct to use this time as a basis for a computation's `step_while`
270    /// method unless the session has just been flushed.
271    pub fn advance_to(&mut self, time: T) {
272        assert!(self.handle.epoch().less_equal(&time));
273        assert!(&self.time.less_equal(&time));
274        self.time = time;
275    }
276
277    /// Reveals the current time of the session.
278    pub fn epoch(&self) -> &T { &self.time }
279    /// Reveals the current time of the session.
280    pub fn time(&self) -> &T { &self.time }
281
282    /// Closes the input, flushing and sealing the wrapped timely input.
283    pub fn close(self) { }
284}
285
286impl<T: Timestamp+Clone, D: Data, R: Semigroup+'static> Drop for InputSession<T, D, R> {
287    fn drop(&mut self) {
288        self.flush();
289    }
290}
differential_dataflow/input.rs

differential_dataflow/
input.rs
