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+'static, I::Item: Data;
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 D: Data, R: Semigroup+'static{
104 let (handle, stream) = self.new_input();
105 (InputSession::from(handle), stream.as_collection())
106 }
107 fn new_collection_from<I>(&mut self, data: I) -> (InputSession<<G as ScopeParent>::Timestamp, I::Item, isize>, Collection<G, I::Item, isize>)
108 where I: IntoIterator+'static, I::Item: Data {
109 self.new_collection_from_raw(data.into_iter().map(|d| (d, <G::Timestamp as timely::progress::Timestamp>::minimum(), 1)))
110 }
111 fn new_collection_from_raw<D,R,I>(&mut self, data: I) -> (InputSession<<G as ScopeParent>::Timestamp, D, R>, Collection<G, D, R>)
112 where
113 D: Data,
114 R: Semigroup+'static,
115 I: IntoIterator<Item=(D,<Self as ScopeParent>::Timestamp,R)>+'static,
116 {
117 use timely::dataflow::operators::ToStream;
118
119 let (handle, stream) = self.new_input();
120 let source = data.to_stream(self).as_collection();
121
122 (InputSession::from(handle), stream.as_collection().concat(&source))
123 }}
124
125/// An input session wrapping a single timely dataflow capability.
126///
127/// Each timely dataflow message has a corresponding capability, which is a logical time in the
128/// timely dataflow system. Differential dataflow updates can happen at a much higher rate than
129/// timely dataflow's progress tracking infrastructure supports, because the logical times are
130/// promoted to data and updates are batched together. The `InputSession` type does this batching.
131///
132/// # Examples
133///
134/// ```
135/// use timely::Config;
136/// use differential_dataflow::input::Input;
137///
138/// ::timely::execute(Config::thread(), |worker| {
139///
140/// let (mut handle, probe) = worker.dataflow(|scope| {
141/// // create input handle and collection.
142/// let (handle, data) = scope.new_collection_from(0 .. 10);
143/// let probe = data.map(|x| x * 2)
144/// .inspect(|x| println!("{:?}", x))
145/// .probe();
146/// (handle, probe)
147/// });
148///
149/// handle.insert(3);
150/// handle.advance_to(1);
151/// handle.insert(5);
152/// handle.advance_to(2);
153/// handle.flush();
154///
155/// while probe.less_than(handle.time()) {
156/// worker.step();
157/// }
158///
159/// handle.remove(5);
160/// handle.advance_to(3);
161/// handle.flush();
162///
163/// while probe.less_than(handle.time()) {
164/// worker.step();
165/// }
166///
167/// }).unwrap();
168/// ```
169pub struct InputSession<T: Timestamp+Clone, D: Data, R: Semigroup+'static> {
170 time: T,
171 buffer: Vec<(D, T, R)>,
172 handle: Handle<T,(D,T,R)>,
173}
174
175impl<T: Timestamp+Clone, D: Data> InputSession<T, D, isize> {
176 /// Adds an element to the collection.
177 pub fn insert(&mut self, element: D) { self.update(element, 1); }
178 /// Removes an element from the collection.
179 pub fn remove(&mut self, element: D) { self.update(element,-1); }
180}
181
182// impl<T: Timestamp+Clone, D: Data> InputSession<T, D, i64> {
183// /// Adds an element to the collection.
184// pub fn insert(&mut self, element: D) { self.update(element, 1); }
185// /// Removes an element from the collection.
186// pub fn remove(&mut self, element: D) { self.update(element,-1); }
187// }
188
189// impl<T: Timestamp+Clone, D: Data> InputSession<T, D, i32> {
190// /// Adds an element to the collection.
191// pub fn insert(&mut self, element: D) { self.update(element, 1); }
192// /// Removes an element from the collection.
193// pub fn remove(&mut self, element: D) { self.update(element,-1); }
194// }
195
196impl<T: Timestamp+Clone, D: Data, R: Semigroup+'static> InputSession<T, D, R> {
197
198 /// Introduces a handle as collection.
199 pub fn to_collection<G: TimelyInput>(&mut self, scope: &mut G) -> Collection<G, D, R>
200 where
201 G: ScopeParent<Timestamp=T>,
202 {
203 scope
204 .input_from(&mut self.handle)
205 .as_collection()
206 }
207
208 /// Allocates a new input handle.
209 pub fn new() -> Self {
210 let handle: Handle<T,_> = Handle::new();
211 InputSession {
212 time: handle.time().clone(),
213 buffer: Vec::new(),
214 handle,
215 }
216 }
217
218 /// Creates a new session from a reference to an input handle.
219 pub fn from(handle: Handle<T,(D,T,R)>) -> Self {
220 InputSession {
221 time: handle.time().clone(),
222 buffer: Vec::new(),
223 handle,
224 }
225 }
226
227 /// Adds to the weight of an element in the collection.
228 pub fn update(&mut self, element: D, change: R) {
229 if self.buffer.len() == self.buffer.capacity() {
230 if !self.buffer.is_empty() {
231 self.handle.send_batch(&mut self.buffer);
232 }
233 // TODO : This is a fairly arbitrary choice; should probably use `Context::default_size()` or such.
234 self.buffer.reserve(1024);
235 }
236 self.buffer.push((element, self.time.clone(), change));
237 }
238
239 /// Adds to the weight of an element in the collection at a future time.
240 pub fn update_at(&mut self, element: D, time: T, change: R) {
241 assert!(self.time.less_equal(&time));
242 if self.buffer.len() == self.buffer.capacity() {
243 if !self.buffer.is_empty() {
244 self.handle.send_batch(&mut self.buffer);
245 }
246 // TODO : This is a fairly arbitrary choice; should probably use `Context::default_size()` or such.
247 self.buffer.reserve(1024);
248 }
249 self.buffer.push((element, time, change));
250 }
251
252 /// Forces buffered data into the timely dataflow input, and advances its time to match that of the session.
253 ///
254 /// It is important to call `flush` before expecting timely dataflow to report progress. Until this method is
255 /// called, all updates may still be in internal buffers and not exposed to timely dataflow. Once the method is
256 /// called, all buffers are flushed and timely dataflow is advised that some logical times are no longer possible.
257 pub fn flush(&mut self) {
258 self.handle.send_batch(&mut self.buffer);
259 if self.handle.epoch().less_than(&self.time) {
260 self.handle.advance_to(self.time.clone());
261 }
262 }
263
264 /// Advances the logical time for future records.
265 ///
266 /// Importantly, this method does **not** immediately inform timely dataflow of the change. This happens only when
267 /// the session is dropped or flushed. It is not correct to use this time as a basis for a computation's `step_while`
268 /// method unless the session has just been flushed.
269 pub fn advance_to(&mut self, time: T) {
270 assert!(self.handle.epoch().less_equal(&time));
271 assert!(&self.time.less_equal(&time));
272 self.time = time;
273 }
274
275 /// Reveals the current time of the session.
276 pub fn epoch(&self) -> &T { &self.time }
277 /// Reveals the current time of the session.
278 pub fn time(&self) -> &T { &self.time }
279
280 /// Closes the input, flushing and sealing the wrapped timely input.
281 pub fn close(self) { }
282}
283
284impl<T: Timestamp+Clone, D: Data, R: Semigroup+'static> Drop for InputSession<T, D, R> {
285 fn drop(&mut self) {
286 self.flush();
287 }
288}