Arc-backed spines for cross-thread arrangement sharing

differential-dataflow/src/operators/arrange/agent.rs

@@ -283,7 +283,7 @@
                 let activator = scope.activator_for(Rc::clone(&info.address));
                 let queue = self.new_listener(activator);
 
                 let activator = scope.activator_for(info.address);
                 *shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));
 
                 capabilities.borrow_mut().as_mut().unwrap().insert(capability);
@@ -416,7 +416,7 @@
                 let activator = scope.activator_for(Rc::clone(&info.address));
                 let queue = self.new_listener(activator);
 
                 let activator = scope.activator_for(info.address);
                 *shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));
 
                 capabilities.borrow_mut().as_mut().unwrap().insert(capability);
@@ -568,6 +568,15 @@
         ///
         /// This is used to inspect batches for purposes of resource accounting in external systems.
         pub fn trace(&self) -> &Tr { &self.trace }
+        /// The accumulated logical-compaction holds of all referees of the trace.
+        ///
+        /// This is the `MutableAntichain` whose frontier the trace compacts to. Exposed so an
+        /// external referee (for example a cross-runtime publisher) can compute the frontier the
+        /// trace would compact to without its own hold, by cloning this and subtracting that hold.
+        pub fn logical_compaction(&self) -> &MutableAntichain<Tr::Time> { &self.logical_compaction }
+        /// The accumulated physical-compaction holds of all referees of the trace. See
+        /// [`TraceBox::logical_compaction`].
+        pub fn physical_compaction(&self) -> &MutableAntichain<Tr::Time> { &self.physical_compaction }
         /// Replaces elements of `lower` with those of `upper`.
         #[inline]
         pub fn adjust_logical_compaction(&mut self, lower: AntichainRef<Tr::Time>, upper: AntichainRef<Tr::Time>) {

differential-dataflow/src/operators/arrange/mod.rs

@@ -63,6 +63,7 @@ type TraceAgentQueueWriter<Tr> = Weak<(Activator, RefCell<BatchQueue<Tr>>)>;
 pub mod writer;
 pub mod agent;
 pub mod arrangement;
+pub mod sharing;
 
 pub mod upsert;
 

differential-dataflow/src/trace/implementations/mod.rs

@@ -53,6 +53,13 @@ pub use self::ord_neu::OrdKeySpine as KeySpine;
 pub use self::ord_neu::OrdKeyBatcher as KeyBatcher;
 pub use self::ord_neu::RcOrdKeyBuilder as KeyBuilder;
 
+// `Arc`-backed variants of the `Val`/`Key` aliases, whose batches can be shared across threads.
+// The batcher is shared with the `Rc` variants, since batchers do not reference-count batches.
+pub use self::ord_neu::ArcOrdValSpine as ArcValSpine;
+pub use self::ord_neu::ArcOrdValBuilder as ArcValBuilder;
+pub use self::ord_neu::ArcOrdKeySpine as ArcKeySpine;
+pub use self::ord_neu::ArcOrdKeyBuilder as ArcKeyBuilder;
+
 use std::convert::TryInto;
 
 use serde::{Deserialize, Serialize};

differential-dataflow/src/trace/implementations/ord_neu.rs

@@ -9,11 +9,13 @@
 //! and should consume fewer resources (computation and memory) when it applies.
 
 use std::rc::Rc;
+use std::sync::Arc;
 
 use crate::trace::implementations::spine_fueled::Spine;
 use crate::trace::implementations::merge_batcher::MergeBatcher;
 use crate::trace::implementations::merge_batcher::vec::VecMerger;
 use crate::trace::rc_blanket_impls::RcBuilder;
+use crate::trace::arc_blanket_impls::ArcBuilder;
 
 use super::{Layout, Vector};
 
@@ -22,17 +24,29 @@ pub use self::key_batch::{OrdKeyBatch, OrdKeyBuilder};
 
 /// A trace implementation using a spine of ordered lists.
 pub type OrdValSpine<K, V, T, R> = Spine<Rc<OrdValBatch<Vector<((K,V),T,R)>>>>;
+/// An `Arc`-backed variant of [`OrdValSpine`].
+///
+/// Its batches are `Arc`'d rather than `Rc`'d, so a batch whose contents are `Send + Sync` can be
+/// read from a thread other than the one maintaining the trace. Prefer [`OrdValSpine`] unless that
+/// cross-thread sharing is needed, since atomic reference counting is marginally more expensive.
+pub type ArcOrdValSpine<K, V, T, R> = Spine<Arc<OrdValBatch<Vector<((K,V),T,R)>>>>;
 /// A batcher using ordered lists.
 pub type OrdValBatcher<K, V, T, R> = MergeBatcher<VecMerger<(K, V), T, R>>;
 /// A builder using ordered lists.
 pub type RcOrdValBuilder<K, V, T, R> = RcBuilder<OrdValBuilder<Vector<((K,V),T,R)>, Vec<((K,V),T,R)>>>;
+/// An `Arc`-backed variant of [`RcOrdValBuilder`], pairing with [`ArcOrdValSpine`].
+pub type ArcOrdValBuilder<K, V, T, R> = ArcBuilder<OrdValBuilder<Vector<((K,V),T,R)>, Vec<((K,V),T,R)>>>;
 
 /// A trace implementation using a spine of ordered lists.
 pub type OrdKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<Vector<((K,()),T,R)>>>>;
+/// An `Arc`-backed variant of [`OrdKeySpine`], readable from other threads. See [`ArcOrdValSpine`].
+pub type ArcOrdKeySpine<K, T, R> = Spine<Arc<OrdKeyBatch<Vector<((K,()),T,R)>>>>;
 /// A batcher for ordered lists.
 pub type OrdKeyBatcher<K, T, R> = MergeBatcher<VecMerger<(K, ()), T, R>>;
 /// A builder for ordered lists.
 pub type RcOrdKeyBuilder<K, T, R> = RcBuilder<OrdKeyBuilder<Vector<((K,()),T,R)>, Vec<((K,()),T,R)>>>;
+/// An `Arc`-backed variant of [`RcOrdKeyBuilder`], pairing with [`ArcOrdKeySpine`].
+pub type ArcOrdKeyBuilder<K, T, R> = ArcBuilder<OrdKeyBuilder<Vector<((K,()),T,R)>, Vec<((K,()),T,R)>>>;
 
 pub use layers::{Vals, Upds};
 /// Layers are containers of lists of some type.

differential-dataflow/src/trace/mod.rs

@@ -437,3 +437,122 @@ pub mod rc_blanket_impls {
         fn done(self) -> Rc<B> { Rc::new(self.merger.done()) }
     }
 }
+
+/// Blanket implementations for atomically reference counted batches.
+///
+/// These mirror [`rc_blanket_impls`], but use `Arc` so that batches whose contents
+/// are `Send + Sync` can be shared across threads. This enables reading a trace's
+/// batches from outside the worker that maintains the trace, for example to import
+/// a snapshot of an arrangement into a dataflow running elsewhere.
+pub mod arc_blanket_impls {
+
+    use std::sync::Arc;
+
+    use timely::progress::{Antichain, frontier::AntichainRef};
+    use super::{Batch, BatchReader, Builder, Merger, Navigable, Cursor, Description};
+
+    impl<B: BatchReader + Navigable> Navigable for Arc<B> {
+        /// The type used to enumerate the batch's contents.
+        type Cursor = ArcBatchCursor<B::Cursor>;
+        /// Acquires a cursor to the batch's contents.
+        fn cursor(&self) -> Self::Cursor {
+            ArcBatchCursor::new((**self).cursor())
+        }
+    }
+
+    impl<B: BatchReader> BatchReader for Arc<B> {
+
+        type Time = B::Time;
+        /// The number of updates in the batch.
+        fn len(&self) -> usize { (**self).len() }
+        /// Describes the times of the updates in the batch.
+        fn description(&self) -> &Description<Self::Time> { (**self).description() }
+    }
+
+    /// Wrapper to provide cursor to nested scope.
+    pub struct ArcBatchCursor<C> {
+        cursor: C,
+    }
+
+    impl<C> ArcBatchCursor<C> {
+        fn new(cursor: C) -> Self {
+            ArcBatchCursor {
+                cursor,
+            }
+        }
+    }
+
+    impl<C: Cursor> Cursor for ArcBatchCursor<C> {
+
+        type Storage = Arc<C::Storage>;
+
+        type Key<'a> = C::Key<'a>;
+        type ValOwn = C::ValOwn;
+        type Val<'a> = C::Val<'a>;
+        type Time = C::Time;
+        type TimeGat<'a> = C::TimeGat<'a>;
+        type Diff = C::Diff;
+        type DiffGat<'a> = C::DiffGat<'a>;
+        type KeyContainer = C::KeyContainer;
+        type ValContainer = C::ValContainer;
+        type TimeContainer = C::TimeContainer;
+        type DiffContainer = C::DiffContainer;
+
+        #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) }
+        #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) }
+
+        #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) }
+        #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) }
+
+        #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { self.cursor.get_key(storage) }
+        #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(storage) }
+
+        #[inline]
+        fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L) {
+            self.cursor.map_times(storage, logic)
+        }
+
+        #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) }
+        #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) }
+
+        #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) }
+        #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) }
+
+        #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) }
+        #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) }
+    }
+
+    /// An immutable collection of updates.
+    impl<B: Batch> Batch for Arc<B> {
+        type Merger = ArcMerger<B>;
+        fn empty(lower: Antichain<Self::Time>, upper: Antichain<Self::Time>) -> Self {
+            Arc::new(B::empty(lower, upper))
+        }
+    }
+
+    /// Wrapper type for building atomically reference counted batches.
+    pub struct ArcBuilder<B: Builder> { builder: B }
+
+    /// Functionality for building batches from ordered update sequences.
+    impl<B: Builder> Builder for ArcBuilder<B> {
+        type Input = B::Input;
+        type Time = B::Time;
+        type Output = Arc<B::Output>;
+        fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self { ArcBuilder { builder: B::with_capacity(keys, vals, upds) } }
+        fn push(&mut self, input: &mut Self::Input) { self.builder.push(input) }
+        fn done(self, description: Description<Self::Time>) -> Arc<B::Output> { Arc::new(self.builder.done(description)) }
+        fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
+            Arc::new(B::seal(chain, description))
+        }
+    }
+
+    /// Wrapper type for merging atomically reference counted batches.
+    pub struct ArcMerger<B:Batch> { merger: B::Merger }
+
+    /// Represents a merge in progress.
+    impl<B:Batch> Merger<Arc<B>> for ArcMerger<B> {
+        fn new(source1: &Arc<B>, source2: &Arc<B>, compaction_frontier: AntichainRef<B::Time>) -> Self { ArcMerger { merger: B::begin_merge(source1, source2, compaction_frontier) } }
+        fn work(&mut self, source1: &Arc<B>, source2: &Arc<B>, fuel: &mut isize) { self.merger.work(source1, source2, fuel) }
+        fn done(self) -> Arc<B> { Arc::new(self.merger.done()) }
+    }
+}