spillway docs

Author: kvc0

Cargo.lock

@@ -57,6 +57,7 @@ socket2                 = { version = "0.6" }
 thiserror               = { version = "2.0" }
 tokio                   = { version = "1.39", features = ["net", "rt"] }
 tokio-rustls            = { version = "0.26" }
+tokio-test              = { version = "0.4" }
 tokio-util              = { version = "0.7" }
 tracing                 = { version = "0.1" }
 webpki-roots            = { version = "1" }

Cargo.toml

@@ -6,11 +6,13 @@ authors.workspace = true
 repository.workspace = true
 edition.workspace = true
 license.workspace = true
-readme.workspace = true
 keywords.workspace = true
 categories.workspace = true
 
 [dependencies]
 futures                 = { workspace = true }
 log                     = { workspace = true }
 rand                    = { workspace = true }
+
+[dev-dependencies]
+tokio-test              = { workspace = true }

spillway/Cargo.toml

@@ -1,3 +1,6 @@
+#![deny(missing_docs)]
+#![doc = include_str!("../README.md")]
+
 mod receiver;
 mod sender;
 mod shared;
@@ -7,6 +10,7 @@ use std::sync::Arc;
 pub use receiver::Receiver;
 pub use sender::Sender;
 
+/// Get a new spillway channel with a default concurrency level.
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     // const PARALLELISM: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
     //     std::thread::available_parallelism()
@@ -17,6 +21,12 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     channel_with_concurrency(8)
 }
 
+/// Get a new spillway channel with the given concurrency level.
+///
+/// Use this when you need lots of parallelism, or when you know how many Senders
+/// you will have. Higher numbers reduce contention, but increase the cost of
+/// parking the Receiver when idle. Thread count is a good starting point for
+/// concurrency.
 pub fn channel_with_concurrency<T>(concurrency: usize) -> (Sender<T>, Receiver<T>) {
     let shared = Arc::new(shared::Shared::new(concurrency));
     let sender = Sender::new(shared.clone());

spillway/src/lib.rs

@@ -2,6 +2,7 @@ use std::{collections::VecDeque, sync::Arc};
 
 use crate::shared::Shared;
 
+/// The receiving half of a Spillway channel.
 pub struct Receiver<T> {
     cursor: usize,
     buffer: VecDeque<T>,
@@ -26,6 +27,20 @@ impl<T> Receiver<T> {
         }
     }
 
+    /// This pattern does check whether the channel is closed. It's useful for examples and
+    /// some kinds of synchronous code, but use `next` or `poll_next` with async code.
+    pub fn try_next(&mut self) -> Option<T> {
+        match self.poll_next(&mut std::task::Context::from_waker(std::task::Waker::noop())) {
+            std::task::Poll::Ready(next) => next,
+            std::task::Poll::Pending => None,
+        }
+    }
+
+    /// The raw next value for the Receiver.
+    ///
+    /// * `Poll::Pending` when caught up and waiting for new messages.
+    /// * `Poll::Ready(Some(value))` for the next value.
+    /// * `Poll::Ready(None)` when all senders have been dropped and the Receiver is caught up. The Receiver will never receive more messages and you should drop it.
     pub fn poll_next(&mut self, context: &mut std::task::Context) -> std::task::Poll<Option<T>> {
         match self.buffer.pop_front() {
             Some(next) => std::task::Poll::Ready(Some(next)),
@@ -79,6 +94,10 @@ impl<T> Receiver<T> {
         }
     }
 
+    /// The next value for the Receiver.
+    ///
+    /// * Some(T) is the next value.
+    /// * None when all senders have been dropped and the Receiver is caught up. The Receiver will never receive more messages and you should drop it.
     #[inline]
     pub async fn next(&mut self) -> Option<T> {
         std::future::poll_fn(|context| self.poll_next(context)).await

spillway/src/receiver.rs

@@ -2,6 +2,9 @@ use std::sync::Arc;
 
 use crate::shared::Shared;
 
+/// The sending half of a Spillway channel.
+///
+/// You `clone()` to create more Senders.
 pub struct Sender<T> {
     chute: usize,
     shared: Arc<Shared<T>>,
@@ -43,6 +46,16 @@ impl<T> Sender<T> {
         }
     }
 
+    /// Send a value to the Receiver.
+    ///
+    /// Messages are only guaranteed to arrive in order on a per-Sender basis.
+    ///
+    /// If you send 1, 2, 3 in this Sender, you will receive 1, 2, 3 in that order.
+    /// If you send 4, 5, 6 from another Sender, you will receive 4, 5, 6 in that order too.
+    ///
+    /// However, you might receive 1, 4, 5, 2, 3, 6 or any other interleaving. But
+    /// 1 will always appear before 2, and 2 before 3; and 4 will always appear before 5,
+    /// and 5 before 6.
     pub fn send(&self, value: T) -> Result<(), T> {
         self.shared.send(self.chute, value)
     }