differential_dataflow/operators/arrange/agent.rs
1//! Shared read access to a trace.
2
3use std::rc::{Rc, Weak};
4use std::cell::RefCell;
5use std::collections::VecDeque;
6
7use timely::dataflow::Scope;
8use timely::dataflow::operators::generic::{OperatorInfo, source};
9use timely::progress::Timestamp;
10use timely::progress::{Antichain, frontier::AntichainRef};
11use timely::dataflow::operators::CapabilitySet;
12
13use crate::trace::{Trace, TraceReader, BatchReader};
14
15use timely::scheduling::Activator;
16
17use super::{TraceWriter, TraceAgentQueueWriter, TraceAgentQueueReader, Arranged};
18use super::TraceReplayInstruction;
19
20use crate::trace::wrappers::frontier::{TraceFrontier, BatchFrontier};
21
22
23/// A `TraceReader` wrapper which can be imported into other dataflows.
24///
25/// The `TraceAgent` is the default trace type produced by `arranged`, and it can be extracted
26/// from the dataflow in which it was defined, and imported into other dataflows.
27pub struct TraceAgent<Tr: TraceReader> {
28 trace: Rc<RefCell<trace_box::TraceBox<Tr>>>,
29 queues: Weak<RefCell<Vec<TraceAgentQueueWriter<Tr>>>>,
30 logical_compaction: Antichain<Tr::Time>,
31 physical_compaction: Antichain<Tr::Time>,
32 temp_antichain: Antichain<Tr::Time>,
33
34 operator: OperatorInfo,
35 logging: Option<crate::logging::Logger>,
36}
37
38impl<Tr: TraceReader> TraceReader for TraceAgent<Tr> {
39
40 type Time = Tr::Time;
41 type Batch = Tr::Batch;
42
43 fn set_logical_compaction(&mut self, frontier: AntichainRef<Tr::Time>) {
44 // This method does not enforce that `frontier` is greater or equal to `self.logical_compaction`.
45 // Instead, it determines the joint consequences of both guarantees and moves forward with that.
46 crate::lattice::antichain_join_into(&self.logical_compaction.borrow()[..], &frontier[..], &mut self.temp_antichain);
47 self.trace.borrow_mut().adjust_logical_compaction(self.logical_compaction.borrow(), self.temp_antichain.borrow());
48 ::std::mem::swap(&mut self.logical_compaction, &mut self.temp_antichain);
49 self.temp_antichain.clear();
50 }
51 fn get_logical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> {
52 self.logical_compaction.borrow()
53 }
54 fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) {
55 // This method does not enforce that `frontier` is greater or equal to `self.physical_compaction`.
56 // Instead, it determines the joint consequences of both guarantees and moves forward with that.
57 crate::lattice::antichain_join_into(&self.physical_compaction.borrow()[..], &frontier[..], &mut self.temp_antichain);
58 self.trace.borrow_mut().adjust_physical_compaction(self.physical_compaction.borrow(), self.temp_antichain.borrow());
59 ::std::mem::swap(&mut self.physical_compaction, &mut self.temp_antichain);
60 self.temp_antichain.clear();
61 }
62 fn get_physical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> {
63 self.physical_compaction.borrow()
64 }
65 fn batches_through(&mut self, frontier: AntichainRef<'_, Tr::Time>) -> Option<Vec<Self::Batch>> {
66 self.trace.borrow_mut().trace.batches_through(frontier)
67 }
68 fn map_batches<F: FnMut(&Self::Batch)>(&self, f: F) { self.trace.borrow().trace.map_batches(f) }
69}
70
71impl<Tr: TraceReader> TraceAgent<Tr> {
72 /// Creates a new agent from a trace reader.
73 pub fn new(trace: Tr, operator: OperatorInfo, logging: Option<crate::logging::Logger>) -> (Self, TraceWriter<Tr>)
74 where
75 Tr: Trace,
76 {
77 let trace = Rc::new(RefCell::new(trace_box::TraceBox::new(trace)));
78 let queues = Rc::new(RefCell::new(Vec::new()));
79
80 if let Some(logging) = &logging {
81 logging.log(
82 crate::logging::TraceShare { operator: operator.global_id, diff: 1 }
83 );
84 }
85
86 let reader = TraceAgent {
87 trace: Rc::clone(&trace),
88 queues: Rc::downgrade(&queues),
89 logical_compaction: trace.borrow().logical_compaction.frontier().to_owned(),
90 physical_compaction: trace.borrow().physical_compaction.frontier().to_owned(),
91 temp_antichain: Antichain::new(),
92 operator,
93 logging,
94 };
95
96 let writer = TraceWriter::new(
97 vec![Tr::Time::minimum()],
98 Rc::downgrade(&trace),
99 queues,
100 );
101
102 (reader, writer)
103 }
104
105 /// Attaches a new shared queue to the trace.
106 ///
107 /// The queue is first populated with existing batches from the trace,
108 /// The queue will be immediately populated with existing historical batches from the trace, and until the reference
109 /// is dropped the queue will receive new batches as produced by the source `arrange` operator.
110 pub fn new_listener(&mut self, activator: Activator) -> TraceAgentQueueReader<Tr>
111 {
112 // create a new queue for progress and batch information.
113 let mut new_queue = VecDeque::new();
114
115 // add the existing batches from the trace
116 let mut upper = None;
117 self.trace
118 .borrow_mut()
119 .trace
120 .map_batches(|batch| {
121 new_queue.push_back(TraceReplayInstruction::Batch(batch.clone(), Some(Tr::Time::minimum())));
122 upper = Some(batch.upper().clone());
123 });
124
125 if let Some(upper) = upper {
126 new_queue.push_back(TraceReplayInstruction::Frontier(upper));
127 }
128
129 let reference = Rc::new((activator, RefCell::new(new_queue)));
130
131 // wraps the queue in a ref-counted ref cell and enqueue/return it.
132 if let Some(queue) = self.queues.upgrade() {
133 queue.borrow_mut().push(Rc::downgrade(&reference));
134 }
135 reference.0.activate();
136 reference
137 }
138
139 /// The [OperatorInfo] of the underlying Timely operator
140 pub fn operator(&self) -> &OperatorInfo {
141 &self.operator
142 }
143
144 /// Obtain a reference to the inner [`trace_box::TraceBox`]. It is the caller's obligation to maintain
145 /// the trace box and this trace agent's invariants. Specifically, it is undefined behavior
146 /// to mutate the trace box. Keeping strong references can prevent resource reclamation.
147 ///
148 /// This method is subject to changes and removal and should not be considered part of a stable
149 /// interface.
150 pub fn trace_box_unstable(&self) -> Rc<RefCell<trace_box::TraceBox<Tr>>> {
151 Rc::clone(&self.trace)
152 }
153}
154
155impl<Tr: TraceReader+'static> TraceAgent<Tr> {
156 /// Copies an existing collection into the supplied scope.
157 ///
158 /// This method creates an `Arranged` collection that should appear indistinguishable from applying `arrange`
159 /// directly to the source collection brought into the local scope. The only caveat is that the initial state
160 /// of the collection is its current state, and updates occur from this point forward. The historical changes
161 /// the collection experienced in the past are accumulated, and the distinctions from the initial collection
162 /// are no longer evident.
163 ///
164 /// The current behavior is that the introduced collection accumulates updates to some times less or equal
165 /// to `self.get_logical_compaction()`. There is *not* currently a guarantee that the updates are accumulated *to*
166 /// the frontier, and the resulting collection history may be weirdly partial until this point. In particular,
167 /// the historical collection may move through configurations that did not actually occur, even if eventually
168 /// arriving at the correct collection. This is probably a bug; although we get to the right place in the end,
169 /// the intermediate computation could do something that the original computation did not, like diverge.
170 ///
171 /// I would expect the semantics to improve to "updates are advanced to `self.get_logical_compaction()`", which
172 /// means the computation will run as if starting from exactly this frontier. It is not currently clear whose
173 /// responsibility this should be (the trace/batch should only reveal these times, or an operator should know
174 /// to advance times before using them).
175 ///
176 /// # Examples
177 ///
178 /// ```
179 /// use timely::Config;
180 /// use differential_dataflow::input::Input;
181 /// use differential_dataflow::trace::Trace;
182 /// use differential_dataflow::trace::implementations::{ValBuilder, ValSpine};
183 ///
184 /// ::timely::execute(Config::thread(), |worker| {
185 ///
186 /// // create a first dataflow
187 /// let mut trace = worker.dataflow::<u32,_,_>(|scope| {
188 /// // create input handle and collection.
189 /// scope.new_collection_from(0 .. 10).1
190 /// .arrange_by_self()
191 /// .trace
192 /// });
193 ///
194 /// // do some work.
195 /// worker.step();
196 /// worker.step();
197 ///
198 /// // create a second dataflow
199 /// worker.dataflow(move |scope| {
200 /// trace.import(scope)
201 /// .reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<_,_,_,_>,_,_>(
202 /// "Reduce",
203 /// |_key, src, dst| dst.push((*src[0].0, 1)),
204 /// |vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
205 /// )
206 /// .as_collection(|k,v| (k.clone(), v.clone()));
207 /// });
208 ///
209 /// }).unwrap();
210 /// ```
211 pub fn import<'scope>(&mut self, scope: Scope<'scope, Tr::Time>) -> Arranged<'scope, TraceAgent<Tr>>
212 {
213 self.import_named(scope, "ArrangedSource")
214 }
215
216 /// Same as `import`, but allows to name the source.
217 pub fn import_named<'scope>(&mut self, scope: Scope<'scope, Tr::Time>, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
218 {
219 // Drop ShutdownButton and return only the arrangement.
220 self.import_core(scope, name).0
221 }
222
223 /// Imports an arrangement into the supplied scope.
224 ///
225 /// # Examples
226 ///
227 /// ```
228 /// use timely::Config;
229 /// use timely::dataflow::ProbeHandle;
230 /// use timely::dataflow::operators::Probe;
231 /// use differential_dataflow::input::InputSession;
232 /// use differential_dataflow::trace::Trace;
233 ///
234 /// ::timely::execute(Config::thread(), |worker| {
235 ///
236 /// let mut input = InputSession::<_,(),isize>::new();
237 /// let mut probe = ProbeHandle::new();
238 ///
239 /// // create a first dataflow
240 /// let mut trace = worker.dataflow::<u32,_,_>(|scope| {
241 /// // create input handle and collection.
242 /// input.to_collection(scope)
243 /// .arrange_by_self()
244 /// .trace
245 /// });
246 ///
247 /// // do some work.
248 /// worker.step();
249 /// worker.step();
250 ///
251 /// // create a second dataflow
252 /// let mut shutdown = worker.dataflow(|scope| {
253 /// let (arrange, button) = trace.import_core(scope, "Import");
254 /// arrange.stream.probe_with(&mut probe);
255 /// button
256 /// });
257 ///
258 /// worker.step();
259 /// worker.step();
260 /// assert!(!probe.done());
261 ///
262 /// shutdown.press();
263 ///
264 /// worker.step();
265 /// worker.step();
266 /// assert!(probe.done());
267 ///
268 /// }).unwrap();
269 /// ```
270 pub fn import_core<'scope>(&mut self, scope: Scope<'scope, Tr::Time>, name: &str) -> (Arranged<'scope, TraceAgent<Tr>>, ShutdownButton<CapabilitySet<Tr::Time>>)
271 {
272 let trace = self.clone();
273
274 let mut shutdown_button = None;
275
276 let stream = {
277
278 let shutdown_button_ref = &mut shutdown_button;
279 source(scope, name, move |capability, info| {
280
281 let capabilities = Rc::new(RefCell::new(Some(CapabilitySet::new())));
282
283 let activator = scope.activator_for(Rc::clone(&info.address));
284 let queue = self.new_listener(activator);
285
286 let activator = scope.activator_for(info.address);
287 *shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));
288
289 capabilities.borrow_mut().as_mut().unwrap().insert(capability);
290
291 move |output| {
292
293 let mut capabilities = capabilities.borrow_mut();
294 if let Some(ref mut capabilities) = *capabilities {
295
296 let mut borrow = queue.1.borrow_mut();
297 for instruction in borrow.drain(..) {
298 match instruction {
299 TraceReplayInstruction::Frontier(frontier) => {
300 capabilities.downgrade(&frontier.borrow()[..]);
301 },
302 TraceReplayInstruction::Batch(batch, hint) => {
303 if let Some(time) = hint {
304 if !batch.is_empty() {
305 let delayed = capabilities.delayed(&time);
306 output.session(&delayed).give(batch);
307 }
308 }
309 }
310 }
311 }
312 }
313 }
314 })
315 };
316
317 (Arranged { stream, trace }, shutdown_button.unwrap())
318 }
319
320 /// Imports an arrangement into the supplied scope.
321 ///
322 /// This variant of import uses the `get_logical_compaction` to forcibly advance timestamps in updates.
323 ///
324 /// # Examples
325 ///
326 /// ```
327 /// use timely::Config;
328 /// use timely::progress::frontier::AntichainRef;
329 /// use timely::dataflow::ProbeHandle;
330 /// use timely::dataflow::operators::Probe;
331 /// use timely::dataflow::operators::Inspect;
332 /// use differential_dataflow::input::InputSession;
333 /// use differential_dataflow::trace::Trace;
334 /// use differential_dataflow::trace::TraceReader;
335 /// use differential_dataflow::input::Input;
336 ///
337 /// ::timely::execute(Config::thread(), |worker| {
338 ///
339 /// let mut probe = ProbeHandle::new();
340 ///
341 /// // create a first dataflow
342 /// let (mut handle, mut trace) = worker.dataflow::<u32,_,_>(|scope| {
343 /// // create input handle and collection.
344 /// let (handle, stream) = scope.new_collection();
345 /// let trace = stream.arrange_by_self().trace;
346 /// (handle, trace)
347 /// });
348 ///
349 /// handle.insert(0); handle.advance_to(1); handle.flush(); worker.step();
350 /// handle.remove(0); handle.advance_to(2); handle.flush(); worker.step();
351 /// handle.insert(1); handle.advance_to(3); handle.flush(); worker.step();
352 /// handle.remove(1); handle.advance_to(4); handle.flush(); worker.step();
353 /// handle.insert(0); handle.advance_to(5); handle.flush(); worker.step();
354 ///
355 /// trace.set_logical_compaction(AntichainRef::new(&[5]));
356 ///
357 /// // create a second dataflow
358 /// let mut shutdown = worker.dataflow(|scope| {
359 /// let (arrange, button) = trace.import_frontier(scope, "Import");
360 /// arrange
361 /// .as_collection(|k,v| (*k,*v))
362 /// .inner
363 /// .inspect(|(d,t,r)| {
364 /// assert!(t >= &5);
365 /// })
366 /// .probe_with(&mut probe);
367 ///
368 /// button
369 /// });
370 ///
371 /// worker.step();
372 /// worker.step();
373 /// assert!(!probe.done());
374 ///
375 /// shutdown.press();
376 ///
377 /// worker.step();
378 /// worker.step();
379 /// assert!(probe.done());
380 ///
381 /// }).unwrap();
382 /// ```
383 pub fn import_frontier<'scope>(&mut self, scope: Scope<'scope, Tr::Time>, name: &str) -> (Arranged<'scope, TraceFrontier<TraceAgent<Tr>>>, ShutdownButton<CapabilitySet<Tr::Time>>)
384 where
385 Tr: TraceReader,
386 {
387 // This frontier describes our only guarantee on the compaction frontier.
388 let since = self.get_logical_compaction().to_owned();
389 self.import_frontier_core(scope, name, since, Antichain::new())
390 }
391
392 /// Import a trace restricted to a specific time interval `[since, until)`.
393 ///
394 /// All updates present in the input trace will be first advanced to `since`, and then either emitted,
395 /// or if greater or equal to `until`, suppressed. Once all times are certain to be greater or equal
396 /// to `until` the operator capability will be dropped.
397 ///
398 /// Invoking this method with an `until` of `Antichain::new()` will perform no filtering, as the empty
399 /// frontier indicates the end of times.
400 pub fn import_frontier_core<'scope>(&mut self, scope: Scope<'scope, Tr::Time>, name: &str, since: Antichain<Tr::Time>, until: Antichain<Tr::Time>) -> (Arranged<'scope, TraceFrontier<TraceAgent<Tr>>>, ShutdownButton<CapabilitySet<Tr::Time>>)
401 where
402 Tr: TraceReader,
403 {
404 let trace = self.clone();
405 let trace = TraceFrontier::make_from(trace, since.borrow(), until.borrow());
406
407 let mut shutdown_button = None;
408
409 let stream = {
410
411 let shutdown_button_ref = &mut shutdown_button;
412 source(scope, name, move |capability, info| {
413
414 let capabilities = Rc::new(RefCell::new(Some(CapabilitySet::new())));
415
416 let activator = scope.activator_for(Rc::clone(&info.address));
417 let queue = self.new_listener(activator);
418
419 let activator = scope.activator_for(info.address);
420 *shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));
421
422 capabilities.borrow_mut().as_mut().unwrap().insert(capability);
423
424 move |output| {
425
426 let mut capabilities = capabilities.borrow_mut();
427 if let Some(ref mut capabilities) = *capabilities {
428 let mut borrow = queue.1.borrow_mut();
429 for instruction in borrow.drain(..) {
430 // If we have dropped the capabilities due to `until`, attempt no further work.
431 // Without the capabilities, we should soon be shut down (once this loop ends).
432 if !capabilities.is_empty() {
433 match instruction {
434 TraceReplayInstruction::Frontier(frontier) => {
435 if timely::PartialOrder::less_equal(&until, &frontier) {
436 // It might be nice to actively *drop* `capabilities`, but it seems
437 // complicated logically (i.e. we'd have to break out of the loop).
438 capabilities.downgrade(&[]);
439 } else {
440 capabilities.downgrade(&frontier.borrow()[..]);
441 }
442 },
443 TraceReplayInstruction::Batch(batch, hint) => {
444 if let Some(time) = hint {
445 if !batch.is_empty() {
446 let delayed = capabilities.delayed(&time);
447 output.session(&delayed).give(BatchFrontier::make_from(batch, since.borrow(), until.borrow()));
448 }
449 }
450 }
451 }
452 }
453 }
454 }
455 }
456 })
457 };
458
459 (Arranged { stream, trace }, shutdown_button.unwrap())
460 }
461}
462
463
464
465/// Wrapper than can drop shared references.
466pub struct ShutdownButton<T> {
467 reference: Rc<RefCell<Option<T>>>,
468 activator: Activator,
469}
470
471impl<T> ShutdownButton<T> {
472 /// Creates a new ShutdownButton.
473 pub fn new(reference: Rc<RefCell<Option<T>>>, activator: Activator) -> Self {
474 Self { reference, activator }
475 }
476 /// Push the shutdown button, dropping the shared objects.
477 pub fn press(&mut self) {
478 *self.reference.borrow_mut() = None;
479 self.activator.activate();
480 }
481}
482
483impl<Tr: TraceReader> Clone for TraceAgent<Tr> {
484 fn clone(&self) -> Self {
485
486 if let Some(logging) = &self.logging {
487 logging.log(
488 crate::logging::TraceShare { operator: self.operator.global_id, diff: 1 }
489 );
490 }
491
492 // increase counts for wrapped `TraceBox`.
493 let empty_frontier = Antichain::new();
494 self.trace.borrow_mut().adjust_logical_compaction(empty_frontier.borrow(), self.logical_compaction.borrow());
495 self.trace.borrow_mut().adjust_physical_compaction(empty_frontier.borrow(), self.physical_compaction.borrow());
496
497 TraceAgent {
498 trace: Rc::clone(&self.trace),
499 queues: Weak::clone(&self.queues),
500 logical_compaction: self.logical_compaction.clone(),
501 physical_compaction: self.physical_compaction.clone(),
502 operator: self.operator.clone(),
503 logging: self.logging.clone(),
504 temp_antichain: Antichain::new(),
505 }
506 }
507}
508
509impl<Tr: TraceReader> Drop for TraceAgent<Tr> {
510 fn drop(&mut self) {
511
512 if let Some(logging) = &self.logging {
513 logging.log(
514 crate::logging::TraceShare { operator: self.operator.global_id, diff: -1 }
515 );
516 }
517
518 // decrement borrow counts to remove all holds
519 let empty_frontier = Antichain::new();
520 self.trace.borrow_mut().adjust_logical_compaction(self.logical_compaction.borrow(), empty_frontier.borrow());
521 self.trace.borrow_mut().adjust_physical_compaction(self.physical_compaction.borrow(), empty_frontier.borrow());
522 }
523}
524
525/// A trace wrapper suitable for use through shared reference counted ownership.
526///
527/// The wrapper mainly accumulates the expressed compaction constraints from many,
528/// and presents their implications to the wrapped trace.
529pub mod trace_box {
530
531 use timely::progress::{frontier::{AntichainRef, MutableAntichain}};
532
533 use crate::trace::TraceReader;
534
535 /// A wrapper around a trace which tracks the frontiers of all referees.
536 ///
537 /// This is an internal type, unlikely to be useful to higher-level programs, but exposed just in case.
538 /// This type is equivalent to a `RefCell`, in that it wraps the mutable state that multiple referrers
539 /// may influence.
540 pub struct TraceBox<Tr: TraceReader> {
541 /// accumulated holds on times for advancement.
542 pub (crate) logical_compaction: MutableAntichain<Tr::Time>,
543 /// accumulated holds on times for distinction.
544 pub (crate) physical_compaction: MutableAntichain<Tr::Time>,
545 /// The wrapped trace.
546 pub (crate) trace: Tr,
547 }
548
549 impl<Tr: TraceReader> TraceBox<Tr> {
550 /// Moves an existing trace into a shareable trace wrapper.
551 ///
552 /// The trace may already exist and have non-initial advance and distinguish frontiers. The boxing
553 /// process will fish these out and make sure that they are used for the initial read capabilities.
554 pub fn new(mut trace: Tr) -> Self {
555
556 let mut logical_compaction = MutableAntichain::new();
557 logical_compaction.update_iter(trace.get_logical_compaction().iter().cloned().map(|t| (t,1)));
558 let mut physical_compaction = MutableAntichain::new();
559 physical_compaction.update_iter(trace.get_physical_compaction().iter().cloned().map(|t| (t,1)));
560
561 TraceBox {
562 logical_compaction,
563 physical_compaction,
564 trace,
565 }
566 }
567 /// Borrowed access to the underlying trace.
568 ///
569 /// This is used to inspect batches for purposes of resource accounting in external systems.
570 pub fn trace(&self) -> &Tr { &self.trace }
571 /// Replaces elements of `lower` with those of `upper`.
572 #[inline]
573 pub fn adjust_logical_compaction(&mut self, lower: AntichainRef<Tr::Time>, upper: AntichainRef<Tr::Time>) {
574 self.logical_compaction.update_iter(upper.iter().cloned().map(|t| (t,1)));
575 self.logical_compaction.update_iter(lower.iter().cloned().map(|t| (t,-1)));
576 self.trace.set_logical_compaction(self.logical_compaction.frontier());
577 }
578 /// Replaces elements of `lower` with those of `upper`.
579 #[inline]
580 pub fn adjust_physical_compaction(&mut self, lower: AntichainRef<Tr::Time>, upper: AntichainRef<Tr::Time>) {
581 self.physical_compaction.update_iter(upper.iter().cloned().map(|t| (t,1)));
582 self.physical_compaction.update_iter(lower.iter().cloned().map(|t| (t,-1)));
583 self.trace.set_physical_compaction(self.physical_compaction.frontier());
584 }
585 }
586
587}