Share pipeline resources through worker allocator

communication/src/allocator/generic.rs

@@ -98,7 +98,16 @@ impl Allocator {
         (crate::allocator::thread::ThreadPusher<T>,
          crate::allocator::thread::ThreadPuller<T>)
     {
-        crate::allocator::thread::Thread::new_from(identifier, Rc::clone(self.events()))
+        let recycler = match self {
+            Allocator::Thread(t) => t.recycler(),
+            Allocator::Process(p) => p.recycler(),
+            Allocator::Tcp(z) => z.recycler(),
+        };
+        crate::allocator::thread::Thread::new_from_with_recycler(
+            identifier,
+            Rc::clone(self.events()),
+            recycler,
+        )
     }
 }
 

communication/src/allocator/mod.rs

@@ -232,4 +232,10 @@ impl Process {
             Process::Bytes(pb) => pb.await_events(duration),
         }
     }
+    pub(crate) fn recycler(&self) -> thread::RecyclerHandle {
+        match self {
+            Process::Typed(p) => p.recycler(),
+            Process::Bytes(pb) => pb.recycler(),
+        }
+    }
 }

communication/src/allocator/process.rs

@@ -70,6 +70,10 @@ pub struct Process {
 impl Process {
     /// Access the wrapped inner allocator.
     pub fn inner(&mut self) -> &mut Thread { &mut self.inner }
+
+    pub(crate) fn recycler(&self) -> crate::allocator::thread::RecyclerHandle {
+        self.inner.recycler()
+    }
 }
 
 impl PeerBuilder for Process {

communication/src/allocator/thread.rs

@@ -3,7 +3,8 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 use std::time::Duration;
-use std::collections::VecDeque;
+use std::any::{Any, TypeId};
+use std::collections::{HashMap, VecDeque};
 
 use crate::allocator::{Allocate, AllocateBuilder};
 use crate::allocator::counters::Pusher as CountPusher;
@@ -24,13 +25,19 @@ impl AllocateBuilder for ThreadBuilder {
 pub struct Thread {
     /// Shared counts of messages in channels.
     events: Rc<RefCell<Vec<usize>>>,
+    /// Reusable values shared by all pipeline channels allocated by this worker.
+    recycler: RecyclerHandle,
 }
 
 impl Allocate for Thread {
     fn index(&self) -> usize { 0 }
     fn peers(&self) -> usize { 1 }
     fn allocate<T: 'static>(&mut self, identifier: usize) -> (Vec<Box<dyn Push<T>>>, Box<dyn Pull<T>>) {
-        let (pusher, puller) = Thread::new_from(identifier, Rc::clone(&self.events));
+        let (pusher, puller) = Thread::new_from_with_recycler(
+            identifier,
+            Rc::clone(&self.events),
+            Rc::clone(&self.recycler),
+        );
         (vec![Box::new(pusher)], Box::new(puller))
     }
     fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
@@ -58,49 +65,173 @@ impl Thread {
     pub fn new_from<T: 'static>(identifier: usize, events: Rc<RefCell<Vec<usize>>>)
         -> (ThreadPusher<T>, ThreadPuller<T>)
     {
-        let shared = Rc::new(RefCell::new((VecDeque::<T>::new(), VecDeque::<T>::new())));
-        let pusher = Pusher { target: Rc::clone(&shared) };
+        Self::new_from_with_recycler(identifier, events, Default::default())
+    }
+
+    pub(crate) fn new_from_with_recycler<T: 'static>(
+        identifier: usize,
+        events: Rc<RefCell<Vec<usize>>>,
+        recycler: RecyclerHandle,
+    ) -> (ThreadPusher<T>, ThreadPuller<T>) {
+        let shared = Rc::new(RefCell::new(VecDeque::<T>::new()));
+        let pool = recycler.borrow_mut().pool::<T>();
+        let pusher = Pusher {
+            target: Rc::clone(&shared),
+            pool: Rc::clone(&pool),
+        };
         let pusher = CountPusher::new(pusher, identifier, Rc::clone(&events));
-        let puller = Puller { source: shared, current: None };
+        let puller = Puller { source: shared, current: None, pool };
         let puller = CountPuller::new(puller, identifier, events);
         (pusher, puller)
     }
+
+    pub(crate) fn recycler(&self) -> RecyclerHandle {
+        Rc::clone(&self.recycler)
+    }
+}
+
+/// Maximum number of reusable values retained for each concrete channel type.
+const PER_TYPE_LIMIT: usize = 2;
+
+pub(crate) type RecyclerHandle = Rc<RefCell<Recycler>>;
+
+/// Worker-owned storage shared by pipeline channels of the same concrete type.
+#[derive(Default)]
+pub(crate) struct Recycler {
+    pools: HashMap<TypeId, Box<dyn Any>>,
+}
+
+impl Recycler {
+    fn pool<T: 'static>(&mut self) -> Pool<T> {
+        Rc::clone(self
+            .pools
+            .entry(TypeId::of::<T>())
+            .or_insert_with(|| Box::new(Pool::<T>::default()))
+            .downcast_ref::<Pool<T>>()
+            .expect("TypeId must identify the recycler pool type"))
+    }
+}
+
+type Pool<T> = Rc<RefCell<Vec<T>>>;
+
+fn recycle<T>(pool: &Pool<T>, value: T) -> Result<(), T> {
+    let mut pool = pool.borrow_mut();
+    if pool.len() < PER_TYPE_LIMIT {
+        pool.push(value);
+        Ok(())
+    } else {
+        Err(value)
+    }
 }
 
 
 /// The push half of an intra-thread channel.
 pub struct Pusher<T> {
-    target: Rc<RefCell<(VecDeque<T>, VecDeque<T>)>>,
+    target: Rc<RefCell<VecDeque<T>>>,
+    pool: Pool<T>,
 }
 
-impl<T> Push<T> for Pusher<T> {
+impl<T: 'static> Push<T> for Pusher<T> {
     #[inline]
     fn push(&mut self, element: &mut Option<T>) {
-        let mut borrow = self.target.borrow_mut();
+        let sent = element.is_some();
         if let Some(element) = element.take() {
-            borrow.0.push_back(element);
+            self.target.borrow_mut().push_back(element);
+        }
+        // `Push::done` calls `push` with `None`; do not remove and immediately
+        // drop a pooled value merely to communicate that control signal.
+        if sent {
+            *element = self.pool.borrow_mut().pop();
         }
-        *element = borrow.1.pop_front();
     }
 }
 
 /// The pull half of an intra-thread channel.
-pub struct Puller<T> {
+pub struct Puller<T: 'static> {
     current: Option<T>,
-    source: Rc<RefCell<(VecDeque<T>, VecDeque<T>)>>,
+    source: Rc<RefCell<VecDeque<T>>>,
+    pool: Pool<T>,
 }
 
-impl<T> Pull<T> for Puller<T> {
+impl<T: 'static> Pull<T> for Puller<T> {
     #[inline]
     fn pull(&mut self) -> &mut Option<T> {
-        let mut borrow = self.source.borrow_mut();
-        // if let Some(element) = self.current.take() {
-        //     // TODO : Arbitrary constant.
-        //     if borrow.1.len() < 16 {
-        //         borrow.1.push_back(element);
-        //     }
-        // }
-        self.current = borrow.0.pop_front();
+        if let Some(element) = self.current.take() {
+            let _ = recycle(&self.pool, element);
+        }
+        self.current = self.source.borrow_mut().pop_front();
         &mut self.current
     }
 }
+
+impl<T: 'static> Drop for Puller<T> {
+    fn drop(&mut self) {
+        if let Some(element) = self.current.take() {
+            let _ = recycle(&self.pool, element);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn channels_of_the_same_type_share_returned_values() {
+        let allocator = Thread::default();
+        let events = Rc::clone(&allocator.events);
+        let recycler = allocator.recycler();
+        let (mut first_push, mut first_pull) = Thread::new_from_with_recycler::<Vec<u8>>(
+            0,
+            Rc::clone(&events),
+            Rc::clone(&recycler),
+        );
+        let (mut second_push, _second_pull) =
+            Thread::new_from_with_recycler::<Vec<u8>>(1, events, recycler);
+
+        let value = Vec::with_capacity(1024);
+        let allocation = value.as_ptr();
+        let mut slot = Some(value);
+        first_push.push(&mut slot);
+        assert!(slot.is_none());
+
+        first_pull.pull().as_mut().unwrap().clear();
+        assert!(first_pull.pull().is_none());
+
+        let mut second = Some(Vec::new());
+        second_push.push(&mut second);
+        let returned = second.expect("second channel should acquire the returned value");
+        assert_eq!(returned.as_ptr(), allocation);
+        assert_eq!(returned.capacity(), 1024);
+    }
+
+    #[test]
+    fn recycler_bounds_each_concrete_type() {
+        let mut recycler = Recycler::default();
+        let bytes = recycler.pool::<Vec<u8>>();
+        assert!(recycle(&bytes, Vec::new()).is_ok());
+        assert!(recycle(&bytes, Vec::new()).is_ok());
+        assert!(recycle(&bytes, Vec::new()).is_err());
+
+        let words = recycler.pool::<Vec<u64>>();
+        assert!(recycle(&words, Vec::new()).is_ok());
+        assert!(recycle(&words, Vec::new()).is_ok());
+        assert!(recycle(&words, Vec::new()).is_err());
+    }
+
+    #[test]
+    fn done_does_not_discard_a_recycled_value() {
+        let recycler = RecyclerHandle::default();
+        let pool = recycler.borrow_mut().pool::<Vec<u8>>();
+        recycle(&pool, Vec::with_capacity(1024)).unwrap();
+        let (mut push, _pull) = Thread::new_from_with_recycler::<Vec<u8>>(
+            0,
+            Rc::new(RefCell::new(Vec::new())),
+            Rc::clone(&recycler),
+        );
+
+        push.done();
+
+        assert_eq!(pool.borrow_mut().pop().unwrap().capacity(), 1024);
+    }
+}

communication/src/allocator/zero_copy/allocator.rs

@@ -149,6 +149,12 @@ pub struct TcpAllocator {
     to_local:   HashMap<usize, Rc<RefCell<VecDeque<Bytes>>>>,   // to worker-local typed pullers.
 }
 
+impl TcpAllocator {
+    pub(crate) fn recycler(&self) -> crate::allocator::thread::RecyclerHandle {
+        self.inner.recycler()
+    }
+}
+
 impl Allocate for TcpAllocator {
     fn index(&self) -> usize { self.index }
     fn peers(&self) -> usize { self.peers }

communication/src/allocator/zero_copy/allocator_process.rs

@@ -100,6 +100,7 @@ impl ProcessBuilder {
             sends,
             recvs,
             to_local: HashMap::new(),
+            recycler: Default::default(),
         }
     }
 }
@@ -130,6 +131,13 @@ pub struct ProcessAllocator {
     sends:      Vec<Rc<RefCell<SendEndpoint<MergeQueue>>>>, // sends[x] -> goes to thread x.
     recvs:      Vec<MergeQueue>,                            // recvs[x] <- from thread x.
     to_local:   HashMap<usize, Rc<RefCell<VecDeque<Bytes>>>>,          // to worker-local typed pullers.
+    recycler: crate::allocator::thread::RecyclerHandle,
+}
+
+impl ProcessAllocator {
+    pub(crate) fn recycler(&self) -> crate::allocator::thread::RecyclerHandle {
+        Rc::clone(&self.recycler)
+    }
 }
 
 impl Allocate for ProcessAllocator {