Enable all previously enabled crates (#679)

Enable crates previously disabled, and fix their compile errors.

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>

Author: antiguru

Cargo.toml

@@ -3,13 +3,13 @@ members = [
     "differential-dataflow",
     # "advent_of_code_2017",
     "dogsdogsdogs",
-    #"experiments",
-    #"interactive",
-    #"server",
-    #"server/dataflows/degr_dist",
-    #"server/dataflows/neighborhood",
-    #"server/dataflows/random_graph",
-    #"server/dataflows/reachability",
+    "experiments",
+    "interactive",
+    "server",
+    "server/dataflows/degr_dist",
+    "server/dataflows/neighborhood",
+    "server/dataflows/random_graph",
+    "server/dataflows/reachability",
     #"tpchlike",
     #"doop",
     "mdbook",

experiments/src/bin/arrange.rs

@@ -51,11 +51,11 @@ fn main() {
             let (handle, data) = scope.new_collection();
 
             let probe = match comp {
-                Comp::Nothing => data.probe(),
-                Comp::Exchange => data.inner.exchange(|&(x,_,_): &((usize,()),_,_)| x.0 as u64).probe(),
-                Comp::Arrange => data.arrange_by_self().stream.probe(),
-                Comp::Count => data.arrange_by_self().count_total().probe(),
-                Comp::Distinct => data.arrange_by_self().distinct_total().probe(),
+                Comp::Nothing => data.probe().0,
+                Comp::Exchange => data.inner.exchange(|&(x,_,_): &((usize,()),_,_)| x.0 as u64).probe().0,
+                Comp::Arrange => data.arrange_by_self().stream.probe().0,
+                Comp::Count => data.arrange_by_self().count_total().probe().0,
+                Comp::Distinct => data.arrange_by_self().distinct_total().probe().0,
             };
 
             (handle, probe)

experiments/src/bin/attend.rs

@@ -17,12 +17,14 @@ fn main() {
 
             let (input, graph) = scope.new_collection();
 
+            let graph2 = graph.clone();
             let organizers = graph.explode(|(x,y)| Some((x, (1,0))).into_iter().chain(Some((y, (0,1))).into_iter()))
                                   .threshold_total(|_,w| if w.1 == 0 { 1 } else { 0 });
 
             organizers
+                .clone()
                 .iterate(|scope, attend| {
-                    graph.enter(&scope)
+                    graph2.enter(&scope)
                          .semijoin(attend)
                          .map(|(_,y)| y)
                          .threshold_total(|_,w| if w >= &3 { 1 } else { 0 })

experiments/src/bin/deals-interactive.rs

@@ -45,8 +45,8 @@ fn main() {
             // each edge should exist in both directions.
             let graph = graph.arrange_by_key();
 
-            let probe =
-            interactive(&graph, query1, query2, query3)
+            let (probe, _) =
+            interactive(graph, query1, query2, query3)
                 .filter(move |_| inspect)
                 // .map(|_| ())
                 .consolidate()
@@ -205,34 +205,38 @@ fn main() {
 }
 
 fn interactive<G: Scope>(
-    edges: &Arrange<G, Node, Node, isize>,
+    edges: Arrange<G, Node, Node, isize>,
     tc_1: VecCollection<G, Node>,
     tc_2: VecCollection<G, Node>,
     sg_x: VecCollection<G, Node>
 ) -> VecCollection<G, Node>
 where G::Timestamp: Lattice{
 
     // descendants of tc_1:
+    let tc_1_enter = tc_1.clone();
+    let edges_q1 = edges.clone();
     let query1 =
     tc_1.map(|x| (x,x))
         .iterate(|scope, inner|
-            edges
+            edges_q1
                 .enter(&scope)
                 .join_core(inner.arrange_by_key(), |_,&y,&q| [(y,q)])
-                .concat(tc_1.enter(&scope).map(|x| (x,x)))
+                .concat(tc_1_enter.enter(&scope).map(|x| (x,x)))
                 .distinct()
         )
         .map(|(x,q)| (q,x));
 
     // ancestors of tc_2:
+    let tc_2_enter = tc_2.clone();
+    let edges_q2 = edges.clone();
     let query2 =
     tc_2.map(|x| (x,x))
         .iterate(|scope, inner|
-            edges
+            edges_q2
                 .as_collection(|&k,&v| (v,k))
                 .enter(&scope)
                 .join_core(inner.arrange_by_key(), |_,&y,&q| [(y,q)])
-                .concat(tc_2.enter(&scope).map(|x| (x,x)))
+                .concat(tc_2_enter.enter(&scope).map(|x| (x,x)))
                 .distinct()
         )
         .map(|(x,q)| (q,x));
@@ -242,31 +246,35 @@ where G::Timestamp: Lattice{
     // sg(X,Y) <- magic(X), par(X,Xp), par(Y,Yp), sg(Xp,Yp).
 
     // ancestors of sg_x:
+    let sg_x_enter = sg_x.clone();
+    let sg_x_semijoin = sg_x.clone();
+    let edges_magic = edges.clone();
     let magic =
     sg_x.iterate(|scope, inner|
-            edges
+            edges_magic
                 .as_collection(|&k,&v| (v,k))
                 .enter(&scope)
                 .semijoin(inner)
                 .map(|(_x,y)| y)
-                .concat(sg_x.enter(&scope))
+                .concat(sg_x_enter.enter(&scope))
                 .distinct()
         );
 
     let magic_edges =
-    edges
-        .join_core(magic.arrange_by_self(), |k,v,_| [(k.clone(), v.clone())])
+    edges.clone()
+        .join_core(magic.clone().arrange_by_self(), |k,v,_| [(k.clone(), v.clone())])
         .map(|(x,y)|(y,x))
-        .semijoin(magic)
+        .semijoin(magic.clone())
         .map(|(x,y)|(y,x));
 
+    let magic_enter = magic.clone();
     let query3 =
     magic
         .map(|x| (x,x))   // for query q, sg(x,x)
         .iterate(|scope, inner| {
 
             let edges = edges.enter(&scope);
-            let magic = magic.enter(&scope);
+            let magic = magic_enter.enter(&scope);
             let magic_edges = magic_edges.enter(&scope);
 
             let result =
@@ -279,7 +287,7 @@ where G::Timestamp: Lattice{
             // result.map(|_| ()).consolidate().inspect(|x| println!("\t{:?}", x));
             result
         })
-        .semijoin(sg_x);
+        .semijoin(sg_x_semijoin);
 
     query1.concat(query2).concat(query3).map(|(q,_)| q)
 }

experiments/src/bin/deals.rs

@@ -44,8 +44,8 @@ fn main() {
             let graph = graph.arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>();
 
             match program.as_str() {
-                "tc"    => tc(&graph).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("tc count: {:?}", x)).probe(),
-                "sg"    => sg(&graph).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("sg count: {:?}", x)).probe(),
+                "tc"    => tc(graph.clone()).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("tc count: {:?}", x)).probe(),
+                "sg"    => sg(graph).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("sg count: {:?}", x)).probe(),
                 _       => panic!("must specify one of 'tc', 'sg'.")
             };
 
@@ -83,54 +83,54 @@ fn main() {
 use timely::order::Product;
 
 // returns pairs (n, s) indicating node n can be reached from a root in s steps.
-fn tc<G: Scope<Timestamp=()>>(edges: &EdgeArranged<G, Node, Node, Present>) -> VecCollection<G, Edge, Present> {
+fn tc<G: Scope<Timestamp=()>>(edges: EdgeArranged<G, Node, Node, Present>) -> VecCollection<G, Edge, Present> {
 
     // repeatedly update minimal distances each node can be reached from each root
     edges.stream.scope().iterative::<Iter,_,_>(|scope| {
 
-            let inner = Variable::new(scope, Product::new(Default::default(), 1));
-            let edges = edges.enter(&inner.scope());
+            let (inner, inner_collection) = Variable::new(scope, Product::new(Default::default(), 1));
+            let edges = edges.enter(scope);
 
             let result =
-            inner
+            inner_collection
                 .map(|(x,y)| (y,x))
                 .arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>()
-                .join_core(edges, |_y,&x,&z| Some((x, z)))
+                .join_core(edges.clone(), |_y,&x,&z| Some((x, z)))
                 .concat(edges.as_collection(|&k,&v| (k,v)))
                 .arrange::<KeyBatcher<_,_,_>, KeyBuilder<_,_,_>, KeySpine<_,_,_>>()
-                .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
+                .threshold_semigroup(|_,_,x: Option<&Present>| if x.is_none() { Some(Present) } else { None })
                 ;
 
-            inner.set(result);
+            inner.set(result.clone());
             result.leave()
         }
     )
 }
 
 // returns pairs (n, s) indicating node n can be reached from a root in s steps.
-fn sg<G: Scope<Timestamp=()>>(edges: &EdgeArranged<G, Node, Node, Present>) -> VecCollection<G, Edge, Present> {
+fn sg<G: Scope<Timestamp=()>>(edges: EdgeArranged<G, Node, Node, Present>) -> VecCollection<G, Edge, Present> {
 
-    let peers = edges.join_core(edges, |_,&x,&y| Some((x,y))).filter(|&(x,y)| x != y);
+    let peers = edges.clone().join_core(edges.clone(), |_,&x,&y| Some((x,y))).filter(|&(x,y)| x != y);
 
     // repeatedly update minimal distances each node can be reached from each root
     peers.scope().iterative::<Iter,_,_>(|scope| {
 
-            let inner = Variable::new(scope, Product::new(Default::default(), 1));
-            let edges = edges.enter(&inner.scope());
-            let peers = peers.enter(&inner.scope());
+            let (inner, inner_collection) = Variable::new(scope, Product::new(Default::default(), 1));
+            let edges = edges.enter(scope);
+            let peers = peers.enter(scope);
 
             let result =
-            inner
+            inner_collection
                 .arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>()
-                .join_core(edges, |_,&x,&z| Some((x, z)))
+                .join_core(edges.clone(), |_,&x,&z| Some((x, z)))
                 .arrange::<ValBatcher<_,_,_,_>, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>()
                 .join_core(edges, |_,&x,&z| Some((x, z)))
                 .concat(peers)
                 .arrange::<KeyBatcher<_,_,_>, KeyBuilder<_,_,_>, KeySpine<_,_,_>>()
-                .threshold_semigroup(|_,_,x| if x.is_none() { Some(Present) } else { None })
+                .threshold_semigroup(|_,_,x: Option<&Present>| if x.is_none() { Some(Present) } else { None })
                 ;
 
-            inner.set(result);
+            inner.set(result.clone());
             result.leave()
         }
     )