Retire SemigroupVariable

Author: frankmcsherry

differential-dataflow/examples/monoid-bfs.rs

@@ -129,11 +129,11 @@ where
     // repeatedly update minimal distances each node can be reached from each root
     roots.scope().iterative::<u32,_,_>(|scope| {
 
-        use differential_dataflow::operators::iterate::SemigroupVariable;
+        use differential_dataflow::operators::iterate::Variable;
         use differential_dataflow::trace::implementations::{KeySpine, KeyBuilder};
 
         use timely::order::Product;
-        let (variable, collection) = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+        let (variable, collection) = Variable::new(scope, Product::new(Default::default(), 1));
 
         let edges = edges.enter(scope);
         let roots = roots.enter(scope);

differential-dataflow/src/algorithms/graphs/propagate.rs

@@ -79,15 +79,15 @@ where
 
     nodes.scope().iterative::<usize,_,_>(|scope| {
 
-        use crate::operators::iterate::SemigroupVariable;
+        use crate::operators::iterate::Variable;
         use crate::trace::implementations::{ValBuilder, ValSpine};
 
         use timely::order::Product;
 
         let edges = edges.enter(scope);
         let nodes = nodes.enter_at(scope, move |r| 256 * (64 - (logic(&r.1)).leading_zeros() as usize));
 
-        let (proposals_bind, proposals) = SemigroupVariable::new(scope, Product::new(Default::default(), 1usize));
+        let (proposals_bind, proposals) = Variable::new(scope, Product::new(Default::default(), 1usize));
 
         let labels =
         proposals

differential-dataflow/src/operators/iterate.rs

@@ -110,7 +110,7 @@ impl<G: Scope<Timestamp: Lattice>, D: Ord+Data+Debug, R: Semigroup+'static> Iter
                 // wrapped by `variable`, but it also results in substantially more
                 // diffs produced; `result` is post-consolidation, and means fewer
                 // records are yielded out of the loop.
-                let (variable, collection) = SemigroupVariable::new(subgraph, Product::new(Default::default(), 1));
+                let (variable, collection) = Variable::new(subgraph, Product::new(Default::default(), 1));
                 let result = logic(subgraph.clone(), collection);
                 variable.set(result.clone());
                 result.leave()
@@ -166,7 +166,7 @@ pub type VecVariable<G, D, R> = Variable<G, Vec<(D, <G as ScopeParent>::Timestam
 impl<G, C: Container> Variable<G, C>
 where
     G: Scope<Timestamp: Lattice>,
-    C: crate::collection::containers::Negate + crate::collection::containers::ResultsIn<<G::Timestamp as Timestamp>::Summary>,
+    C: crate::collection::containers::ResultsIn<<G::Timestamp as Timestamp>::Summary>,
 {
     /// Creates a new initially empty `Variable`.
     ///
@@ -179,10 +179,10 @@ where
     }
 
     /// Creates a new `Variable` from a supplied `source` stream.
-    pub fn new_from(source: Collection<G, C>, step: <G::Timestamp as Timestamp>::Summary) -> (Self, Collection<G, C>) where C: Clone {
+    pub fn new_from(source: Collection<G, C>, step: <G::Timestamp as Timestamp>::Summary) -> (Self, Collection<G, C>) where C: Clone + crate::collection::containers::Negate {
         let (feedback, updates) = source.inner.scope().feedback(step.clone());
         let collection = Collection::<G, C>::new(updates).concat(source.clone());
-        (Variable { feedback, source: Some(source), step }, collection)
+        (Variable { feedback, source: Some(source.negate()), step }, collection)
     }
 
     /// Set the definition of the `Variable` to a collection.
@@ -191,61 +191,10 @@ where
     /// which may be recursively defined in terms of the variable itself.
     pub fn set(mut self, mut result: Collection<G, C>) {
         if let Some(source) = self.source.take() {
-            result = result.concat(source.negate());
+            result = result.concat(source);
         }
-        self.set_concat(result)
-    }
-
-    /// Set the definition of the `Variable` to a collection concatenated to `self`.
-    ///
-    /// This method is a specialization of `set` which has the effect of concatenating
-    /// `result` and `self` before calling `set`. This method avoids some dataflow
-    /// complexity related to retracting the initial input, and will do less work in
-    /// that case.
-    ///
-    /// This behavior can also be achieved by using `new` to create an empty initial
-    /// collection, and then using `self.set(self.concat(result))`.
-    pub fn set_concat(self, result: Collection<G, C>) {
-        let step = self.step;
-        result
-            .results_in(step)
-            .inner
-            .connect_loop(self.feedback);
-    }
-}
-
-/// A recursively defined collection that only "grows".
-///
-/// `SemigroupVariable` is a weakening of `Variable` to allow difference types
-/// that do not implement `Abelian` and only implement `Semigroup`. This means
-/// that it can be used in settings where the difference type does not support
-/// negation.
-pub struct SemigroupVariable<G, C>
-where
-    G: Scope<Timestamp: Lattice>,
-    C: Container,
-{
-    feedback: Handle<G, C>,
-    step: <G::Timestamp as Timestamp>::Summary,
-}
-
-impl<G, C: Container> SemigroupVariable<G, C>
-where
-    G: Scope<Timestamp: Lattice>,
-    C: crate::collection::containers::ResultsIn<<G::Timestamp as Timestamp>::Summary>,
-{
-    /// Creates a new initially empty `SemigroupVariable`.
-    pub fn new(scope: &mut G, step: <G::Timestamp as Timestamp>::Summary) -> (Self, Collection<G, C>) {
-        let (feedback, updates) = scope.feedback(step.clone());
-        let collection = Collection::<G,C>::new(updates);
-        (SemigroupVariable { feedback, step }, collection)
-    }
-
-    /// Adds a new source of data to `self`.
-    pub fn set(self, result: Collection<G, C>) {
-        let step = self.step;
         result
-            .results_in(step)
+            .results_in(self.step)
             .inner
             .connect_loop(self.feedback);
     }

experiments/src/bin/deals.rs

@@ -10,7 +10,7 @@ use differential_dataflow::trace::implementations::{ValSpine, KeySpine, KeyBatch
 use differential_dataflow::operators::arrange::TraceAgent;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::operators::arrange::Arrange;
-use differential_dataflow::operators::iterate::SemigroupVariable;
+use differential_dataflow::operators::iterate::Variable;
 use differential_dataflow::difference::Present;
 
 type EdgeArranged<G, K, V, R> = Arranged<G, TraceAgent<ValSpine<K, V, <G as ScopeParent>::Timestamp, R>>>;
@@ -88,7 +88,7 @@ fn tc<G: Scope<Timestamp=()>>(edges: &EdgeArranged<G, Node, Node, Present>) -> V
     // repeatedly update minimal distances each node can be reached from each root
     edges.stream.scope().iterative::<Iter,_,_>(|scope| {
 
-            let inner = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+            let inner = Variable::new(scope, Product::new(Default::default(), 1));
             let edges = edges.enter(&inner.scope());
 
             let result =
@@ -115,7 +115,7 @@ fn sg<G: Scope<Timestamp=()>>(edges: &EdgeArranged<G, Node, Node, Present>) -> V
     // repeatedly update minimal distances each node can be reached from each root
     peers.scope().iterative::<Iter,_,_>(|scope| {
 
-            let inner = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+            let inner = Variable::new(scope, Product::new(Default::default(), 1));
             let edges = edges.enter(&inner.scope());
             let peers = peers.enter(&inner.scope());
 

experiments/src/bin/graphs-static.rs

@@ -6,7 +6,7 @@ use timely::dataflow::operators::ToStream;
 use differential_dataflow::input::Input;
 use differential_dataflow::VecCollection;
 use differential_dataflow::operators::*;
-use differential_dataflow::operators::iterate::SemigroupVariable;
+use differential_dataflow::operators::iterate::Variable;
 use differential_dataflow::AsCollection;
 
 use graph_map::GraphMMap;
@@ -118,7 +118,7 @@ fn reach<G: Scope<Timestamp = ()>> (
         let graph = graph.enter(scope);
         let roots = roots.enter(scope);
 
-        let inner = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+        let inner = Variable::new(scope, Product::new(Default::default(), 1));
 
         let result =
         graph.join_core(inner.arrange_by_self(), |_src,&dst,&()| Some(dst))
@@ -144,7 +144,7 @@ fn bfs<G: Scope<Timestamp = ()>> (
         let graph = graph.enter(scope);
         let roots = roots.enter(scope);
 
-        let inner = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+        let inner = Variable::new(scope, Product::new(Default::default(), 1));
         let result =
         graph.join_core(inner.arrange_by_key(), |_src,&dest,&dist| [(dest, dist+1)])
              .concat(roots)
@@ -176,7 +176,7 @@ fn connected_components<G: Scope<Timestamp = ()>>(
         let reverse = reverse.enter(scope);
         let nodes = nodes.enter(scope);
 
-        let inner = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+        let inner = Variable::new(scope, Product::new(Default::default(), 1));
 
         let labels = inner.arrange_by_key();
         let f_prop = labels.join_core(forward, |_k,l,d| Some((*d,*l)));

experiments/src/bin/graspan1.rs

@@ -9,7 +9,7 @@ use differential_dataflow::input::Input;
 use differential_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine};
 use differential_dataflow::operators::*;
 use differential_dataflow::operators::arrange::Arrange;
-use differential_dataflow::operators::iterate::SemigroupVariable;
+use differential_dataflow::operators::iterate::Variable;
 
 type Node = u32;
 type Time = ();
@@ -41,7 +41,7 @@ fn main() {
                 let nodes = nodes.enter(inner).map(|(a,b)| (b,a));
                 let edges = edges.enter(inner);
 
-                let labels = SemigroupVariable::new(inner, Product::new(Default::default(), 1));
+                let labels = Variable::new(inner, Product::new(Default::default(), 1));
 
                 let next =
                 labels.join_core(edges, |_b, a, c| Some((*c, *a)))

experiments/src/bin/graspan2.rs

@@ -4,7 +4,7 @@ use std::fs::File;
 use timely::dataflow::Scope;
 use timely::order::Product;
 
-use differential_dataflow::operators::iterate::SemigroupVariable;
+use differential_dataflow::operators::iterate::Variable;
 
 use differential_dataflow::VecCollection;
 use differential_dataflow::input::Input;
@@ -57,8 +57,8 @@ fn unoptimized() {
                     let assignment = assignment.enter(scope);
                     let dereference = dereference.enter(scope);
 
-                    let value_flow = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
-                    let memory_alias = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+                    let value_flow = Variable::new(scope, Product::new(Default::default(), 1));
+                    let memory_alias = Variable::new(scope, Product::new(Default::default(), 1));
 
                     let value_flow_arranged = value_flow.arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>();
                     let memory_alias_arranged = memory_alias.arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>();
@@ -182,8 +182,8 @@ fn optimized() {
                     let assignment = assignment.enter(scope);
                     let dereference = dereference.enter(scope);
 
-                    let value_flow = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
-                    let memory_alias = SemigroupVariable::new(scope, Product::new(Default::default(), 1));
+                    let value_flow = Variable::new(scope, Product::new(Default::default(), 1));
+                    let memory_alias = Variable::new(scope, Product::new(Default::default(), 1));
 
                     let value_flow_arranged = value_flow.arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>();
                     let memory_alias_arranged = memory_alias.arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>();