Separate chunker from batcher

Cargo.toml

@@ -23,9 +23,9 @@ rust-version = "1.86"
 
 [workspace.dependencies]
 differential-dataflow = { path = "differential-dataflow", default-features = false, version = "0.22.0" }
-timely = { version = "0.28", default-features = false }
+timely = { version = "0.29", default-features = false }
 columnar = { version = "0.12", default-features = false }
-#timely = { git = "https://github.com/TimelyDataflow/timely-dataflow" }
+#timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false }
 #timely = { path = "../timely-dataflow/timely/", default-features = false }
 
 [workspace.lints.clippy]

README.md

@@ -22,7 +22,7 @@ Alternately, here is a fragment that computes the set of nodes reachable from a
 ```rust
 let reachable =
 roots.iterate(|scope, reach|
-    edges.enter(&scope)
+    edges.enter(scope)
          .semijoin(reach)
          .map(|(src, dst)| dst)
          .concat(reach)
@@ -337,7 +337,7 @@ edges.iterate(|scope, inner| {
                       .map(|(node,_)| node);
 
     // keep edges between active vertices
-    edges.enter(&scope)
+    edges.enter(scope)
          .semijoin(active)
          .map(|(src,dst)| (dst,src))
          .semijoin(active)

advent_of_code_2017/Cargo.toml

@@ -11,4 +11,4 @@ workspace = true
 
 [dependencies]
 differential-dataflow = { workspace = true }
-timely = { git = "https://github.com/frankmcsherry/timely-dataflow" }
+timely = { workspace = true }

advent_of_code_2017/src/bin/day_06.rs

@@ -24,7 +24,7 @@ fn main() {
 
             let stable = banks.iterate(|scope, iter|
                 iter.map_in_place(|banks| recycle(banks))
-                    .concat(banks.enter(&scope))
+                    .concat(banks.enter(scope))
                     .distinct()
             );
 
@@ -43,7 +43,7 @@ fn main() {
             loop_point
                 .iterate(|scope, iter|
                     iter.map_in_place(|banks| recycle(banks))
-                        .concat(loop_point.enter(&scope))
+                        .concat(loop_point.enter(scope))
                         .distinct()
                 )
                 .map(|_| ((),()))

advent_of_code_2017/src/bin/day_07.rs

@@ -1104,10 +1104,10 @@ tvhftq (35)";
 
             let total_weights: VecCollection<_,String> = weights
                 .iterate(|scope, inner| {
-                    parents.enter(&scope)
+                    parents.enter(scope)
                            .semijoin(inner)
                            .map(|(_, parent)| parent)
-                           .concat(weights.enter(&scope))
+                           .concat(weights.enter(scope))
                 });
 
             parents

advent_of_code_2017/src/bin/day_08.rs

@@ -1107,7 +1107,7 @@ wui inc -120 if i > -2038";
                 .map(|_| ((0, String::new()), 0))
                 .iterate(|scope, valid| {
 
-                    let edits = edits.enter(&scope);
+                    let edits = edits.enter(scope);
 
                     valid
                         .prefix_sum_at(edits.map(|(key,_)| key), 0, |_k,x,y| *x + *y)

advent_of_code_2017/src/bin/day_09.rs

@@ -120,7 +120,7 @@ where
                     if input.len() > 1 { result = combine(result, &(input[1].0).1); }
                     output.push((result, 1));
                 })
-                .concat(unit_ranges.enter(&scope))
+                .concat(unit_ranges.enter(scope))
         )
 }
 
@@ -154,10 +154,10 @@ where
         .iterate(|scope, state| {
             aggregates
                 .filter(|&((_, log),_)| log < 64)    // the log = 64 interval doesn't help us here (overflows).
-                .enter(&scope)
+                .enter(scope)
                 .map(|((pos, log), data)| (pos, (log, data)))
                 .join_map(state, move |&pos, &(log, ref data), state| (pos + (1 << log), combine(state, data)))
-                .concat(init_state.enter(&scope))
+                .concat(init_state.enter(scope))
                 .distinct()
         })
         .consolidate()

advent_of_code_2017/src/bin/day_12.rs

@@ -2035,8 +2035,8 @@ fn main() {
             let labels =
             nodes
                 .iterate(|scope, label| {
-                    let edges = edges.enter(&scope);
-                    let nodes = nodes.enter(&scope);
+                    let edges = edges.enter(scope);
+                    let nodes = nodes.enter(scope);
                     label
                         .join_map(edges, |_src, &lbl, &tgt| (tgt, lbl))
                         .concat(nodes)

diagnostics/Cargo.toml

@@ -1,6 +1,7 @@
 [package]
 name = "diagnostics"
 version = "0.1.0"
+publish = false
 edition.workspace = true
 rust-version.workspace = true
 

diagnostics/src/logging.rs

@@ -34,7 +34,6 @@ use differential_dataflow::operators::arrange::TraceAgent;
 use differential_dataflow::trace::implementations::{KeySpine, ValSpine};
 use differential_dataflow::{AsCollection, VecCollection};
 
-use timely::communication::Allocate;
 use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::capture::{Event, EventLink, Replay, Capture};
@@ -244,12 +243,11 @@ fn quantize(time: Duration, interval: Duration) -> Duration {
 }
 
 /// Quantize timestamps in a collection's inner stream.
-fn quantize_collection<S, D>(
-    collection: VecCollection<S, D, i64>,
+fn quantize_collection<'scope, D>(
+    collection: VecCollection<'scope, Duration, D, i64>,
     interval: Duration,
-) -> VecCollection<S, D, i64>
+) -> VecCollection<'scope, Duration, D, i64>
 where
-    S: Scope<Timestamp = Duration>,
     D: differential_dataflow::Data,
 {
     collection
@@ -276,7 +274,7 @@ where
 ///
 /// Returns a [`LoggingState`] with trace handles and a [`SinkHandle`] for
 /// the WebSocket thread.
-pub fn register<A: Allocate>(worker: &mut Worker<A>, log_logging: bool) -> LoggingState {
+pub fn register(worker: &mut Worker, log_logging: bool) -> LoggingState {
     let start = Instant::now();
 
     // Event links for logging capture (worker-internal, Rc-based).
@@ -377,8 +375,8 @@ pub fn register<A: Allocate>(worker: &mut Worker<A>, log_logging: bool) -> Loggi
     }
 }
 
-fn install_loggers<A: Allocate>(
-    worker: &mut Worker<A>,
+fn install_loggers(
+    worker: &mut Worker,
     t_link: Rc<EventLink<Duration, Vec<(Duration, TimelyEvent)>>>,
     d_link: Rc<EventLink<Duration, Vec<(Duration, DifferentialEvent)>>>,
 ) {
@@ -402,11 +400,11 @@ fn install_loggers<A: Allocate>(
 // ============================================================================
 
 /// Internal: collections before arrangement, used for the cross-join.
-struct TimelyCollections<S: Scope> {
-    operators: VecCollection<S, (usize, String, Vec<usize>), i64>,
-    channels: VecCollection<S, (usize, Vec<usize>, (usize, usize), (usize, usize)), i64>,
-    elapsed: VecCollection<S, usize, i64>,
-    messages: VecCollection<S, usize, i64>,
+struct TimelyCollections<'scope> {
+    operators: VecCollection<'scope, Duration, (usize, String, Vec<usize>), i64>,
+    channels: VecCollection<'scope, Duration, (usize, Vec<usize>, (usize, usize), (usize, usize)), i64>,
+    elapsed: VecCollection<'scope, Duration, usize, i64>,
+    messages: VecCollection<'scope, Duration, usize, i64>,
 }
 
 #[derive(Default)]
@@ -416,16 +414,16 @@ struct TimelyDemuxState {
 }
 
 /// Build timely logging collections and arrangements.
-fn construct_timely<S: Scope<Timestamp = Duration>>(
-    scope: &mut S,
-    stream: Stream<S, Vec<(Duration, TimelyEvent)>>,
-) -> (TimelyTraces, TimelyCollections<S>) {
+fn construct_timely<'scope>(
+    scope: Scope<'scope, Duration>,
+    stream: Stream<'scope, Duration, Vec<(Duration, TimelyEvent)>>,
+) -> (TimelyTraces, TimelyCollections<'scope>) {
     type OpUpdate = ((usize, String, Vec<usize>), Duration, i64);
     type ChUpdate = ((usize, Vec<usize>, (usize, usize), (usize, usize)), Duration, i64);
     type ElUpdate = (usize, Duration, i64);
     type MsgUpdate = (usize, Duration, i64);
 
-    let mut demux = OperatorBuilder::new("Timely Demux".to_string(), scope.clone());
+    let mut demux = OperatorBuilder::new("Timely Demux".to_string(), scope);
     let mut input = demux.new_input(stream, Pipeline);
 
     let (op_out, operates) = demux.new_output::<Vec<OpUpdate>>();
@@ -536,24 +534,24 @@ fn construct_timely<S: Scope<Timestamp = Duration>>(
 // ============================================================================
 
 /// Internal: collections before arrangement, used for the cross-join.
-struct DifferentialCollections<S: Scope> {
-    arrangement_batches: VecCollection<S, usize, i64>,
-    arrangement_records: VecCollection<S, usize, i64>,
-    sharing: VecCollection<S, usize, i64>,
-    batcher_records: VecCollection<S, usize, i64>,
-    batcher_size: VecCollection<S, usize, i64>,
-    batcher_capacity: VecCollection<S, usize, i64>,
-    batcher_allocations: VecCollection<S, usize, i64>,
+struct DifferentialCollections<'scope> {
+    arrangement_batches: VecCollection<'scope, Duration, usize, i64>,
+    arrangement_records: VecCollection<'scope, Duration, usize, i64>,
+    sharing: VecCollection<'scope, Duration, usize, i64>,
+    batcher_records: VecCollection<'scope, Duration, usize, i64>,
+    batcher_size: VecCollection<'scope, Duration, usize, i64>,
+    batcher_capacity: VecCollection<'scope, Duration, usize, i64>,
+    batcher_allocations: VecCollection<'scope, Duration, usize, i64>,
 }
 
 /// Build differential logging collections and arrangements.
-fn construct_differential<S: Scope<Timestamp = Duration>>(
-    scope: &mut S,
-    stream: Stream<S, Vec<(Duration, DifferentialEvent)>>,
-) -> (DifferentialTraces, DifferentialCollections<S>) {
+fn construct_differential<'scope>(
+    scope: Scope<'scope, Duration>,
+    stream: Stream<'scope, Duration, Vec<(Duration, DifferentialEvent)>>,
+) -> (DifferentialTraces, DifferentialCollections<'scope>) {
     type Update = (usize, Duration, i64);
 
-    let mut demux = OperatorBuilder::new("Differential Demux".to_string(), scope.clone());
+    let mut demux = OperatorBuilder::new("Differential Demux".to_string(), scope);
     let mut input = demux.new_input(stream, Pipeline);
 
     let (bat_out, batches) = demux.new_output::<Vec<Update>>();

differential-dataflow/examples/accumulate.rs

@@ -7,7 +7,7 @@ fn main() {
     let keys: usize = std::env::args().nth(1).unwrap().parse().unwrap();
     let batch: usize = 10_000;
 
-    // This computation demonstrates in-place accumulation of arbitrarily large 
+    // This computation demonstrates in-place accumulation of arbitrarily large
     // volumes of input data, consuming space bounded by the number of distinct keys.
     timely::execute_from_args(std::env::args().skip(2), move |worker| {
 
@@ -17,11 +17,10 @@ fn main() {
         let mut input = worker.dataflow::<(), _, _>(|scope| {
             let (input, data) = scope.new_collection::<_, isize>();
 
-            use timely::dataflow::Scope;
             scope.iterative::<u32,_,_>(|inner| {
                 data.enter_at(inner, |_| 0)
                     .consolidate()
-                    .leave()
+                    .leave(scope)
             });
 
             input
@@ -41,7 +40,7 @@ fn main() {
             }
             counter += batch;
 
-            worker.step(); 
+            worker.step();
             let elapsed = timer.elapsed();
 
             if elapsed.as_secs() as usize > last_sec {
@@ -54,4 +53,4 @@ fn main() {
         }
 
     }).unwrap();
-}
+}

differential-dataflow/examples/arrange.rs

@@ -2,7 +2,6 @@ use rand::{Rng, SeedableRng, StdRng};
 
 use timely::dataflow::operators::*;
 use timely::order::Product;
-use timely::scheduling::Scheduler;
 
 use differential_dataflow::input::Input;
 use differential_dataflow::AsCollection;
@@ -109,8 +108,8 @@ fn main() {
             // repeatedly update minimal distances each node can be reached from each root
             roots.clone().iterate(|scope, dists| {
 
-                let edges = edges.enter(&scope);
-                let roots = roots.enter(&scope);
+                let edges = edges.enter(scope);
+                let roots = roots.enter(scope);
 
                 dists.arrange_by_key()
                      .join_core(edges, |_k,l,d| Some((*d, l+1)))
@@ -175,4 +174,4 @@ fn main() {
             }
         }
     }).unwrap();
-}
+}

differential-dataflow/examples/bfs.rs

@@ -1,6 +1,5 @@
 use rand::{Rng, SeedableRng, StdRng};
 
-use timely::dataflow::*;
 use timely::dataflow::operators::probe::Handle;
 
 use differential_dataflow::input::Input;
@@ -91,18 +90,18 @@ fn main() {
 }
 
 // returns pairs (n, s) indicating node n can be reached from a root in s steps.
-fn bfs<G>(edges: VecCollection<G, Edge>, roots: VecCollection<G, Node>) -> VecCollection<G, (Node, u32)>
+fn bfs<'scope, T>(edges: VecCollection<'scope, T, Edge>, roots: VecCollection<'scope, T, Node>) -> VecCollection<'scope, T, (Node, u32)>
 where
-    G: Scope<Timestamp: Lattice+Ord>,
+    T: timely::progress::Timestamp + Lattice + Ord,
 {
     // initialize roots as reaching themselves at distance 0
     let nodes = roots.map(|x| (x, 0));
 
     // repeatedly update minimal distances each node can be reached from each root
     nodes.clone().iterate(|scope, inner| {
 
-        let nodes = nodes.enter(&scope);
-        let edges = edges.enter(&scope);
+        let nodes = nodes.enter(scope);
+        let edges = edges.enter(scope);
 
         inner.join_map(edges, |_k,l,d| (*d, l+1))
              .concat(nodes)