Reorganize dirctories

Author: jeehoonkang

crossbeam-circbuf/src/lib.rs

@@ -12,34 +12,13 @@ extern crate crossbeam_epoch as epoch;
 extern crate crossbeam_utils as utils;
 
 mod buffer;
+mod try_recv;
 
 #[doc(hidden)] // for doc-tests
-pub mod mp;
-#[doc(hidden)] // for doc-tests
-pub mod sp;
-
-pub use mp::mc as mpmc;
-pub use sp::mc as spmc;
-pub use sp::sc as spsc;
+pub mod sp_inner;
 
-/// The return type for `try_recv` methods.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum TryRecv<T> {
-    /// Received a value.
-    Data(T),
-    /// Not received a value because the buffer is empty.
-    Empty,
-    /// Lost the race to a concurrent operation. Try again.
-    Retry,
-}
+pub use try_recv::TryRecv;
 
-impl<T> TryRecv<T> {
-    /// Applies a function to the content of `TryRecv::Data`.
-    pub fn map<U, F: FnOnce(T) -> U>(self, f: F) -> TryRecv<U> {
-        match self {
-            TryRecv::Data(v) => TryRecv::Data(f(v)),
-            TryRecv::Empty => TryRecv::Empty,
-            TryRecv::Retry => TryRecv::Retry,
-        }
-    }
-}
+pub mod mpmc;
+pub mod spmc;
+pub mod spsc;

crossbeam-circbuf/src/mp/mod.rs

@@ -1,4 +1,53 @@
 //! Internal data structures used in single-producer queues.
+//!
+//! TODO: move
+//!
+//! Concurrent single-producer queues based on circular buffer.
+//!
+//! [`CircBuf`] is a circular buffer, which is basically a fixed-sized array that has two ends: tx
+//! and rx. A [`CircBuf`] can [`send`] values into the tx end and [`CircBuf::try_recv`] values from
+//! the rx end. A [`CircBuf`] doesn't implement `Sync` so it cannot be shared among multiple
+//! threads. However, it can create [`Receiver`]s, and those can be easily cloned, shared, and sent
+//! to other threads. [`Receiver`]s can only [`Receiver::try_recv`] values from the rx end.
+//!
+//! Here's a visualization of a [`CircBuf`] of capacity 4, consisting of 2 values `a` and `b`.
+//!
+//! ```text
+//!    ___
+//!   | a | <- rx (CircBuf::try_recv, Receiver::try_recv)
+//!   | b |
+//!   |   | <- tx (CircBuf::send)
+//!   |   |
+//!    ¯¯¯
+//! ```
+//!
+//! [`DynamicCircBuf`] is a dynamically growable and shrinkable circular buffer. Internally,
+//! [`DynamicCircBuf`] has a [`CircBuf`] and resizes it when necessary.
+//!
+//!
+//! # Usage: fair work-stealing schedulers
+//!
+//! This data structure can be used in fair work-stealing schedulers for multiple threads as
+//! follows.
+//!
+//! Each thread owns a [`CircBuf`] (or [`DynamicCircBuf`]) and creates a [`Receiver`] that is shared
+//! among all other threads (or creates one [`Receiver`] for each of the other threads).
+//!
+//! Each thread is executing a loop in which it attempts to [`CircBuf::try_recv`] some task from its
+//! own [`CircBuf`] and perform it. If the buffer is empty, it attempts to [`Receiver::try_recv`]
+//! work from other threads instead. When performing a task, a thread may produce more tasks by
+//! [`send`]ing them to its buffer.
+//!
+//! It is worth noting that it is discouraged to use work-stealing deque for fair schedulers,
+//! because its `pop()` may return a task that is just `push()`ed, effectively scheduling the same
+//! work repeatedly.
+//!
+//! [`CircBuf`]: struct.CircBuf.html
+//! [`DynamicCircBuf`]: struct.DynamicCircBuf.html
+//! [`Receiver`]: struct.Receiver.html
+//! [`send`]: struct.CircBuf.html#method.send
+//! [`CircBuf::try_recv`]: struct.CircBuf.html#method.try_recv
+//! [`Receiver::try_recv`]: struct.Receiver.html#method.try_recv
 
 use std::cell::Cell;
 use std::fmt;
@@ -104,7 +153,7 @@ impl<T> CircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::CircBuf;
+    /// use crossbeam_circbuf::sp_inner::CircBuf;
     ///
     /// // The capacity will be rounded up to 1024.
     /// let cb = CircBuf::<i32>::new(1000);
@@ -128,7 +177,7 @@ impl<T> CircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::CircBuf;
+    /// use crossbeam_circbuf::sp_inner::CircBuf;
     ///
     /// let cb = CircBuf::new(16);
     /// cb.send(1).unwrap();
@@ -147,7 +196,7 @@ impl<T> CircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::{CircBuf, TryRecv};
+    /// use crossbeam_circbuf::sp_inner::{CircBuf, TryRecv};
     ///
     /// let cb = CircBuf::new(16);
     /// cb.send(1).unwrap();
@@ -268,7 +317,7 @@ impl<T> CircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::{CircBuf, TryRecv};
+    /// use crossbeam_circbuf::sp_inner::{CircBuf, TryRecv};
     /// use std::thread;
     ///
     /// let cb = CircBuf::new(16);
@@ -351,7 +400,7 @@ impl<T> DynamicCircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::DynamicCircBuf;
+    /// use crossbeam_circbuf::sp_inner::DynamicCircBuf;
     ///
     /// let cb = DynamicCircBuf::<i32>::new();
     /// ```
@@ -367,7 +416,7 @@ impl<T> DynamicCircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::DynamicCircBuf;
+    /// use crossbeam_circbuf::sp_inner::DynamicCircBuf;
     ///
     /// // The minimum capacity will be rounded up to 1024.
     /// let cb = DynamicCircBuf::<i32>::with_min_capacity(1000);
@@ -391,7 +440,7 @@ impl<T> DynamicCircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::DynamicCircBuf;
+    /// use crossbeam_circbuf::sp_inner::DynamicCircBuf;
     ///
     /// let cb = DynamicCircBuf::new();
     /// cb.send(1);
@@ -430,7 +479,7 @@ impl<T> DynamicCircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::{DynamicCircBuf, TryRecv};
+    /// use crossbeam_circbuf::sp_inner::{DynamicCircBuf, TryRecv};
     ///
     /// let cb = DynamicCircBuf::new();
     /// cb.send(1);
@@ -464,7 +513,7 @@ impl<T> DynamicCircBuf<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::{DynamicCircBuf, TryRecv};
+    /// use crossbeam_circbuf::sp_inner::{DynamicCircBuf, TryRecv};
     /// use std::thread;
     ///
     /// let cb = DynamicCircBuf::new();
@@ -558,7 +607,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::{DynamicCircBuf, TryRecv};
+    /// use crossbeam_circbuf::sp_inner::{DynamicCircBuf, TryRecv};
     ///
     /// let cb = DynamicCircBuf::new();
     /// let r = cb.receiver();
@@ -615,7 +664,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::{DynamicCircBuf, TryRecv};
+    /// use crossbeam_circbuf::sp_inner::{DynamicCircBuf, TryRecv};
     ///
     /// let cb = DynamicCircBuf::new();
     /// let r = cb.receiver();
@@ -706,7 +755,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
-    /// use crossbeam_circbuf::sp::internal::{DynamicCircBuf, TryRecv};
+    /// use crossbeam_circbuf::sp_inner::{DynamicCircBuf, TryRecv};
     ///
     /// let cb = DynamicCircBuf::new();
     /// let r = cb.receiver();

crossbeam-circbuf/src/mp/mc.rs crossbeam-circbuf/src/mpmc.rs

@@ -24,16 +24,16 @@
 /// }).join().unwrap();
 /// ```
 pub mod bounded {
-    use super::super::internal;
+    use crate::sp_inner;
     pub use TryRecv;
 
     /// A bounded SPMC queue.
     #[derive(Debug)]
-    pub struct Queue<T>(internal::CircBuf<T>);
+    pub struct Queue<T>(sp_inner::CircBuf<T>);
 
     /// The receiver of a bounded SPMC queue.
     #[derive(Debug)]
-    pub struct Receiver<T>(internal::Receiver<T>);
+    pub struct Receiver<T>(sp_inner::Receiver<T>);
 
     impl<T> Queue<T> {
         /// Creates a bounded SPMC queue with the specified capacity.
@@ -49,7 +49,7 @@ pub mod bounded {
         /// ```
         pub fn new(cap: usize) -> Self {
             Queue {
-                0: internal::CircBuf::new(cap),
+                0: sp_inner::CircBuf::new(cap),
             }
         }
 
@@ -202,16 +202,16 @@ pub mod bounded {
 /// }).join().unwrap();
 /// ```
 pub mod unbounded {
-    use super::super::internal;
+    use crate::sp_inner;
     pub use TryRecv;
 
     /// an unbounded SPMC queue.
     #[derive(Debug)]
-    pub struct Queue<T>(internal::DynamicCircBuf<T>);
+    pub struct Queue<T>(sp_inner::DynamicCircBuf<T>);
 
     /// The receiver of an unbounded SPMC queue.
     #[derive(Debug)]
-    pub struct Receiver<T>(internal::Receiver<T>);
+    pub struct Receiver<T>(sp_inner::Receiver<T>);
 
     impl<T> Queue<T> {
         /// Creates an unbounded SPMC queue.
@@ -225,7 +225,7 @@ pub mod unbounded {
         /// ```
         pub fn new() -> Self {
             Queue {
-                0: internal::DynamicCircBuf::new(),
+                0: sp_inner::DynamicCircBuf::new(),
             }
         }
 
@@ -243,7 +243,7 @@ pub mod unbounded {
         /// ```
         pub fn with_min_capacity(min_cap: usize) -> Self {
             Queue {
-                0: internal::DynamicCircBuf::with_min_capacity(min_cap),
+                0: sp_inner::DynamicCircBuf::with_min_capacity(min_cap),
             }
         }
 

crossbeam-circbuf/src/sp/mod.rs

@@ -23,16 +23,16 @@
 /// }).join().unwrap();
 /// ```
 pub mod bounded {
-    use sp::internal;
+    use crate::sp_inner;
     pub use TryRecv;
 
     /// The sender of a bounded SPSC queue.
     #[derive(Debug)]
-    pub struct Sender<T>(internal::CircBuf<T>);
+    pub struct Sender<T>(sp_inner::CircBuf<T>);
 
     /// The receiver of a bounded SPSC queue.
     #[derive(Debug)]
-    pub struct Receiver<T>(internal::Receiver<T>);
+    pub struct Receiver<T>(sp_inner::Receiver<T>);
 
     unsafe impl<T> Send for Receiver<T> {}
 
@@ -49,7 +49,7 @@ pub mod bounded {
     /// let (tx, rx) = spsc::new::<u32>(16);
     /// ```
     pub fn new<T>(cap: usize) -> (Sender<T>, Receiver<T>) {
-        let circbuf = internal::CircBuf::new(cap);
+        let circbuf = sp_inner::CircBuf::new(cap);
         let receiver = circbuf.receiver();
         let sender = Sender { 0: circbuf };
         let receiver = Receiver { 0: receiver };
@@ -121,16 +121,16 @@ pub mod bounded {
 /// }).join().unwrap();
 /// ```
 pub mod unbounded {
-    use sp::internal;
+    use crate::sp_inner;
     pub use TryRecv;
 
     /// The sender of an unbounded SPSC queue.
     #[derive(Debug)]
-    pub struct Sender<T>(internal::DynamicCircBuf<T>);
+    pub struct Sender<T>(sp_inner::DynamicCircBuf<T>);
 
     /// The receiver of an unbounded SPSC queue.
     #[derive(Debug)]
-    pub struct Receiver<T>(internal::Receiver<T>);
+    pub struct Receiver<T>(sp_inner::Receiver<T>);
 
     unsafe impl<T> Send for Receiver<T> {}
 
@@ -144,7 +144,7 @@ pub mod unbounded {
     /// let (tx, rx) = spsc::new::<u32>();
     /// ```
     pub fn new<T>() -> (Sender<T>, Receiver<T>) {
-        let circbuf = internal::DynamicCircBuf::new();
+        let circbuf = sp_inner::DynamicCircBuf::new();
         let receiver = circbuf.receiver();
         let sender = Sender { 0: circbuf };
         let receiver = Receiver { 0: receiver };
@@ -165,7 +165,7 @@ pub mod unbounded {
     /// let (tx, rx) = spsc::with_min_capacity::<u32>(1000);
     /// ```
     pub fn with_min_capacity<T>(min_cap: usize) -> (Sender<T>, Receiver<T>) {
-        let circbuf = internal::DynamicCircBuf::with_min_capacity(min_cap);
+        let circbuf = sp_inner::DynamicCircBuf::with_min_capacity(min_cap);
         let receiver = circbuf.receiver();
         let sender = Sender { 0: circbuf };
         let receiver = Receiver { 0: receiver };