Replace VecChunker with ContainerChunker

Author: frankmcsherry

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

@@ -10,120 +10,6 @@ use crate::containers::TimelyStack;
 use crate::consolidation::{consolidate_updates, Consolidate};
 use crate::difference::Semigroup;
 
-/// Chunk a stream of vectors into chains of vectors.
-pub struct VecChunker<T> {
-    pending: Vec<T>,
-    ready: VecDeque<Vec<T>>,
-    empty: Option<Vec<T>>,
-}
-
-impl<T> Default for VecChunker<T> {
-    fn default() -> Self {
-        Self {
-            pending: Vec::default(),
-            ready: VecDeque::default(),
-            empty: None,
-        }
-    }
-}
-
-impl<K, V, T, R> VecChunker<((K, V), T, R)>
-where
-    K: Ord,
-    V: Ord,
-    T: Ord,
-    R: Semigroup,
-{
-    const BUFFER_SIZE_BYTES: usize = 8 << 10;
-    fn chunk_capacity() -> usize {
-        let size = ::std::mem::size_of::<((K, V), T, R)>();
-        if size == 0 {
-            Self::BUFFER_SIZE_BYTES
-        } else if size <= Self::BUFFER_SIZE_BYTES {
-            Self::BUFFER_SIZE_BYTES / size
-        } else {
-            1
-        }
-    }
-
-    /// Form chunks out of pending data, if needed. This function is meant to be applied to
-    /// potentially full buffers, and ensures that if the buffer was full when called it is at most
-    /// half full when the function returns.
-    ///
-    /// `form_chunk` does the following:
-    /// * If pending is full, consolidate.
-    /// * If after consolidation it's more than half full, peel off chunks,
-    ///   leaving behind any partial chunk in pending.
-    fn form_chunk(&mut self) {
-        consolidate_updates(&mut self.pending);
-        if self.pending.len() >= Self::chunk_capacity() {
-            while self.pending.len() > Self::chunk_capacity() {
-                let mut chunk = Vec::with_capacity(Self::chunk_capacity());
-                chunk.extend(self.pending.drain(..chunk.capacity()));
-                self.ready.push_back(chunk);
-            }
-        }
-    }
-}
-
-impl<'a, K, V, T, R> PushInto<&'a mut Vec<((K, V), T, R)>> for VecChunker<((K, V), T, R)>
-where
-    K: Ord + Clone,
-    V: Ord + Clone,
-    T: Ord + Clone,
-    R: Semigroup + Clone,
-{
-    fn push_into(&mut self, container: &'a mut Vec<((K, V), T, R)>) {
-        // Ensure `self.pending` has the desired capacity. We should never have a larger capacity
-        // because we don't write more than capacity elements into the buffer.
-        // Important: Consolidation requires `pending` to have twice the chunk capacity to
-        // amortize its cost. Otherwise, it risks to do quadratic work.
-        if self.pending.capacity() < Self::chunk_capacity() * 2 {
-            self.pending.reserve(Self::chunk_capacity() * 2 - self.pending.len());
-        }
-
-        let mut drain = container.drain(..).peekable();
-        while drain.peek().is_some() {
-            self.pending.extend((&mut drain).take(self.pending.capacity() - self.pending.len()));
-            if self.pending.len() == self.pending.capacity() {
-                self.form_chunk();
-            }
-        }
-    }
-}
-
-impl<K, V, T, R> ContainerBuilder for VecChunker<((K, V), T, R)>
-where
-    K: Ord + Clone + 'static,
-    V: Ord + Clone + 'static,
-    T: Ord + Clone + 'static,
-    R: Semigroup + Clone + 'static,
-{
-    type Container = Vec<((K, V), T, R)>;
-
-    fn extract(&mut self) -> Option<&mut Self::Container> {
-        if let Some(ready) = self.ready.pop_front() {
-            self.empty = Some(ready);
-            self.empty.as_mut()
-        } else {
-            None
-        }
-    }
-
-    fn finish(&mut self) -> Option<&mut Self::Container> {
-        if !self.pending.is_empty() {
-            consolidate_updates(&mut self.pending);
-            while !self.pending.is_empty() {
-                let mut chunk = Vec::with_capacity(Self::chunk_capacity());
-                chunk.extend(self.pending.drain(..std::cmp::min(self.pending.len(), chunk.capacity())));
-                self.ready.push_back(chunk);
-            }
-        }
-        self.empty = self.ready.pop_front();
-        self.empty.as_mut()
-    }
-}
-
 /// Chunk a stream of vectors into chains of vectors.
 pub struct ColumnationChunker<T: Columnation> {
     pending: Vec<T>,

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

@@ -11,7 +11,7 @@
 use std::rc::Rc;
 
 use crate::containers::TimelyStack;
-use crate::trace::implementations::chunker::{ColumnationChunker, VecChunker};
+use crate::trace::implementations::chunker::{ColumnationChunker, ContainerChunker};
 use crate::trace::implementations::spine_fueled::Spine;
 use crate::trace::implementations::merge_batcher::MergeBatcher;
 use crate::trace::implementations::merge_batcher::container::{VecInternalMerger, ColInternalMerger};
@@ -25,7 +25,7 @@ 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)>>>>;
 /// A batcher using ordered lists.
-pub type OrdValBatcher<K, V, T, R> = MergeBatcher<Vec<((K,V),T,R)>, VecChunker<((K,V),T,R)>, VecInternalMerger<(K, V), T, R>>;
+pub type OrdValBatcher<K, V, T, R> = MergeBatcher<Vec<((K,V),T,R)>, ContainerChunker<Vec<((K,V),T,R)>>, VecInternalMerger<(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)>>>;
 
@@ -42,7 +42,7 @@ pub type ColValBuilder<K, V, T, R> = RcBuilder<OrdValBuilder<TStack<((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)>>>>;
 /// A batcher for ordered lists.
-pub type OrdKeyBatcher<K, T, R> = MergeBatcher<Vec<((K,()),T,R)>, VecChunker<((K,()),T,R)>, VecInternalMerger<(K, ()), T, R>>;
+pub type OrdKeyBatcher<K, T, R> = MergeBatcher<Vec<((K,()),T,R)>, ContainerChunker<Vec<((K,()),T,R)>>, VecInternalMerger<(K, ()), T, R>>;
 /// A builder for ordered lists.
 pub type RcOrdKeyBuilder<K, T, R> = RcBuilder<OrdKeyBuilder<Vector<((K,()),T,R)>, Vec<((K,()),T,R)>>>;
 

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

@@ -12,7 +12,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::Hashable;
 use crate::containers::TimelyStack;
-use crate::trace::implementations::chunker::{ColumnationChunker, VecChunker};
+use crate::trace::implementations::chunker::{ColumnationChunker, ContainerChunker};
 use crate::trace::implementations::merge_batcher::MergeBatcher;
 use crate::trace::implementations::merge_batcher::container::{VecInternalMerger, ColInternalMerger};
 use crate::trace::implementations::spine_fueled::Spine;
@@ -25,7 +25,7 @@ use self::val_batch::{RhhValBatch, RhhValBuilder};
 /// A trace implementation using a spine of ordered lists.
 pub type VecSpine<K, V, T, R> = Spine<Rc<RhhValBatch<Vector<((K,V),T,R)>>>>;
 /// A batcher for ordered lists.
-pub type VecBatcher<K,V,T,R> = MergeBatcher<Vec<((K,V),T,R)>, VecChunker<((K,V),T,R)>, VecInternalMerger<(K, V), T, R>>;
+pub type VecBatcher<K,V,T,R> = MergeBatcher<Vec<((K,V),T,R)>, ContainerChunker<Vec<((K,V),T,R)>>, VecInternalMerger<(K, V), T, R>>;
 /// A builder for ordered lists.
 pub type VecBuilder<K,V,T,R> = RcBuilder<RhhValBuilder<Vector<((K,V),T,R)>, Vec<((K,V),T,R)>>>;