feat(recycling): add customizable recycling policies (#33)

## Motivation

Currently, all queues and channels in `thingbuf` require that items in
the queue/channel implement `Default`, because `Default` is used to fill
slots when they are initially allocated. Furthermore, when slots are
checked out for writing to, they are not cleared prior to being written
to --- the user code is responsible for clearing them if needed.

The `StringBuf` type currently implements special behavior
_specifically_ for `String`s, where the `String` is cleared in place
prior to writing to, but this only works for `String`s. `StringBuf` also
provides an API for limiting the maximum capacity of "empty" strings, so
that they can be shrunk down to that capacity when returning them to the
pool. This allows introducing an upper bound on the capacity allocated
by unused strings. However, again, this only works with `String`s and is
only provided by the `StringBuf` type.

This isn't ideal --- users shouldn't _have_ to be responsible for
clearing non-`String` types when reusing allocations.

## Solution

This branch introduces a new `Recycle<T>` trait that defines a policy
for how `T`-typed pooled objects should be reused. `Recycle<T>` defines
two methods:

* `fn new_element(&self) -> T` creates a new element
* `fn recycle(&self, element: &mut T)` clears a pooled element for reuse

This allows a `Recycle` implementation to define the lifecycle of a
pooled item.

In addition, we define a couple of pre-made `Recycle` implementations:

* `DefaultRecycle`, which implements `Recycle` for all types `T` where
  `T: Default + Clone`. This is used by all `thingbuf` types by default.
  It creates new elements using `Default::default`, and recycles them
  using `element.clone_from(T::default())`.

  `Clone::clone_from` is not _guaranteed_ to re-use existing capacity,
  but it's overridden by most array-based collections (such as the ones
  in the standard library) to do so --- it should be equivalent to
  `.clear()` when cloning from an empty collection. However, this policy
  will still *work* with types that don't have a clear-in-place
  function.

* `WithCapacity` implements `Recycle` only for types that define
  `with_capacity`, `shrink_to`, and `clear` methods, like all
  array-based collections in the Rust standard library. Unlike
  `DefaultRecycle`, it is _guaranteed_ to clear elements in place and
  retain any previously allocated capacity.

  It can also be configured to add both upper and lower bounds on
  capacity. When there is a lower bound, new elements are allocated with
  that value as their initial capacity, rather than being allocated with
  0 capacity. When an upper bound is set, it will call `shrink_to` prior
  to clearing elements, to limit the total allocated capacity retained
  by the pool.

  `WithCapacity` currently implements `Recycle` for all `alloc` and
  `std` types that define the requisite methods: `Vec`, `String`,
  `VecDeque`, and `BinaryHeap` when the `alloc` feature is enabled, and
  `HashMap` and `HashSet` as well, when the `std` feature is enabled.

Finally, I've modified the existing queue and channel types to allow
configuring them to use a `Recycle` implementation. The `StringBuf` type
is removed, as it's now obviated by the new APIs.

## Future Work

We may wish to factor out the `recycling` module into its own crate, so
that it can be used in other libraries.

Closes #30

Signed-off-by: Eliza Weisman <[email protected]>

src/lib.rs

@@ -2,36 +2,36 @@
 #![cfg_attr(not(feature = "std"), no_std)]
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 use core::{cmp, fmt, mem::MaybeUninit, ops, ptr};
-
 #[macro_use]
 mod macros;
 
 mod loom;
-mod recycle;
+pub mod mpsc;
+pub mod recycling;
 mod util;
 mod wait;
 
+pub use self::recycling::Recycle;
+
 #[cfg_attr(docsrs, doc = include_str!("../mpsc_perf_comparison.md"))]
 pub mod mpsc_perf_comparison {
     // Empty module, used only for documentation.
 }
 
+feature! {
+    #![not(all(loom, test))]
+    mod static_thingbuf;
+    pub use self::static_thingbuf::StaticThingBuf;
+}
+
 feature! {
     #![feature = "alloc"]
     extern crate alloc;
 
     mod thingbuf;
     pub use self::thingbuf::ThingBuf;
-
-    mod stringbuf;
-    pub use stringbuf::{StaticStringBuf, StringBuf};
 }
 
-pub mod mpsc;
-
-mod static_thingbuf;
-pub use self::static_thingbuf::StaticThingBuf;
-
 use crate::{
     loom::{
         atomic::{AtomicUsize, Ordering::*},
@@ -97,7 +97,6 @@ impl Core {
             closed,
             idx_mask,
             capacity,
-
             has_dropped_slots: false,
         }
     }
@@ -116,7 +115,6 @@ impl Core {
             gen_mask,
             idx_mask,
             capacity,
-
             #[cfg(debug_assertions)]
             has_dropped_slots: false,
         }
@@ -155,12 +153,13 @@ impl Core {
     }
 
     #[inline(always)]
-    fn push_ref<'slots, T, S>(
+    fn push_ref<'slots, T, S, R>(
         &self,
         slots: &'slots S,
+        recycle: &R,
     ) -> Result<Ref<'slots, T>, mpsc::TrySendError<()>>
     where
-        T: Default,
+        R: Recycle<T>,
         S: ops::Index<usize, Output = Slot<T>> + ?Sized,
     {
         test_println!("push_ref");
@@ -190,13 +189,20 @@ impl Core {
                         // Claim exclusive ownership over the slot
                         let ptr = slot.value.get_mut();
 
-                        if gen == 0 {
-                            unsafe {
-                                // Safety: we have just claimed exclusive ownership over
-                                // this slot.
-                                ptr.deref().write(T::default());
-                            };
-                            test_println!("-> initialized");
+                        // Initialize or recycle the element.
+                        unsafe {
+                            // Safety: we have just claimed exclusive ownership over
+                            // this slot.
+                            let ptr = ptr.deref();
+                            if gen == 0 {
+                                ptr.write(recycle.new_element());
+                                test_println!("-> initialized");
+                            } else {
+                                // Safety: if the generation is > 0, then the
+                                // slot has already been initialized.
+                                recycle.recycle(ptr.assume_init_mut());
+                                test_println!("-> recycled");
+                            }
                         }
 
                         return Ok(Ref {

src/loom.rs

@@ -187,6 +187,16 @@ mod inner {
                 Self::new(T::default())
             }
         }
+
+        impl<T: Clone> Clone for Track<T> {
+            fn clone(&self) -> Self {
+                Self::new(self.get_ref().clone())
+            }
+
+            fn clone_from(&mut self, source: &Self) {
+                self.get_mut().clone_from(source.get_ref());
+            }
+        }
     }
 }
 

src/mpsc.rs

@@ -12,6 +12,7 @@
 
 use crate::{
     loom::{atomic::AtomicUsize, hint},
+    recycling::Recycle,
     wait::{Notify, WaitCell, WaitQueue, WaitResult},
     Core, Ref, Slot,
 };
@@ -109,24 +110,25 @@ impl<N> ChannelCore<N>
 where
     N: Notify + Unpin,
 {
-    fn try_send_ref<'a, T>(
+    fn try_send_ref<'a, T, R>(
         &'a self,
         slots: &'a [Slot<T>],
+        recycle: &R,
     ) -> Result<SendRefInner<'a, T, N>, TrySendError>
     where
-        T: Default,
+        R: Recycle<T>,
     {
-        self.core.push_ref(slots).map(|slot| SendRefInner {
+        self.core.push_ref(slots, recycle).map(|slot| SendRefInner {
             _notify: NotifyRx(&self.rx_wait),
             slot,
         })
     }
 
-    fn try_send<T>(&self, slots: &[Slot<T>], val: T) -> Result<(), TrySendError<T>>
+    fn try_send<T, R>(&self, slots: &[Slot<T>], val: T, recycle: &R) -> Result<(), TrySendError<T>>
     where
-        T: Default,
+        R: Recycle<T>,
     {
-        match self.try_send_ref(slots) {
+        match self.try_send_ref(slots, recycle) {
             Ok(mut slot) => {
                 slot.with_mut(|slot| *slot = val);
                 Ok(())
@@ -147,7 +149,6 @@ where
     ) -> Poll<Option<Ref<'a, T>>>
     where
         S: Index<usize, Output = Slot<T>> + ?Sized,
-        T: Default,
     {
         macro_rules! try_poll_recv {
             () => {