From da1cb14e2cbfe4f7cd8398dc71f70ea990b07d66 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 9 Mar 2020 10:22:08 -0400 Subject: [PATCH 1/5] add Lattice impl for Antichain --- src/lattice.rs | 67 +++++++++++++++++-- src/operators/arrange/writer.rs | 6 +- src/trace/implementations/merge_batcher.rs | 2 +- src/trace/implementations/ord.rs | 12 ++-- src/trace/implementations/spine_fueled_neu.rs | 6 +- 5 files changed, 76 insertions(+), 17 deletions(-) diff --git a/src/lattice.rs b/src/lattice.rs index 998cf9a2d..01408a796 100644 --- a/src/lattice.rs +++ b/src/lattice.rs @@ -5,10 +5,10 @@ //! `Lattice` trait, and all reasoning in operators are done it terms of `Lattice` methods. use timely::order::PartialOrder; -use timely::progress::{Timestamp, Antichain}; +use timely::progress::Antichain; /// A bounded partially ordered type supporting joins and meets. -pub trait Lattice : PartialOrder+Timestamp { +pub trait Lattice : PartialOrder { /// The smallest element greater than or equal to both arguments. /// @@ -194,8 +194,11 @@ implement_lattice!(i16, 0); implement_lattice!(i8, 0); implement_lattice!((), ()); -/// Given two slices representing minimal antichains, -/// returns the "smallest" minimal antichain "greater or equal" to them. +/// Returns the "smallest" minimal antichain "greater or equal" to both inputs. +/// +/// This method is primarily meant for cases where one cannot call use the methods +/// of `Antichain`'s `PartialOrder` implementation, such as when one has references +/// rather than owned antichains. /// /// # Examples /// @@ -223,3 +226,59 @@ pub fn antichain_join(one: &[T], other: &[T]) -> Antichain { } upper } + +/// Returns the "greatest" minimal antichain "less or equal" to both inputs. +/// +/// This method is primarily meant for cases where one cannot call use the methods +/// of `Antichain`'s `PartialOrder` implementation, such as when one has references +/// rather than owned antichains. +/// +/// # Examples +/// +/// ``` +/// # extern crate timely; +/// # extern crate differential_dataflow; +/// # use timely::PartialOrder; +/// # use timely::order::Product; +/// # use differential_dataflow::lattice::Lattice; +/// # use differential_dataflow::lattice::antichain_meet; +/// # fn main() { +/// +/// let f1 = &[Product::new(3, 7), Product::new(5, 6)]; +/// let f2 = &[Product::new(4, 6)]; +/// let meet = antichain_meet(f1, f2); +/// assert_eq!(meet.elements(), &[Product::new(3, 7), Product::new(4, 6)]); +/// # } +/// ``` +pub fn antichain_meet(one: &[T], other: &[T]) -> Antichain { + let mut upper = Antichain::new(); + for time1 in one { + upper.insert(time1.clone()); + } + for time2 in other { + upper.insert(time2.clone()); + } + upper +} + +impl Lattice for Antichain { + fn join(&self, other: &Self) -> Self { + let mut upper = Antichain::new(); + for time1 in self.elements() { + for time2 in other.elements() { + upper.insert(time1.join(time2)); + } + } + upper + } + fn meet(&self, other: &Self) -> Self { + let mut upper = Antichain::new(); + for time1 in self.elements() { + upper.insert(time1.clone()); + } + for time2 in other.elements() { + upper.insert(time2.clone()); + } + upper + } +} \ No newline at end of file diff --git a/src/operators/arrange/writer.rs b/src/operators/arrange/writer.rs index 248fdea79..d7a7def3d 100644 --- a/src/operators/arrange/writer.rs +++ b/src/operators/arrange/writer.rs @@ -22,7 +22,7 @@ use super::TraceReplayInstruction; pub struct TraceWriter where Tr: Trace, - Tr::Time: Lattice+Ord+Clone+std::fmt::Debug+'static, + Tr::Time: Lattice+Timestamp+Ord+Clone+std::fmt::Debug+'static, Tr::Batch: Batch, { /// Current upper limit. @@ -36,7 +36,7 @@ where impl TraceWriter where Tr: Trace, - Tr::Time: Lattice+Ord+Clone+std::fmt::Debug+'static, + Tr::Time: Lattice+Timestamp+Ord+Clone+std::fmt::Debug+'static, Tr::Batch: Batch, { /// Creates a new `TraceWriter`. @@ -105,7 +105,7 @@ where impl Drop for TraceWriter where Tr: Trace, - Tr::Time: Lattice+Ord+Clone+std::fmt::Debug+'static, + Tr::Time: Lattice+Timestamp+Ord+Clone+std::fmt::Debug+'static, Tr::Batch: Batch, { fn drop(&mut self) { diff --git a/src/trace/implementations/merge_batcher.rs b/src/trace/implementations/merge_batcher.rs index 9969417cf..cdbb772bc 100644 --- a/src/trace/implementations/merge_batcher.rs +++ b/src/trace/implementations/merge_batcher.rs @@ -19,7 +19,7 @@ impl Batcher for MergeBatcher where K: Ord+Clone, V: Ord+Clone, - T: Lattice+Ord+Clone, + T: Lattice+timely::progress::Timestamp+Ord+Clone, R: Semigroup, B: Batch, { diff --git a/src/trace/implementations/ord.rs b/src/trace/implementations/ord.rs index 4cf75ecc3..b62299ecf 100644 --- a/src/trace/implementations/ord.rs +++ b/src/trace/implementations/ord.rs @@ -78,7 +78,7 @@ impl Batch for OrdValBatch where K: Ord+Clone+'static, V: Ord+Clone+'static, - T: Lattice+Ord+Clone+::std::fmt::Debug+'static, + T: Lattice+timely::progress::Timestamp+Ord+Clone+::std::fmt::Debug+'static, R: Semigroup, O: OrdOffset, >::Error: Debug, >::Error: Debug { @@ -217,7 +217,7 @@ impl Merger> for OrdValMer where K: Ord+Clone+'static, V: Ord+Clone+'static, - T: Lattice+Ord+Clone+::std::fmt::Debug+'static, + T: Lattice+timely::progress::Timestamp+Ord+Clone+::std::fmt::Debug+'static, R: Semigroup, O: OrdOffset, >::Error: Debug, >::Error: Debug { @@ -360,7 +360,7 @@ impl Builder> for OrdValBu where K: Ord+Clone+'static, V: Ord+Clone+'static, - T: Lattice+Ord+Clone+::std::fmt::Debug+'static, + T: Lattice+timely::progress::Timestamp+Ord+Clone+::std::fmt::Debug+'static, R: Semigroup, O: OrdOffset, >::Error: Debug, >::Error: Debug { @@ -430,7 +430,7 @@ where impl Batch for OrdKeyBatch where K: Ord+Clone+'static, - T: Lattice+Ord+Clone+'static, + T: Lattice+timely::progress::Timestamp+Ord+Clone+'static, R: Semigroup, O: OrdOffset, >::Error: Debug, >::Error: Debug { @@ -536,7 +536,7 @@ where impl Merger> for OrdKeyMerger where K: Ord+Clone+'static, - T: Lattice+Ord+Clone+'static, + T: Lattice+timely::progress::Timestamp+Ord+Clone+'static, R: Semigroup, O: OrdOffset, >::Error: Debug, >::Error: Debug { @@ -681,7 +681,7 @@ where impl Builder> for OrdKeyBuilder where K: Ord+Clone+'static, - T: Lattice+Ord+Clone+'static, + T: Lattice+timely::progress::Timestamp+Ord+Clone+'static, R: Semigroup, O: OrdOffset, >::Error: Debug, >::Error: Debug { diff --git a/src/trace/implementations/spine_fueled_neu.rs b/src/trace/implementations/spine_fueled_neu.rs index 895b5fe90..733f99ffc 100644 --- a/src/trace/implementations/spine_fueled_neu.rs +++ b/src/trace/implementations/spine_fueled_neu.rs @@ -102,7 +102,7 @@ impl TraceReader for Spine where K: Ord+Clone, // Clone is required by `batch::advance_*` (in-place could remove). V: Ord+Clone, // Clone is required by `batch::advance_*` (in-place could remove). - T: Lattice+Ord+Clone+Debug, + T: Lattice+timely::progress::Timestamp+Ord+Clone+Debug, R: Semigroup, B: Batch+Clone+'static, { @@ -239,7 +239,7 @@ impl Trace for Spine where K: Ord+Clone, V: Ord+Clone, - T: Lattice+Ord+Clone+Debug, + T: Lattice+timely::progress::Timestamp+Ord+Clone+Debug, R: Semigroup, B: Batch+Clone+'static, { @@ -373,7 +373,7 @@ impl Spine where K: Ord+Clone, V: Ord+Clone, - T: Lattice+Ord+Clone+Debug, + T: Lattice+timely::progress::Timestamp+Ord+Clone+Debug, R: Semigroup, B: Batch, { From 89eac907a16c858eff1da0d2a1df8ada18cdd7e8 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 9 Mar 2020 15:08:12 -0400 Subject: [PATCH 2/5] update types to Antichain --- examples/cursors.rs | 9 +-- src/lattice.rs | 37 ++++++----- src/operators/arrange/agent.rs | 50 +++++++------- src/operators/arrange/arrangement.rs | 15 +++-- src/operators/arrange/mod.rs | 3 +- src/operators/arrange/writer.rs | 25 +++---- src/operators/count.rs | 5 +- src/operators/join.rs | 17 +++-- src/operators/reduce.rs | 9 ++- src/operators/threshold.rs | 5 +- src/trace/description.rs | 28 ++++---- src/trace/implementations/merge_batcher.rs | 14 ++-- src/trace/implementations/ord.rs | 36 +++++----- src/trace/implementations/spine_fueled_neu.rs | 65 +++++++++---------- src/trace/mod.rs | 64 +++++++++--------- src/trace/wrappers/enter.rs | 29 +++++---- src/trace/wrappers/enter_at.rs | 29 +++++---- src/trace/wrappers/filter.rs | 11 ++-- src/trace/wrappers/freeze.rs | 11 ++-- src/trace/wrappers/frontier.rs | 47 +++++++------- src/trace/wrappers/rc.rs | 52 +++++++-------- tests/import.rs | 3 +- tests/trace.rs | 9 +-- 23 files changed, 290 insertions(+), 283 deletions(-) diff --git a/examples/cursors.rs b/examples/cursors.rs index 41010bc4d..b91ff7ebf 100644 --- a/examples/cursors.rs +++ b/examples/cursors.rs @@ -38,6 +38,7 @@ use std::fmt::Debug; use std::collections::BTreeMap; use timely::dataflow::operators::probe::Handle; +use timely::progress::frontier::AntichainRef; use differential_dataflow::input::Input; use differential_dataflow::operators::arrange::ArrangeByKey; @@ -77,8 +78,8 @@ fn main() { graph.close(); for i in 1..rounds + 1 { /* Advance the trace frontier to enable trace compaction. */ - graph_trace.distinguish_since(&[i]); - graph_trace.advance_by(&[i]); + graph_trace.distinguish_since(AntichainRef::new(&[i])); + graph_trace.advance_by(AntichainRef::new(&[i])); worker.step_while(|| probe.less_than(&i)); dump_cursor(i, worker.index(), &mut graph_trace); } @@ -92,8 +93,8 @@ fn main() { } graph.advance_to(i); graph.flush(); - graph_trace.distinguish_since(&[i]); - graph_trace.advance_by(&[i]); + graph_trace.distinguish_since(AntichainRef::new(&[i])); + graph_trace.advance_by(AntichainRef::new(&[i])); worker.step_while(|| probe.less_than(graph.time())); dump_cursor(i, worker.index(), &mut graph_trace); } diff --git a/src/lattice.rs b/src/lattice.rs index 01408a796..c98af5763 100644 --- a/src/lattice.rs +++ b/src/lattice.rs @@ -5,7 +5,7 @@ //! `Lattice` trait, and all reasoning in operators are done it terms of `Lattice` methods. use timely::order::PartialOrder; -use timely::progress::Antichain; +use timely::progress::{Antichain, frontier::AntichainRef}; /// A bounded partially ordered type supporting joins and meets. pub trait Lattice : PartialOrder { @@ -118,10 +118,12 @@ pub trait Lattice : PartialOrder { /// # use differential_dataflow::lattice::Lattice; /// # fn main() { /// + /// use timely::progress::frontier::AntichainRef; + /// /// let time = Product::new(3, 7); /// let mut advanced = Product::new(3, 7); /// let frontier = vec![Product::new(4, 8), Product::new(5, 3)]; - /// advanced.advance_by(&frontier[..]); + /// advanced.advance_by(AntichainRef::new(&frontier[..])); /// /// // `time` and `advanced` are indistinguishable to elements >= an element of `frontier` /// for i in 0 .. 10 { @@ -138,10 +140,11 @@ pub trait Lattice : PartialOrder { /// # } /// ``` #[inline] - fn advance_by(&mut self, frontier: &[Self]) where Self: Sized { - if let Some(first) = frontier.get(0) { + fn advance_by(&mut self, frontier: AntichainRef) where Self: Sized { + let mut iter = frontier.iter(); + if let Some(first) = iter.next() { let mut result = self.join(first); - for f in &frontier[1..] { + for f in iter { result.meet_assign(&self.join(f)); } *self = result; @@ -196,9 +199,9 @@ implement_lattice!((), ()); /// Returns the "smallest" minimal antichain "greater or equal" to both inputs. /// -/// This method is primarily meant for cases where one cannot call use the methods -/// of `Antichain`'s `PartialOrder` implementation, such as when one has references -/// rather than owned antichains. +/// This method is primarily meant for cases where one cannot use the methods +/// of `Antichain`'s `PartialOrder` implementation, such as when one has only +/// references rather than owned antichains. /// /// # Examples /// @@ -214,7 +217,7 @@ implement_lattice!((), ()); /// let f1 = &[Product::new(3, 7), Product::new(5, 6)]; /// let f2 = &[Product::new(4, 6)]; /// let join = antichain_join(f1, f2); -/// assert_eq!(join.elements(), &[Product::new(4, 7), Product::new(5, 6)]); +/// assert_eq!(&*join.elements(), &[Product::new(4, 7), Product::new(5, 6)]); /// # } /// ``` pub fn antichain_join(one: &[T], other: &[T]) -> Antichain { @@ -229,9 +232,9 @@ pub fn antichain_join(one: &[T], other: &[T]) -> Antichain { /// Returns the "greatest" minimal antichain "less or equal" to both inputs. /// -/// This method is primarily meant for cases where one cannot call use the methods -/// of `Antichain`'s `PartialOrder` implementation, such as when one has references -/// rather than owned antichains. +/// This method is primarily meant for cases where one cannot use the methods +/// of `Antichain`'s `PartialOrder` implementation, such as when one has only +/// references rather than owned antichains. /// /// # Examples /// @@ -247,7 +250,7 @@ pub fn antichain_join(one: &[T], other: &[T]) -> Antichain { /// let f1 = &[Product::new(3, 7), Product::new(5, 6)]; /// let f2 = &[Product::new(4, 6)]; /// let meet = antichain_meet(f1, f2); -/// assert_eq!(meet.elements(), &[Product::new(3, 7), Product::new(4, 6)]); +/// assert_eq!(&*meet.elements(), &[Product::new(3, 7), Product::new(4, 6)]); /// # } /// ``` pub fn antichain_meet(one: &[T], other: &[T]) -> Antichain { @@ -264,8 +267,8 @@ pub fn antichain_meet(one: &[T], other: &[T]) -> Antichain impl Lattice for Antichain { fn join(&self, other: &Self) -> Self { let mut upper = Antichain::new(); - for time1 in self.elements() { - for time2 in other.elements() { + for time1 in self.elements().iter() { + for time2 in other.elements().iter() { upper.insert(time1.join(time2)); } } @@ -273,10 +276,10 @@ impl Lattice for Antichain { } fn meet(&self, other: &Self) -> Self { let mut upper = Antichain::new(); - for time1 in self.elements() { + for time1 in self.elements().iter() { upper.insert(time1.clone()); } - for time2 in other.elements() { + for time2 in other.elements().iter() { upper.insert(time2.clone()); } upper diff --git a/src/operators/arrange/agent.rs b/src/operators/arrange/agent.rs index 1b15aa61d..d53f18c7b 100644 --- a/src/operators/arrange/agent.rs +++ b/src/operators/arrange/agent.rs @@ -7,6 +7,7 @@ use std::collections::VecDeque; use timely::dataflow::Scope; use timely::dataflow::operators::generic::source; use timely::progress::Timestamp; +use timely::progress::{Antichain, frontier::AntichainRef}; use timely::dataflow::operators::CapabilitySet; use lattice::Lattice; @@ -33,8 +34,8 @@ where { trace: Rc>>, queues: Weak>>>, - advance: Vec, - through: Vec, + advance: Antichain, + through: Antichain, operator: ::timely::dataflow::operators::generic::OperatorInfo, logging: Option<::logging::Logger>, @@ -53,23 +54,23 @@ where type Batch = Tr::Batch; type Cursor = Tr::Cursor; - fn advance_by(&mut self, frontier: &[Tr::Time]) { - self.trace.borrow_mut().adjust_advance_frontier(&self.advance[..], frontier); + fn advance_by(&mut self, frontier: AntichainRef) { + self.trace.borrow_mut().adjust_advance_frontier(self.advance.elements(), frontier); self.advance.clear(); self.advance.extend(frontier.iter().cloned()); } - fn advance_frontier(&mut self) -> &[Tr::Time] { - &self.advance[..] + fn advance_frontier(&mut self) -> AntichainRef { + self.advance.elements() } - fn distinguish_since(&mut self, frontier: &[Tr::Time]) { - self.trace.borrow_mut().adjust_through_frontier(&self.through[..], frontier); + fn distinguish_since(&mut self, frontier: AntichainRef) { + self.trace.borrow_mut().adjust_through_frontier(self.through.elements(), frontier); self.through.clear(); self.through.extend(frontier.iter().cloned()); } - fn distinguish_frontier(&mut self) -> &[Tr::Time] { - &self.through[..] + fn distinguish_frontier(&mut self) -> AntichainRef { + self.through.elements() } - fn cursor_through(&mut self, frontier: &[Tr::Time]) -> Option<(Tr::Cursor, >::Storage)> { + fn cursor_through(&mut self, frontier: AntichainRef) -> Option<(Tr::Cursor, >::Storage)> { self.trace.borrow_mut().trace.cursor_through(frontier) } fn map_batches(&mut self, f: F) { self.trace.borrow_mut().trace.map_batches(f) } @@ -98,8 +99,8 @@ where let reader = TraceAgent { trace: trace.clone(), queues: Rc::downgrade(&queues), - advance: trace.borrow().advance_frontiers.frontier().to_vec(), - through: trace.borrow().through_frontiers.frontier().to_vec(), + advance: trace.borrow().advance_frontiers.frontier().to_owned(), + through: trace.borrow().through_frontiers.frontier().to_owned(), operator, logging, }; @@ -130,7 +131,7 @@ where .trace .map_batches(|batch| { new_queue.push_back(TraceReplayInstruction::Batch(batch.clone(), Some(::minimum()))); - upper = Some(batch.upper().to_vec()); + upper = Some(batch.upper().clone()); }); if let Some(upper) = upper { @@ -318,7 +319,7 @@ where for instruction in borrow.drain(..) { match instruction { TraceReplayInstruction::Frontier(frontier) => { - capabilities.downgrade(&frontier[..]); + capabilities.downgrade(&frontier.elements()[..]); }, TraceReplayInstruction::Batch(batch, hint) => { if let Some(time) = hint { @@ -349,6 +350,7 @@ where /// extern crate differential_dataflow; /// /// use timely::Configuration; + /// use timely::progress::frontier::AntichainRef; /// use timely::dataflow::ProbeHandle; /// use timely::dataflow::operators::Probe; /// use timely::dataflow::operators::Inspect; @@ -380,7 +382,7 @@ where /// handle.remove(1); handle.advance_to(4); handle.flush(); worker.step(); /// handle.insert(0); handle.advance_to(5); handle.flush(); worker.step(); /// - /// trace.advance_by(&[5]); + /// trace.advance_by(AntichainRef::new(&[5])); /// /// // create a second dataflow /// let mut shutdown = worker.dataflow(|scope| { @@ -421,14 +423,14 @@ where } /// Import a trace advanced to a specific frontier. - pub fn import_frontier_core(&mut self, scope: &G, name: &str, frontier:Vec) -> (Arranged>>, ShutdownButton>) + pub fn import_frontier_core(&mut self, scope: &G, name: &str, frontier: Vec) -> (Arranged>>, ShutdownButton>) where G: Scope, Tr::Time: Timestamp+ Lattice+Ord+Clone+'static, Tr: TraceReader, { let trace = self.clone(); - let trace = TraceFrontier::make_from(trace, &frontier[..]); + let trace = TraceFrontier::make_from(trace, AntichainRef::new(&frontier[..])); let mut shutdown_button = None; @@ -456,13 +458,13 @@ where for instruction in borrow.drain(..) { match instruction { TraceReplayInstruction::Frontier(frontier) => { - capabilities.downgrade(&frontier[..]); + capabilities.downgrade(&frontier.elements()[..]); }, TraceReplayInstruction::Batch(batch, hint) => { if let Some(time) = hint { if !batch.is_empty() { let delayed = capabilities.delayed(&time); - output.session(&delayed).give(BatchFrontier::make_from(batch, &frontier[..])); + output.session(&delayed).give(BatchFrontier::make_from(batch, AntichainRef::new(&frontier[..]))); } } } @@ -530,8 +532,8 @@ where } // increase counts for wrapped `TraceBox`. - self.trace.borrow_mut().adjust_advance_frontier(&[], &self.advance[..]); - self.trace.borrow_mut().adjust_through_frontier(&[], &self.through[..]); + self.trace.borrow_mut().adjust_advance_frontier(AntichainRef::new(&[]), self.advance.elements()); + self.trace.borrow_mut().adjust_through_frontier(AntichainRef::new(&[]), self.through.elements()); TraceAgent { trace: self.trace.clone(), @@ -558,7 +560,7 @@ where } // decrement borrow counts to remove all holds - self.trace.borrow_mut().adjust_advance_frontier(&self.advance[..], &[]); - self.trace.borrow_mut().adjust_through_frontier(&self.through[..], &[]); + self.trace.borrow_mut().adjust_advance_frontier(self.advance.elements(), AntichainRef::new(&[])); + self.trace.borrow_mut().adjust_through_frontier(self.through.elements(), AntichainRef::new(&[])); } } diff --git a/src/operators/arrange/arrangement.rs b/src/operators/arrange/arrangement.rs index 38ce8ea27..cabc36b13 100644 --- a/src/operators/arrange/arrangement.rs +++ b/src/operators/arrange/arrangement.rs @@ -23,7 +23,7 @@ use timely::dataflow::{Scope, Stream}; use timely::dataflow::operators::generic::Operator; use timely::dataflow::channels::pact::{ParallelizationContract, Pipeline, Exchange}; use timely::progress::Timestamp; -use timely::progress::frontier::Antichain; +use timely::progress::{Antichain, frontier::AntichainRef}; use timely::dataflow::operators::Capability; use timely_sort::Unsigned; @@ -263,7 +263,7 @@ where let mut trace = Some(self.trace.clone()); // release `distinguish_since` capability. - trace.as_mut().unwrap().distinguish_since(&[]); + trace.as_mut().unwrap().distinguish_since(AntichainRef::new(&[])); let mut stash = Vec::new(); let mut capability: Option> = None; @@ -390,13 +390,14 @@ where } // Determine new frontier on queries that may be issued. + // TODO: This code looks very suspect; explain better or fix. let frontier = [ capability.as_ref().map(|c| c.time().clone()), input1.frontier().frontier().get(0).cloned(), ].into_iter().cloned().filter_map(|t| t).min(); if let Some(frontier) = frontier { - trace.as_mut().map(|t| t.advance_by(&[frontier])); + trace.as_mut().map(|t| t.advance_by(AntichainRef::new(&[frontier]))); } else { trace = None; @@ -617,7 +618,7 @@ where } // Extract updates not in advance of `upper`. - let batch = batcher.seal(upper.elements()); + let batch = batcher.seal(upper.clone()); writer.insert(batch.clone(), Some(capability.time().clone())); @@ -632,7 +633,7 @@ where // in messages with new capabilities. let mut new_capabilities = Antichain::new(); - for time in batcher.frontier() { + for time in batcher.frontier().iter() { if let Some(capability) = capabilities.elements().iter().find(|c| c.time().less_equal(time)) { new_capabilities.insert(capability.delayed(time)); } @@ -645,8 +646,8 @@ where } else { // Announce progress updates, even without data. - let _batch = batcher.seal(&input.frontier().frontier()[..]); - writer.seal(&input.frontier().frontier()); + let _batch = batcher.seal(input.frontier().frontier().to_owned()); + writer.seal(input.frontier().frontier().to_owned()); } input_frontier.clear(); diff --git a/src/operators/arrange/mod.rs b/src/operators/arrange/mod.rs index c937e938e..316deca4c 100644 --- a/src/operators/arrange/mod.rs +++ b/src/operators/arrange/mod.rs @@ -44,6 +44,7 @@ use std::cell::RefCell; use std::collections::VecDeque; use timely::scheduling::Activator; +use timely::progress::Antichain; use trace::TraceReader; /// Operating instructions on how to replay a trace. @@ -52,7 +53,7 @@ where Tr: TraceReader, { /// Describes a frontier advance. - Frontier(Vec), + Frontier(Antichain), /// Describes a batch of data and a capability hint. Batch(Tr::Batch, Option), } diff --git a/src/operators/arrange/writer.rs b/src/operators/arrange/writer.rs index d7a7def3d..b6c4c5710 100644 --- a/src/operators/arrange/writer.rs +++ b/src/operators/arrange/writer.rs @@ -8,7 +8,7 @@ use std::cell::RefCell; use lattice::Lattice; use trace::{Trace, Batch, BatchReader}; -use timely::progress::Timestamp; +use timely::progress::{Antichain, Timestamp}; use trace::wrappers::rc::TraceBox; @@ -26,7 +26,7 @@ where Tr::Batch: Batch, { /// Current upper limit. - upper: Vec, + upper: Antichain, /// Shared trace, possibly absent (due to weakness). trace: Weak>>, /// A sequence of private queues into which batches are written. @@ -46,7 +46,9 @@ where queues: Rc>>> ) -> Self { - Self { upper, trace, queues } + let mut temp = Antichain::new(); + temp.extend(upper.into_iter()); + Self { upper: temp, trace, queues } } /// Exerts merge effort, even without additional updates. @@ -64,21 +66,20 @@ where pub fn insert(&mut self, batch: Tr::Batch, hint: Option) { // Something is wrong if not a sequence. - if !(&self.upper[..] == batch.lower()) { + if !(&self.upper == batch.lower()) { println!("{:?} vs {:?}", self.upper, batch.lower()); } - assert!(&self.upper[..] == batch.lower()); + assert!(&self.upper == batch.lower()); assert!(batch.lower() != batch.upper()); - self.upper.clear(); - self.upper.extend(batch.upper().iter().cloned()); + self.upper.clone_from(batch.upper()); // push information to each listener that still exists. let mut borrow = self.queues.borrow_mut(); for queue in borrow.iter_mut() { if let Some(pair) = queue.upgrade() { pair.1.borrow_mut().push_back(TraceReplayInstruction::Batch(batch.clone(), hint.clone())); - pair.1.borrow_mut().push_back(TraceReplayInstruction::Frontier(batch.upper().to_vec())); + pair.1.borrow_mut().push_back(TraceReplayInstruction::Frontier(batch.upper().clone())); pair.0.activate(); } } @@ -92,11 +93,11 @@ where } /// Inserts an empty batch up to `upper`. - pub fn seal(&mut self, upper: &[Tr::Time]) { - if &self.upper[..] != upper { + pub fn seal(&mut self, upper: Antichain) { + if self.upper != upper { use trace::Builder; let builder = >::Builder::new(); - let batch = builder.done(&self.upper[..], upper, &[Tr::Time::minimum()]); + let batch = builder.done(self.upper.clone(), upper, Antichain::from_elem(Tr::Time::minimum())); self.insert(batch, None); } } @@ -109,6 +110,6 @@ where Tr::Batch: Batch, { fn drop(&mut self) { - self.seal(&[]) + self.seal(Antichain::new()) } } diff --git a/src/operators/count.rs b/src/operators/count.rs index 40cd51a94..661a4bb4b 100644 --- a/src/operators/count.rs +++ b/src/operators/count.rs @@ -84,9 +84,8 @@ where for batch in buffer.drain(..) { let mut batch_cursor = batch.cursor(); - let (mut trace_cursor, trace_storage) = trace.cursor_through(batch.lower()).unwrap(); - upper_limit.clear(); - upper_limit.extend(batch.upper().iter().cloned()); + let (mut trace_cursor, trace_storage) = trace.cursor_through(batch.lower().elements()).unwrap(); + upper_limit.clone_from(batch.upper()); while batch_cursor.key_valid(&batch) { diff --git a/src/operators/join.rs b/src/operators/join.rs index 7b67b83bf..0da162aeb 100644 --- a/src/operators/join.rs +++ b/src/operators/join.rs @@ -7,6 +7,7 @@ use std::fmt::Debug; use std::ops::Mul; use std::cmp::Ordering; +use timely::order::PartialOrder; use timely::progress::Timestamp; use timely::dataflow::Scope; use timely::dataflow::operators::generic::{Operator, OutputHandle}; @@ -372,13 +373,12 @@ impl JoinCore for Arranged // and the empty batches themselves (which can be sent as part of trace importing). if acknowledged1.is_none() { acknowledged1 = Some(timely::progress::frontier::Antichain::from_elem(::minimum())); } if let Some(acknowledged1) = &mut acknowledged1 { - if !(batch1.upper().iter().all(|t| acknowledged1.less_equal(t))) { - if !batch1.is_empty() { + if !PartialOrder::less_equal(&*acknowledged1, batch1.upper()) { + if !batch1.is_empty() { panic!("Non-empty batch1 upper not beyond acknowledged frontier: {:?}, {:?}", batch1.upper(), acknowledged1); } } - acknowledged1.clear(); - acknowledged1.extend(batch1.upper().iter().cloned()); + acknowledged1.clone_from(batch1.upper()); } } } @@ -407,13 +407,12 @@ impl JoinCore for Arranged // and the empty batches themselves (which can be sent as part of trace importing). if acknowledged2.is_none() { acknowledged2 = Some(timely::progress::frontier::Antichain::from_elem(::minimum())); } if let Some(acknowledged2) = &mut acknowledged2 { - if !(batch2.upper().iter().all(|t| acknowledged2.less_equal(t))) { + if !PartialOrder::less_equal(&*acknowledged2, batch2.upper()) { if !batch2.is_empty() { panic!("Non-empty batch2 upper not beyond acknowledged frontier: {:?}, {:?}", batch2.upper(), acknowledged2); } } - acknowledged2.clear(); - acknowledged2.extend(batch2.upper().iter().cloned()); + acknowledged2.clone_from(batch2.upper()); } } } @@ -449,7 +448,7 @@ impl JoinCore for Arranged // shut down or advance trace2. if trace2.is_some() && input1.frontier().is_empty() { trace2 = None; } if let Some(ref mut trace2) = trace2 { - trace2.advance_by(&input1.frontier().frontier()[..]); + trace2.advance_by(input1.frontier().frontier()); // At this point, if we haven't seen any input batches we should establish a frontier anyhow. if acknowledged2.is_none() { acknowledged2 = Some(Antichain::from_elem(::minimum())); @@ -463,7 +462,7 @@ impl JoinCore for Arranged // shut down or advance trace1. if trace1.is_some() && input2.frontier().is_empty() { trace1 = None; } if let Some(ref mut trace1) = trace1 { - trace1.advance_by(&input2.frontier().frontier()[..]); + trace1.advance_by(input2.frontier().frontier()); // At this point, if we haven't seen any input batches we should establish a frontier anyhow. if acknowledged1.is_none() { acknowledged1 = Some(Antichain::from_elem(::minimum())); diff --git a/src/operators/reduce.rs b/src/operators/reduce.rs index dd2a6a368..d9c114c07 100644 --- a/src/operators/reduce.rs +++ b/src/operators/reduce.rs @@ -432,8 +432,7 @@ where batches.swap(&mut input_buffer); for batch in input_buffer.drain(..) { - upper_limit.clear(); - upper_limit.extend(batch.upper().iter().cloned()); + upper_limit.clone_from(batch.upper()); batch_cursors.push(batch.cursor()); batch_storage.push(batch); } @@ -582,7 +581,7 @@ where if output_upper.elements() != output_lower.elements() { - let batch = builder.done(output_lower.elements(), output_upper.elements(), &[G::Timestamp::minimum()]); + let batch = builder.done(output_lower.clone(), output_upper.clone(), Antichain::from_elem(G::Timestamp::minimum())); // ship batch to the output, and commit to the output trace. output.session(&capabilities[index]).give(batch.clone()); @@ -619,10 +618,10 @@ where capabilities = new_capabilities; // ensure that observed progres is reflected in the output. - output_writer.seal(upper_limit.elements()); + output_writer.seal(upper_limit.clone()); } else { - output_writer.seal(upper_limit.elements()); + output_writer.seal(upper_limit.clone()); } // We only anticipate future times in advance of `upper_limit`. diff --git a/src/operators/threshold.rs b/src/operators/threshold.rs index 81a2e6ebd..2d4d24595 100644 --- a/src/operators/threshold.rs +++ b/src/operators/threshold.rs @@ -133,10 +133,9 @@ where for batch in buffer.drain(..) { let mut batch_cursor = batch.cursor(); - let (mut trace_cursor, trace_storage) = trace.cursor_through(batch.lower()).unwrap(); + let (mut trace_cursor, trace_storage) = trace.cursor_through(batch.lower().elements()).unwrap(); - upper_limit.clear(); - upper_limit.extend(batch.upper().iter().cloned()); + upper_limit.clone_from(batch.upper()); while batch_cursor.key_valid(&batch) { let key = batch_cursor.key(&batch); diff --git a/src/trace/description.rs b/src/trace/description.rs index 095930ecb..573887751 100644 --- a/src/trace/description.rs +++ b/src/trace/description.rs @@ -55,6 +55,8 @@ //! will often be a logic bug, as `since` does not advance without a corresponding advance in //! times at which data may possibly be sent. +use timely::{PartialOrder, progress::Antichain}; + /// Describes an interval of partially ordered times. /// /// A `Description` indicates a set of partially ordered times, and a moment at which they are @@ -65,31 +67,31 @@ #[derive(Clone, Debug, Abomonation)] pub struct Description