rebase on master (TimelyDataflow#312)

Author: frankmcsherry

src/operators/arrange/agent.rs

@@ -74,7 +74,7 @@ where
     fn cursor_through(&mut self, frontier: AntichainRef<Tr::Time>) -> Option<(Tr::Cursor, <Tr::Cursor as Cursor<Tr::Key, Tr::Val, Tr::Time, Tr::R>>::Storage)> {
         self.trace.borrow_mut().trace.cursor_through(frontier)
     }
-    fn map_batches<F: FnMut(&Self::Batch)>(&mut self, f: F) { self.trace.borrow_mut().trace.map_batches(f) }
+    fn map_batches<F: FnMut(&Self::Batch)>(&self, f: F) { self.trace.borrow().trace.map_batches(f) }
 }
 
 impl<Tr> TraceAgent<Tr>

src/trace/implementations/graph.rs

@@ -60,7 +60,7 @@ where
     }
     fn get_physical_compaction(&mut self) -> &[Product<RootTimestamp,()>] { &self.spine.get_physical_compaction() }
 
-    fn map_batches<F: FnMut(&Self::Batch)>(&mut self, f: F) {
+    fn map_batches<F: FnMut(&Self::Batch)>(&self, f: F) {
         self.spine.map_batches(f)
     }
 }

src/trace/implementations/spine_fueled.rs

@@ -215,7 +215,7 @@ where
     }
     fn get_physical_compaction(&mut self) -> &[T] { &self.physical_frontier[..] }
 
-    fn map_batches<F: FnMut(&Self::Batch)>(&mut self, mut f: F) {
+    fn map_batches<F: FnMut(&Self::Batch)>(&self, mut f: F) {
         for batch in self.merging.iter().rev() {
             match *batch {
                 Some(MergeState::Merging(ref batch1, ref batch2, _, _)) => { f(batch1); f(batch2); },

src/trace/implementations/spine_fueled_neu.rs

@@ -224,7 +224,7 @@ where
     }
     fn get_physical_compaction(&mut self) -> AntichainRef<T> { self.physical_frontier.borrow() }
 
-    fn map_batches<F: FnMut(&Self::Batch)>(&mut self, mut f: F) {
+    fn map_batches<F: FnMut(&Self::Batch)>(&self, mut f: F) {
         for batch in self.merging.iter().rev() {
             match batch {
                 MergeState::Double(MergeVariant::InProgress(batch1, batch2, _)) => { f(batch1); f(batch2); },

src/trace/mod.rs

@@ -154,7 +154,7 @@ pub trait TraceReader {
     /// This is currently used only to extract historical data to prime late-starting operators who want to reproduce
     /// the stream of batches moving past the trace. It could also be a fine basis for a default implementation of the
     /// cursor methods, as they (by default) just move through batches accumulating cursors into a cursor list.
-    fn map_batches<F: FnMut(&Self::Batch)>(&mut self, f: F);
+    fn map_batches<F: FnMut(&Self::Batch)>(&self, f: F);
 
     /// Reads the upper frontier of committed times.
     ///

src/trace/wrappers/enter.rs

@@ -50,7 +50,7 @@ where
     type Batch = BatchEnter<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Batch, TInner>;
     type Cursor = CursorEnter<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Cursor, TInner>;
 
-    fn map_batches<F: FnMut(&Self::Batch)>(&mut self, mut f: F) {
+    fn map_batches<F: FnMut(&Self::Batch)>(&self, mut f: F) {
         self.trace.map_batches(|batch| {
             f(&Self::Batch::make_from(batch.clone()));
         })

src/trace/wrappers/enter_at.rs

@@ -65,7 +65,7 @@ where
     type Batch = BatchEnter<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Batch, TInner,F>;
     type Cursor = CursorEnter<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Cursor, TInner,F>;
 
-    fn map_batches<F2: FnMut(&Self::Batch)>(&mut self, mut f: F2) {
+    fn map_batches<F2: FnMut(&Self::Batch)>(&self, mut f: F2) {
         let logic = self.logic.clone();
         self.trace.map_batches(|batch| {
             f(&Self::Batch::make_from(batch.clone(), logic.clone()));

src/trace/wrappers/filter.rs

@@ -43,7 +43,7 @@ where
     type Batch = BatchFilter<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Batch, F>;
     type Cursor = CursorFilter<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Cursor, F>;
 
-    fn map_batches<F2: FnMut(&Self::Batch)>(&mut self, mut f: F2) {
+    fn map_batches<F2: FnMut(&Self::Batch)>(&self, mut f: F2) {
         let logic = self.logic.clone();
         self.trace
             .map_batches(|batch| f(&Self::Batch::make_from(batch.clone(), logic.clone())))

src/trace/wrappers/freeze.rs

@@ -96,7 +96,7 @@ where
     type Batch = BatchFreeze<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Batch, F>;
     type Cursor = CursorFreeze<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Cursor, F>;
 
-    fn map_batches<F2: FnMut(&Self::Batch)>(&mut self, mut f: F2) {
+    fn map_batches<F2: FnMut(&Self::Batch)>(&self, mut f: F2) {
         let func = &self.func;
         self.trace.map_batches(|batch| {
             f(&Self::Batch::make_from(batch.clone(), func.clone()));

src/trace/wrappers/frontier.rs

@@ -50,7 +50,7 @@ where
     type Batch = BatchFrontier<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Batch>;
     type Cursor = CursorFrontier<Tr::Key, Tr::Val, Tr::Time, Tr::R, Tr::Cursor>;
 
-    fn map_batches<F: FnMut(&Self::Batch)>(&mut self, mut f: F) {
+    fn map_batches<F: FnMut(&Self::Batch)>(&self, mut f: F) {
         let frontier = self.frontier.borrow();
         self.trace.map_batches(|batch| f(&Self::Batch::make_from(batch.clone(), frontier)))
     }