feat(mpsc): initial sync and async channel APIs (#2)

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

Cargo.toml

@@ -13,4 +13,12 @@ default = ["std"]
 [dependencies]
 
 [dev-dependencies]
-loom = "0.5"
+loom = { version = "0.5", features = ["checkpoint", "futures"] }
+# So that we can use `poll_fn` in tests.
+futures-util = "0.3"
+
+[profile.test]
+opt-level = 3
+
+[patch.crates-io]
+loom = { git = "https://github.com/tokio-rs/loom", branch = "master"}

src/lib.rs

@@ -1,48 +1,25 @@
 #![cfg_attr(not(feature = "std"), no_std)]
-
+#![cfg_attr(docsrs, feature(doc_cfg))]
 use core::{fmt, mem::MaybeUninit, ops::Index};
 
-#[cfg(feature = "alloc")]
-extern crate alloc;
-
-macro_rules! test_println {
-    ($($arg:tt)*) => {
-        if cfg!(test) {
-            if crate::util::panicking() {
-                // getting the thread ID while panicking doesn't seem to play super nicely with loom's
-                // mock lazy_static...
-                println!("[PANIC {:>17}:{:<3}] {}", file!(), line!(), format_args!($($arg)*))
-            } else {
-                println!("[{:?} {:>17}:{:<3}] {}", crate::loom::thread::current().id(), file!(), line!(), format_args!($($arg)*))
-            }
-        }
-    }
-}
-
-macro_rules! test_dbg {
-    ($e:expr) => {
-        match $e {
-            e => {
-                #[cfg(test)]
-                test_println!("{} = {:?}", stringify!($e), &e);
-                e
-            }
-        }
-    };
-}
+#[macro_use]
+mod macros;
 
 mod loom;
 mod util;
 
-#[cfg(feature = "alloc")]
-mod thingbuf;
-#[cfg(feature = "alloc")]
-pub use self::thingbuf::ThingBuf;
-#[cfg(feature = "alloc")]
-mod stringbuf;
+feature! {
+    #![feature = "alloc"]
+    extern crate alloc;
 
-#[cfg(feature = "alloc")]
-pub use stringbuf::{StaticStringBuf, StringBuf};
+    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;
@@ -55,6 +32,14 @@ use crate::{
     util::{Backoff, CachePadded},
 };
 
+pub struct Ref<'slot, T> {
+    slot: &'slot Slot<T>,
+    new_state: usize,
+}
+
+#[derive(Debug)]
+pub struct AtCapacity(pub(crate) usize);
+
 #[derive(Debug)]
 struct Core {
     head: CachePadded<AtomicUsize>,
@@ -65,15 +50,7 @@ struct Core {
     capacity: usize,
 }
 
-pub struct Ref<'slot, T> {
-    slot: &'slot Slot<T>,
-    new_state: usize,
-}
-
-#[derive(Debug)]
-pub struct AtCapacity(usize);
-
-pub struct Slot<T> {
+struct Slot<T> {
     value: UnsafeCell<MaybeUninit<T>>,
     state: AtomicUsize,
 }

src/loom.rs

@@ -6,7 +6,7 @@ mod inner {
         pub use loom::sync::atomic::*;
         pub use std::sync::atomic::Ordering;
     }
-    pub(crate) use loom::{cell::UnsafeCell, hint, thread};
+    pub(crate) use loom::{cell::UnsafeCell, future, hint, sync, thread};
 
     pub(crate) fn model(f: impl Fn() + Sync + Send + 'static) {
         let iteration = core::sync::atomic::AtomicUsize::new(0);
@@ -72,6 +72,13 @@ mod inner {
 #[cfg(not(test))]
 mod inner {
     #![allow(dead_code)]
+    pub(crate) mod sync {
+        pub use core::sync::*;
+
+        #[cfg(feature = "alloc")]
+        pub use alloc::sync::*;
+    }
+
     pub(crate) use core::sync::atomic;
 
     #[cfg(feature = "std")]

src/macros.rs

@@ -0,0 +1,67 @@
+macro_rules! test_println {
+    ($($arg:tt)*) => {
+        if cfg!(test) {
+            if crate::util::panic::panicking() {
+                // getting the thread ID while panicking doesn't seem to play super nicely with loom's
+                // mock lazy_static...
+                println!("[PANIC {:>17}:{:<3}] {}", file!(), line!(), format_args!($($arg)*))
+            } else {
+                println!("[{:?} {:>17}:{:<3}] {}", crate::loom::thread::current().id(), file!(), line!(), format_args!($($arg)*))
+            }
+        }
+    }
+}
+
+macro_rules! test_dbg {
+    ($e:expr) => {
+        match $e {
+            e => {
+                #[cfg(test)]
+                test_println!("{} = {:?}", stringify!($e), &e);
+                e
+            }
+        }
+    };
+}
+
+macro_rules! feature {
+    (
+        #![$meta:meta]
+        $($item:item)*
+    ) => {
+        $(
+            #[cfg($meta)]
+            #[cfg_attr(docsrs, doc(cfg($meta)))]
+            $item
+        )*
+    }
+}
+
+#[allow(unused_macros)]
+macro_rules! unreachable_unchecked {
+    ($($arg:tt)+) => {
+        crate::unreachable_unchecked!(@inner , format_args!(": {}", format_args!($($arg)*)))
+    };
+
+    () => {
+        crate::unreachable_unchecked!(@inner ".")
+    };
+
+    (@inner $msg:expr) => {
+        #[cfg(debug_assertions)] {
+            panic!(
+                "internal error: entered unreachable code{}\n\n\
+                /!\\ EXTREMELY SERIOUS WARNING /!\\\n
+                This code should NEVER be entered; in release mode, this would \
+                have been an `unreachable_unchecked` hint. The fact that this \
+                occurred means something VERY bad is going on. \n\
+                Please contact the `thingbuf` maintainers immediately. Sorry!",
+                $msg,
+            );
+        }
+        #[cfg(not(debug_assertions))]
+        unsafe {
+            core::hint::unreachable_unchecked();
+        }
+    };
+}

src/mpsc.rs

@@ -0,0 +1,31 @@
+//! Multi-producer, single-consumer channels using [`ThingBuf`](crate::ThingBuf).
+//!
+//! The default MPSC channel returned by the [`channel`] function is
+//! _asynchronous_: receiving from the channel is an `async fn`, and the
+//! receiving task willwait when there are no messages in the channel.
+//!
+//! If the "std" feature flag is enabled, this module also provides a
+//! synchronous channel, in the [`sync`] module. The synchronous  receiver will
+//! instead wait for new messages by blocking the current thread. Naturally,
+//! this requires the Rust standard library. A synchronous channel
+//! can be constructed using the [`sync::channel`] function.
+mod async_impl;
+pub use self::async_impl::*;
+
+feature! {
+    #![feature = "std"]
+    pub mod sync;
+}
+
+#[derive(Debug)]
+#[non_exhaustive]
+pub enum TrySendError {
+    AtCapacity(crate::AtCapacity),
+    Closed(Closed),
+}
+
+#[derive(Debug)]
+pub struct Closed(pub(crate) ());
+
+#[cfg(test)]
+mod tests;

src/mpsc/async_impl.rs

@@ -0,0 +1,280 @@
+use super::*;
+use crate::{
+    loom::{
+        self,
+        atomic::{AtomicUsize, Ordering},
+        sync::Arc,
+    },
+    util::wait::{WaitCell, WaitResult},
+    Ref, ThingBuf,
+};
+use core::{
+    fmt,
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll, Waker},
+};
+
+/// Returns a new synchronous multi-producer, single consumer channel.
+pub fn channel<T>(thingbuf: ThingBuf<T>) -> (Sender<T>, Receiver<T>) {
+    let inner = Arc::new(Inner {
+        thingbuf,
+        rx_wait: WaitCell::new(),
+        tx_count: AtomicUsize::new(1),
+    });
+    let tx = Sender {
+        inner: inner.clone(),
+    };
+    let rx = Receiver { inner };
+    (tx, rx)
+}
+
+pub struct Sender<T> {
+    inner: Arc<Inner<T>>,
+}
+
+pub struct Receiver<T> {
+    inner: Arc<Inner<T>>,
+}
+
+pub struct SendRef<'a, T> {
+    inner: &'a Inner<T>,
+    slot: Ref<'a, T>,
+}
+
+/// A [`Future`] that tries to receive a reference from a [`Receiver`].
+///
+/// This type is returned by [`Receiver::recv_ref`].
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct RecvRefFuture<'a, T> {
+    rx: &'a Receiver<T>,
+}
+
+/// A [`Future`] that tries to receive a value from a [`Receiver`].
+///
+/// This type is returned by [`Receiver::recv`].
+///
+/// This is equivalent to the [`RecvRefFuture`] future, but the value is moved out of
+/// the [`ThingBuf`] after it is received. This means that allocations are not
+/// reused.
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct RecvFuture<'a, T> {
+    rx: &'a Receiver<T>,
+}
+
+struct Inner<T> {
+    thingbuf: ThingBuf<T>,
+    rx_wait: WaitCell<Waker>,
+    tx_count: AtomicUsize,
+}
+
+// === impl Sender ===
+
+impl<T: Default> Sender<T> {
+    pub fn try_send_ref(&self) -> Result<SendRef<'_, T>, TrySendError> {
+        self.inner
+            .thingbuf
+            .push_ref()
+            .map(|slot| SendRef {
+                inner: &*self.inner,
+                slot,
+            })
+            .map_err(|e| {
+                if self.inner.rx_wait.is_rx_closed() {
+                    TrySendError::Closed(Closed(()))
+                } else {
+                    self.inner.rx_wait.notify();
+                    TrySendError::AtCapacity(e)
+                }
+            })
+    }
+
+    pub fn try_send(&self, val: T) -> Result<(), TrySendError> {
+        self.try_send_ref()?.with_mut(|slot| {
+            *slot = val;
+        });
+        Ok(())
+    }
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        test_dbg!(self.inner.tx_count.fetch_add(1, Ordering::Relaxed));
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        if test_dbg!(self.inner.tx_count.fetch_sub(1, Ordering::Release)) > 1 {
+            return;
+        }
+
+        // if we are the last sender, synchronize
+        test_dbg!(self.inner.tx_count.load(Ordering::SeqCst));
+        self.inner.rx_wait.close_tx();
+    }
+}
+
+// === impl Receiver ===
+
+impl<T: Default> Receiver<T> {
+    pub fn recv_ref(&self) -> RecvRefFuture<'_, T> {
+        RecvRefFuture { rx: self }
+    }
+
+    pub fn recv(&self) -> RecvFuture<'_, T> {
+        RecvFuture { rx: self }
+    }
+
+    /// # Returns
+    ///
+    ///  * `Poll::Pending` if no messages are available but the channel is not
+    ///    closed, or if a spurious failure happens.
+    ///  * `Poll::Ready(Some(Ref<T>))` if a message is available.
+    ///  * `Poll::Ready(None)` if the channel has been closed and all messages
+    ///    sent before it was closed have been received.
+    ///
+    /// When the method returns [`Poll::Pending`], the [`Waker`] in the provided
+    /// [`Context`] is scheduled to receive a wakeup when a message is sent on any
+    /// sender, or when the channel is closed.  Note that on multiple calls to
+    /// `poll_recv_ref`, only the [`Waker`] from the [`Context`] passed to the most
+    /// recent call is scheduled to receive a wakeup.
+    pub fn poll_recv_ref(&self, cx: &mut Context<'_>) -> Poll<Option<Ref<'_, T>>> {
+        loop {
+            if let Some(r) = self.try_recv_ref() {
+                return Poll::Ready(Some(r));
+            }
+
+            // Okay, no value is ready --- try to wait.
+            match test_dbg!(self.inner.rx_wait.wait_with(|| cx.waker().clone())) {
+                WaitResult::TxClosed => {
+                    // All senders have been dropped, but the channel might
+                    // still have messages in it. Return `Ready`; if the recv'd ref
+                    // is `None` then we've popped everything.
+                    return Poll::Ready(self.try_recv_ref());
+                }
+                WaitResult::Wait => {
+                    // make sure nobody sent a message while we were registering
+                    // the waiter...
+                    // XXX(eliza): a nicer solution _might_ just be to pack the
+                    // waiter state into the tail idx or something or something
+                    // but that kind of defeats the purpose of just having a
+                    // nice "wrap a queue into a channel" API...
+                    if let Some(val) = self.try_recv_ref() {
+                        return Poll::Ready(Some(val));
+                    }
+                    return Poll::Pending;
+                }
+                WaitResult::Notified => {
+                    loom::hint::spin_loop();
+                }
+            };
+        }
+    }
+
+    /// # Returns
+    ///
+    ///  * `Poll::Pending` if no messages are available but the channel is not
+    ///    closed, or if a spurious failure happens.
+    ///  * `Poll::Ready(Some(message))` if a message is available.
+    ///  * `Poll::Ready(None)` if the channel has been closed and all messages
+    ///    sent before it was closed have been received.
+    ///
+    /// When the method returns [`Poll::Pending`], the [`Waker`] in the provided
+    /// [`Context`] is scheduled to receive a wakeup when a message is sent on any
+    /// sender, or when the channel is closed.  Note that on multiple calls to
+    /// `poll_recv`, only the [`Waker`] from the [`Context`] passed to the most
+    /// recent call is scheduled to receive a wakeup.
+    pub fn poll_recv(&self, cx: &mut Context<'_>) -> Poll<Option<T>> {
+        self.poll_recv_ref(cx)
+            .map(|opt| opt.map(|mut r| r.with_mut(core::mem::take)))
+    }
+
+    fn try_recv_ref(&self) -> Option<Ref<'_, T>> {
+        self.inner.thingbuf.pop_ref()
+    }
+
+    pub fn is_closed(&self) -> bool {
+        test_dbg!(self.inner.tx_count.load(Ordering::SeqCst)) <= 1
+    }
+}
+
+impl<T> Drop for Receiver<T> {
+    fn drop(&mut self) {
+        self.inner.rx_wait.close_rx();
+    }
+}
+
+// === impl SendRef ===
+
+impl<T> SendRef<'_, T> {
+    #[inline]
+    pub fn with<U>(&self, f: impl FnOnce(&T) -> U) -> U {
+        self.slot.with(f)
+    }
+
+    #[inline]
+    pub fn with_mut<U>(&mut self, f: impl FnOnce(&mut T) -> U) -> U {
+        self.slot.with_mut(f)
+    }
+}
+
+impl<T> Drop for SendRef<'_, T> {
+    #[inline]
+    fn drop(&mut self) {
+        test_println!("drop async SendRef");
+        self.inner.rx_wait.notify();
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for SendRef<'_, T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.with(|val| fmt::Debug::fmt(val, f))
+    }
+}
+
+impl<T: fmt::Display> fmt::Display for SendRef<'_, T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.with(|val| fmt::Display::fmt(val, f))
+    }
+}
+
+impl<T: fmt::Write> fmt::Write for SendRef<'_, T> {
+    #[inline]
+    fn write_str(&mut self, s: &str) -> fmt::Result {
+        self.with_mut(|val| val.write_str(s))
+    }
+
+    #[inline]
+    fn write_char(&mut self, c: char) -> fmt::Result {
+        self.with_mut(|val| val.write_char(c))
+    }
+
+    #[inline]
+    fn write_fmt(&mut self, f: fmt::Arguments<'_>) -> fmt::Result {
+        self.with_mut(|val| val.write_fmt(f))
+    }
+}
+
+// === impl RecvRefFuture ===
+
+impl<'a, T: Default> Future for RecvRefFuture<'a, T> {
+    type Output = Option<Ref<'a, T>>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.rx.poll_recv_ref(cx)
+    }
+}
+
+// === impl Recv ===
+
+impl<'a, T: Default> Future for RecvFuture<'a, T> {
+    type Output = Option<T>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.rx.poll_recv(cx)
+    }
+}

src/mpsc/sync.rs

@@ -0,0 +1,216 @@
+//! A synchronous multi-producer, single-consumer channel.
+//!
+//! This provides an equivalent API to the [`mpsc`](crate::mpsc) module, but the
+//! [`Receiver`] type in this module waits by blocking the current thread,
+//! rather than asynchronously yielding.
+use super::{Closed, TrySendError};
+use crate::{
+    loom::{
+        self,
+        atomic::{AtomicUsize, Ordering},
+        sync::Arc,
+        thread::{self, Thread},
+    },
+    util::wait::{WaitCell, WaitResult},
+    Ref, ThingBuf,
+};
+use core::fmt;
+
+/// Returns a new asynchronous multi-producer, single consumer channel.
+pub fn channel<T>(thingbuf: ThingBuf<T>) -> (Sender<T>, Receiver<T>) {
+    let inner = Arc::new(Inner {
+        thingbuf,
+        rx_wait: WaitCell::new(),
+        tx_count: AtomicUsize::new(1),
+    });
+    let tx = Sender {
+        inner: inner.clone(),
+    };
+    let rx = Receiver { inner };
+    (tx, rx)
+}
+
+pub struct Sender<T> {
+    inner: Arc<Inner<T>>,
+}
+
+pub struct Receiver<T> {
+    inner: Arc<Inner<T>>,
+}
+
+pub struct SendRef<'a, T> {
+    inner: &'a Inner<T>,
+    slot: Ref<'a, T>,
+}
+
+struct Inner<T> {
+    thingbuf: ThingBuf<T>,
+    rx_wait: WaitCell<Thread>,
+    tx_count: AtomicUsize,
+}
+
+// === impl Sender ===
+
+impl<T: Default> Sender<T> {
+    pub fn try_send_ref(&self) -> Result<SendRef<'_, T>, TrySendError> {
+        self.inner
+            .thingbuf
+            .push_ref()
+            .map(|slot| SendRef {
+                inner: &*self.inner,
+                slot,
+            })
+            .map_err(|e| {
+                if self.inner.rx_wait.is_rx_closed() {
+                    TrySendError::Closed(Closed(()))
+                } else {
+                    self.inner.rx_wait.notify();
+                    TrySendError::AtCapacity(e)
+                }
+            })
+    }
+
+    pub fn try_send(&self, val: T) -> Result<(), TrySendError> {
+        self.try_send_ref()?.with_mut(|slot| {
+            *slot = val;
+        });
+        Ok(())
+    }
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        test_dbg!(self.inner.tx_count.fetch_add(1, Ordering::Relaxed));
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        if test_dbg!(self.inner.tx_count.fetch_sub(1, Ordering::Release)) > 1 {
+            return;
+        }
+
+        // if we are the last sender, synchronize
+        test_dbg!(self.inner.tx_count.load(Ordering::SeqCst));
+        self.inner.rx_wait.close_tx();
+    }
+}
+
+// === impl Receiver ===
+
+impl<T: Default> Receiver<T> {
+    pub fn recv_ref(&self) -> Option<Ref<'_, T>> {
+        loop {
+            // If we got a value, return it!
+            if let Some(r) = self.inner.thingbuf.pop_ref() {
+                return Some(r);
+            }
+
+            // otherwise, gotosleep
+            match test_dbg!(self.inner.rx_wait.wait_with(thread::current)) {
+                WaitResult::TxClosed => {
+                    // All senders have been dropped, but the channel might
+                    // still have messages in it...
+                    return self.inner.thingbuf.pop_ref();
+                }
+                WaitResult::Wait => {
+                    // make sure nobody sent a message while we were registering
+                    // the waiter...
+                    // XXX(eliza): a nicer solution _might_ just be to pack the
+                    // waiter state into the tail idx or something or something
+                    // but that kind of defeats the purpose of just having a
+                    // nice "wrap a queue into a channel" API...
+                    if let Some(val) = self.inner.thingbuf.pop_ref() {
+                        return Some(val);
+                    }
+                    test_println!("parking ({:?})", thread::current());
+                    thread::park();
+                }
+                WaitResult::Notified => {
+                    loom::hint::spin_loop();
+                }
+            }
+        }
+    }
+
+    pub fn try_recv_ref(&self) -> Option<Ref<'_, T>> {
+        self.inner.thingbuf.pop_ref()
+    }
+
+    pub fn recv(&self) -> Option<T> {
+        let val = self.recv_ref()?.with_mut(core::mem::take);
+        Some(val)
+    }
+
+    pub fn is_closed(&self) -> bool {
+        test_dbg!(self.inner.tx_count.load(Ordering::SeqCst)) <= 1
+    }
+}
+
+impl<'a, T: Default> Iterator for &'a Receiver<T> {
+    type Item = Ref<'a, T>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.recv_ref()
+    }
+}
+
+impl<T> Drop for Receiver<T> {
+    fn drop(&mut self) {
+        self.inner.rx_wait.close_rx();
+    }
+}
+
+// === impl SendRef ===
+
+impl<T> SendRef<'_, T> {
+    #[inline]
+    pub fn with<U>(&self, f: impl FnOnce(&T) -> U) -> U {
+        self.slot.with(f)
+    }
+
+    #[inline]
+    pub fn with_mut<U>(&mut self, f: impl FnOnce(&mut T) -> U) -> U {
+        self.slot.with_mut(f)
+    }
+}
+
+impl<T> Drop for SendRef<'_, T> {
+    #[inline]
+    fn drop(&mut self) {
+        test_println!("drop SendRef");
+        self.inner.rx_wait.notify();
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for SendRef<'_, T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.with(|val| fmt::Debug::fmt(val, f))
+    }
+}
+
+impl<T: fmt::Display> fmt::Display for SendRef<'_, T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.with(|val| fmt::Display::fmt(val, f))
+    }
+}
+
+impl<T: fmt::Write> fmt::Write for SendRef<'_, T> {
+    #[inline]
+    fn write_str(&mut self, s: &str) -> fmt::Result {
+        self.with_mut(|val| val.write_str(s))
+    }
+
+    #[inline]
+    fn write_char(&mut self, c: char) -> fmt::Result {
+        self.with_mut(|val| val.write_char(c))
+    }
+
+    #[inline]
+    fn write_fmt(&mut self, f: fmt::Arguments<'_>) -> fmt::Result {
+        self.with_mut(|val| val.write_fmt(f))
+    }
+}

src/mpsc/tests.rs

@@ -0,0 +1,5 @@
+use super::*;
+
+mod mpsc_async;
+#[cfg(feature = "std")]
+mod mpsc_sync;

src/mpsc/tests/mpsc_async.rs

@@ -0,0 +1,123 @@
+use super::*;
+use crate::{
+    loom::{self, future, thread},
+    ThingBuf,
+};
+
+#[test]
+fn basically_works() {
+    loom::model(|| {
+        let (tx, rx) = channel(ThingBuf::new(4));
+
+        let p1 = {
+            let tx = tx.clone();
+            thread::spawn(move || {
+                tx.try_send_ref().unwrap().with_mut(|val| *val = 1);
+                tx.try_send_ref().unwrap().with_mut(|val| *val = 2);
+            })
+        };
+        let p2 = thread::spawn(move || {
+            tx.try_send(3).unwrap();
+            tx.try_send(4).unwrap();
+        });
+
+        let mut vals = future::block_on(async move {
+            let mut vals = Vec::new();
+            while let Some(val) = rx.recv_ref().await {
+                val.with(|val| vals.push(*val));
+            }
+            vals
+        });
+
+        vals.sort_unstable();
+        assert_eq!(vals, vec![1, 2, 3, 4]);
+
+        p1.join().unwrap();
+        p2.join().unwrap();
+    })
+}
+
+#[test]
+fn rx_closes() {
+    const ITERATIONS: usize = 6;
+    loom::model(|| {
+        let (tx, rx) = channel(ThingBuf::new(ITERATIONS / 2));
+
+        let producer = thread::spawn(move || {
+            'iters: for i in 0..=ITERATIONS {
+                test_println!("sending {}...", i);
+                'send: loop {
+                    match tx.try_send(i) {
+                        Ok(_) => break 'send,
+                        Err(TrySendError::AtCapacity(_)) => thread::yield_now(),
+                        Err(TrySendError::Closed(_)) => break 'iters,
+                    }
+                }
+                test_println!("sent {}\n", i);
+            }
+        });
+
+        future::block_on(async move {
+            for i in 0..ITERATIONS - 1 {
+                test_println!("receiving {}...", i);
+                let n = rx.recv().await;
+
+                test_println!("recv {:?}\n", n);
+                assert_eq!(n, Some(i));
+            }
+        });
+
+        producer.join().unwrap();
+    })
+}
+
+#[test]
+fn spsc_recv_then_send() {
+    loom::model(|| {
+        let (tx, rx) = channel(ThingBuf::<i32>::new(4));
+        let consumer = thread::spawn(move || {
+            future::block_on(async move {
+                assert_eq!(rx.recv().await.unwrap(), 10);
+            })
+        });
+
+        tx.try_send(10).unwrap();
+        consumer.join().unwrap();
+    })
+}
+
+#[test]
+fn spsc_recv_then_close() {
+    loom::model(|| {
+        let (tx, rx) = channel(ThingBuf::<i32>::new(4));
+        let producer = thread::spawn(move || {
+            drop(tx);
+        });
+
+        future::block_on(async move {
+            let recv = rx.recv().await;
+            assert_eq!(recv, None);
+        });
+
+        producer.join().unwrap();
+    });
+}
+
+#[test]
+fn spsc_recv_then_send_then_close() {
+    loom::model(|| {
+        let (tx, rx) = channel(ThingBuf::<i32>::new(2));
+        let consumer = thread::spawn(move || {
+            future::block_on(async move {
+                assert_eq!(rx.recv().await.unwrap(), 10);
+                assert_eq!(rx.recv().await.unwrap(), 20);
+                assert_eq!(rx.recv().await, None);
+            })
+        });
+
+        tx.try_send(10).unwrap();
+        tx.try_send(20).unwrap();
+        drop(tx);
+        consumer.join().unwrap();
+    })
+}

src/mpsc/tests/mpsc_sync.rs

@@ -0,0 +1,111 @@
+use super::*;
+use crate::{
+    loom::{self, thread},
+    ThingBuf,
+};
+
+#[test]
+fn basically_works() {
+    loom::model(|| {
+        let (tx, rx) = sync::channel(ThingBuf::new(4));
+
+        let p1 = {
+            let tx = tx.clone();
+            thread::spawn(move || {
+                tx.try_send_ref().unwrap().with_mut(|val| *val = 1);
+                tx.try_send_ref().unwrap().with_mut(|val| *val = 2);
+            })
+        };
+        let p2 = thread::spawn(move || {
+            tx.try_send(3).unwrap();
+            tx.try_send(4).unwrap();
+        });
+
+        let mut vals = Vec::new();
+
+        for val in &rx {
+            val.with(|val| vals.push(*val));
+        }
+
+        vals.sort_unstable();
+        assert_eq!(vals, vec![1, 2, 3, 4]);
+
+        p1.join().unwrap();
+        p2.join().unwrap();
+    })
+}
+
+#[test]
+fn rx_closes() {
+    const ITERATIONS: usize = 6;
+    loom::model(|| {
+        let (tx, rx) = sync::channel(ThingBuf::new(ITERATIONS / 2));
+
+        let producer = thread::spawn(move || {
+            'iters: for i in 0..=ITERATIONS {
+                'send: loop {
+                    match tx.try_send(i) {
+                        Ok(_) => break 'send,
+                        Err(TrySendError::AtCapacity(_)) => thread::yield_now(),
+                        Err(TrySendError::Closed(_)) => break 'iters,
+                    }
+                }
+                test_println!("sent {}\n", i);
+            }
+        });
+
+        for i in 0..ITERATIONS - 1 {
+            let n = rx.recv();
+
+            test_println!("recv {:?}\n", n);
+            assert_eq!(n, Some(i));
+        }
+        drop(rx);
+
+        producer.join().unwrap();
+    })
+}
+
+#[test]
+fn spsc_recv_then_send() {
+    loom::model(|| {
+        let (tx, rx) = sync::channel(ThingBuf::<i32>::new(4));
+        let consumer = thread::spawn(move || {
+            assert_eq!(rx.recv().unwrap(), 10);
+        });
+
+        tx.try_send(10).unwrap();
+        consumer.join().unwrap();
+    })
+}
+
+#[test]
+fn spsc_recv_then_close() {
+    loom::model(|| {
+        let (tx, rx) = sync::channel(ThingBuf::<i32>::new(4));
+        let producer = thread::spawn(move || {
+            drop(tx);
+        });
+
+        assert_eq!(rx.recv(), None);
+
+        producer.join().unwrap();
+    });
+}
+
+#[test]
+fn spsc_recv_then_send_then_close() {
+    loom::model(|| {
+        let (tx, rx) = sync::channel(ThingBuf::<i32>::new(2));
+        let consumer = thread::spawn(move || {
+            assert_eq!(rx.recv().unwrap(), 10);
+            assert_eq!(rx.recv().unwrap(), 20);
+            assert_eq!(rx.recv(), None);
+        });
+
+        tx.try_send(10).unwrap();
+        tx.try_send(20).unwrap();
+        drop(tx);
+        consumer.join().unwrap();
+    })
+}

src/sync_channel.rs

@@ -1,20 +0,0 @@
-use super::*;
-use crate::loom::sync::{Condvar, Mutex};
-
-pub struct Sender<T> {
-    inner: Arc<Inner<T>>,
-}
-
-pub struct Receiver<T> {
-    inner: Arc<Inner<T>>,
-}
-
-struct Inner<T> {
-    lock: Mutex<bool>,
-    cv: Condvar,
-    buf: ThingBuf<T>,
-}
-
-impl<T> Receiver<T> {
-    
-}

src/sync_mpsc.rs

@@ -0,0 +1,211 @@
+use super::*;
+use crate::{
+    error::TrySendError,
+    loom::{
+        atomic::{AtomicUsize, Ordering},
+        sync::Arc,
+        thread::{self, Thread},
+    },
+    util::wait::{WaitCell, WaitResult},
+};
+
+#[cfg(test)]
+mod tests;
+
+pub fn channel<T>(thingbuf: ThingBuf<T>) -> (Sender<T>, Receiver<T>) {
+    let inner = Arc::new(Inner {
+        thingbuf,
+        rx_wait: WaitCell::new(),
+        tx_count: AtomicUsize::new(1),
+    });
+    let tx = Sender {
+        inner: inner.clone(),
+    };
+    let rx = Receiver { inner };
+    (tx, rx)
+}
+
+pub struct Sender<T> {
+    inner: Arc<Inner<T>>,
+}
+
+pub struct Receiver<T> {
+    inner: Arc<Inner<T>>,
+}
+
+pub struct SendRef<'a, T> {
+    inner: &'a Inner<T>,
+    slot: Ref<'a, T>,
+}
+
+struct Inner<T> {
+    thingbuf: ThingBuf<T>,
+    rx_wait: WaitCell<Thread>,
+    tx_count: AtomicUsize,
+}
+
+// === impl Sender ===
+
+impl<T: Default> Sender<T> {
+    pub fn try_send_ref(&self) -> Result<SendRef<'_, T>, TrySendError> {
+        self.inner
+            .thingbuf
+            .push_ref()
+            .map(|slot| SendRef {
+                inner: &*self.inner,
+                slot,
+            })
+            .map_err(|e| {
+                if self.inner.rx_wait.is_rx_closed() {
+                    TrySendError::Closed(error::Closed(()))
+                } else {
+                    self.inner.rx_wait.notify();
+                    TrySendError::AtCapacity(e)
+                }
+            })
+    }
+
+    pub fn try_send(&self, val: T) -> Result<(), TrySendError> {
+        self.try_send_ref()?.with_mut(|slot| {
+            *slot = val;
+        });
+        Ok(())
+    }
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        test_dbg!(self.inner.tx_count.fetch_add(1, Ordering::Relaxed));
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        if test_dbg!(self.inner.tx_count.fetch_sub(1, Ordering::Release)) > 1 {
+            return;
+        }
+
+        // if we are the last sender, synchronize
+        test_dbg!(self.inner.tx_count.load(Ordering::SeqCst));
+        self.inner.rx_wait.close_tx();
+    }
+}
+
+// === impl Receiver ===
+
+impl<T: Default> Receiver<T> {
+    pub fn recv_ref(&self) -> Option<Ref<'_, T>> {
+        loop {
+            // If we got a value, return it!
+            if let Some(r) = self.inner.thingbuf.pop_ref() {
+                return Some(r);
+            }
+
+            // otherwise, gotosleep
+            match test_dbg!(self.inner.rx_wait.wait_with(thread::current)) {
+                WaitResult::TxClosed => {
+                    // All senders have been dropped, but the channel might
+                    // still have messages in it...
+                    return self.inner.thingbuf.pop_ref();
+                }
+                WaitResult::Wait => {
+                    // make sure nobody sent a message while we were registering
+                    // the waiter...
+                    // XXX(eliza): a nicer solution _might_ just be to pack the
+                    // waiter state into the tail idx or something or something
+                    // but that kind of defeats the purpose of just having a
+                    // nice "wrap a queue into a channel" API...
+                    if let Some(val) = self.inner.thingbuf.pop_ref() {
+                        return Some(val);
+                    }
+                    test_println!("parking ({:?})", thread::current());
+                    thread::park();
+                }
+                WaitResult::Notified => {
+                    loom::hint::spin_loop();
+                }
+            }
+        }
+    }
+
+    pub fn try_recv_ref(&self) -> Option<Ref<'_, T>> {
+        self.inner.thingbuf.pop_ref()
+    }
+
+    pub fn recv(&self) -> Option<T> {
+        let val = self.recv_ref()?.with_mut(core::mem::take);
+        Some(val)
+    }
+
+    pub fn is_closed(&self) -> bool {
+        test_dbg!(self.inner.tx_count.load(Ordering::SeqCst)) <= 1
+    }
+}
+
+impl<'a, T: Default> Iterator for &'a Receiver<T> {
+    type Item = Ref<'a, T>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.recv_ref()
+    }
+}
+
+impl<T> Drop for Receiver<T> {
+    fn drop(&mut self) {
+        self.inner.rx_wait.close_rx();
+    }
+}
+
+// === impl SendRef ===
+
+impl<T> SendRef<'_, T> {
+    #[inline]
+    pub fn with<U>(&self, f: impl FnOnce(&T) -> U) -> U {
+        self.slot.with(f)
+    }
+
+    #[inline]
+    pub fn with_mut<U>(&mut self, f: impl FnOnce(&mut T) -> U) -> U {
+        self.slot.with_mut(f)
+    }
+}
+
+impl<T> Drop for SendRef<'_, T> {
+    #[inline]
+    fn drop(&mut self) {
+        test_println!("drop SendRef");
+        self.inner.rx_wait.notify();
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for SendRef<'_, T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.with(|val| fmt::Debug::fmt(val, f))
+    }
+}
+
+impl<T: fmt::Display> fmt::Display for SendRef<'_, T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.with(|val| fmt::Display::fmt(val, f))
+    }
+}
+
+impl<T: fmt::Write> fmt::Write for SendRef<'_, T> {
+    #[inline]
+    fn write_str(&mut self, s: &str) -> fmt::Result {
+        self.with_mut(|val| val.write_str(s))
+    }
+
+    #[inline]
+    fn write_char(&mut self, c: char) -> fmt::Result {
+        self.with_mut(|val| val.write_char(c))
+    }
+
+    #[inline]
+    fn write_fmt(&mut self, f: fmt::Arguments<'_>) -> fmt::Result {
+        self.with_mut(|val| val.write_fmt(f))
+    }
+}

src/thingbuf.rs

@@ -6,7 +6,7 @@ use core::{fmt, ptr};
 #[cfg(test)]
 mod tests;
 
-pub struct ThingBuf<T: Default> {
+pub struct ThingBuf<T> {
     core: Core,
     slots: Box<[Slot<T>]>,
 }
@@ -40,7 +40,9 @@ impl<T: Default> ThingBuf<T> {
     pub fn pop_with<U>(&self, f: impl FnOnce(&mut T) -> U) -> Option<U> {
         self.pop_ref().map(|mut r| r.with_mut(f))
     }
+}
 
+impl<T> ThingBuf<T> {
     #[inline]
     pub fn capacity(&self) -> usize {
         self.slots.len()
@@ -57,7 +59,7 @@ impl<T: Default> ThingBuf<T> {
     }
 }
 
-impl<T: Default> Drop for ThingBuf<T> {
+impl<T> Drop for ThingBuf<T> {
     fn drop(&mut self) {
         let tail = self.core.tail.load(Ordering::SeqCst);
         let (idx, gen) = self.core.idx_gen(tail);
@@ -71,7 +73,7 @@ impl<T: Default> Drop for ThingBuf<T> {
     }
 }
 
-impl<T: Default> fmt::Debug for ThingBuf<T> {
+impl<T> fmt::Debug for ThingBuf<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ThingBuf")
             .field("capacity", &self.capacity())

src/thingbuf/tests.rs

@@ -54,6 +54,41 @@ fn push_many_mpsc() {
     })
 }
 
+#[test]
+fn spsc() {
+    const COUNT: usize = 9;
+    loom::model(|| {
+        let q = Arc::new(ThingBuf::new(3));
+
+        let producer = {
+            let q = q.clone();
+            thread::spawn(move || {
+                for i in 0..COUNT {
+                    loop {
+                        if let Ok(mut guard) = q.push_ref() {
+                            guard.with_mut(|val| *val = i);
+                            break;
+                        }
+                        thread::yield_now();
+                    }
+                }
+            })
+        };
+
+        for i in 0..COUNT {
+            loop {
+                if let Some(guard) = q.pop_ref() {
+                    guard.with(|val| assert_eq!(*val, i));
+                    break;
+                }
+                thread::yield_now();
+            }
+        }
+
+        producer.join().unwrap();
+    });
+}
+
 #[test]
 #[ignore] // this takes about a million years to run
 fn linearizable() {

src/util.rs

@@ -1,6 +1,9 @@
 use crate::loom;
 use core::ops::{Deref, DerefMut};
 
+pub(crate) mod panic;
+pub(crate) mod wait;
+
 #[derive(Debug)]
 pub(crate) struct Backoff(u8);
 
@@ -12,16 +15,6 @@ pub(crate) struct Backoff(u8);
 #[derive(Clone, Copy, Default, Hash, PartialEq, Eq, Debug)]
 pub(crate) struct CachePadded<T>(pub(crate) T);
 
-#[cfg(feature = "std")]
-pub(crate) fn panicking() -> bool {
-    std::thread::panicking()
-}
-
-#[cfg(not(feature = "std"))]
-pub(crate) fn panicking() -> bool {
-    false
-}
-
 // === impl Backoff ===
 
 impl Backoff {
@@ -36,8 +29,6 @@ impl Backoff {
     pub(crate) fn spin(&mut self) {
         for _ in 0..test_dbg!(1 << self.0.min(Self::MAX_SPINS)) {
             loom::hint::spin_loop();
-
-            test_println!("spin_loop_hint");
         }
 
         if self.0 <= Self::MAX_SPINS {
@@ -50,7 +41,6 @@ impl Backoff {
         if self.0 <= Self::MAX_SPINS || cfg!(not(any(feature = "std", test))) {
             for _ in 0..1 << self.0 {
                 loom::hint::spin_loop();
-                test_println!("spin_loop_hint");
             }
         }
 

src/util/panic.rs

@@ -0,0 +1,27 @@
+pub(crate) use self::inner::*;
+pub(crate) use core::panic::*;
+
+#[cfg(feature = "std")]
+mod inner {
+    pub(crate) fn panicking() -> bool {
+        std::thread::panicking()
+    }
+
+    pub use std::panic::{catch_unwind, resume_unwind};
+}
+
+#[cfg(not(feature = "std"))]
+mod inner {
+    use super::*;
+    pub(crate) fn panicking() -> bool {
+        false
+    }
+
+    pub(crate) fn catch_unwind<F: FnOnce() -> R + UnwindSafe, R>(f: F) -> Result<R, ()> {
+        Ok(f())
+    }
+
+    pub(crate) fn resume_unwind(payload: ()) -> ! {
+        unreachable_unchecked!("code compiled with no_std cannot unwind!")
+    }
+}

src/util/wait.rs

@@ -0,0 +1,332 @@
+use crate::{
+    loom::{
+        atomic::{AtomicUsize, Ordering::*},
+        UnsafeCell,
+    },
+    util::panic::{self, RefUnwindSafe, UnwindSafe},
+};
+use core::{fmt, task::Waker};
+
+#[cfg(feature = "std")]
+use crate::loom::thread;
+
+/// An atomically registered waiter ([`Waker`] or [`Thread`]).
+///
+/// This is inspired by the [`AtomicWaker` type] used in Tokio's
+/// synchronization primitives, with the following modifications:
+///
+/// - Unlike [`AtomicWaker`], a `WaitCell` is generic over the type of the
+///   waiting value. This means it can be used in both asynchronous code (with
+///   [`Waker`]s), or in synchronous, multi-threaded code (with a [`Thread`]).
+/// - An additional bit of state is added to allow setting a "close" bit. This
+///   is so that closing a channel can be tracked in the same atomic as the
+///   receiver's notification state, reducing the number of separate atomic RMW
+///   ops that have to be synchronized between.
+/// - A `WaitCell` is always woken by value. This is just because I didn't
+///   actually need separate "take waiter" and "wake" steps for any of the uses
+///   in `ThingBuf`...
+///
+/// [`AtomicWaker`]: https://github.com/tokio-rs/tokio/blob/09b770c5db31a1f35631600e1d239679354da2dd/tokio/src/sync/task/atomic_waker.rs
+pub(crate) struct WaitCell<T> {
+    lock: AtomicUsize,
+    waiter: UnsafeCell<Option<T>>,
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub(crate) enum WaitResult {
+    Wait,
+    Notified,
+    TxClosed,
+}
+
+pub(crate) trait Notify {
+    fn notify(self);
+}
+
+// === impl WaitCell ===
+
+impl<T: Notify + UnwindSafe + fmt::Debug> WaitCell<T> {
+    const WAITING: usize = 0b00;
+    const PARKING: usize = 0b01;
+    const NOTIFYING: usize = 0b10;
+    const TX_CLOSED: usize = 0b100;
+    const RX_CLOSED: usize = 0b1000;
+
+    pub(crate) fn new() -> Self {
+        Self {
+            lock: AtomicUsize::new(Self::WAITING),
+            waiter: UnsafeCell::new(None),
+        }
+    }
+
+    pub(crate) fn close_rx(&self) {
+        self.lock.fetch_or(Self::RX_CLOSED, AcqRel);
+    }
+
+    pub(crate) fn is_rx_closed(&self) -> bool {
+        test_dbg!(self.lock.load(Acquire) & Self::RX_CLOSED == Self::RX_CLOSED)
+    }
+
+    pub(crate) fn wait_with(&self, f: impl FnOnce() -> T) -> WaitResult {
+        test_println!("registering waiter");
+
+        // this is based on tokio's AtomicWaker synchronization strategy
+        match test_dbg!(self
+            .lock
+            .compare_exchange(Self::WAITING, Self::PARKING, AcqRel, Acquire,))
+        {
+            // someone else is notifying the receiver, so don't park!
+            Err(actual) if test_dbg!(actual & Self::TX_CLOSED) == Self::TX_CLOSED => {
+                test_println!("-> state = TX_CLOSED");
+                return WaitResult::TxClosed;
+            }
+            Err(actual) if test_dbg!(actual & Self::NOTIFYING) == Self::NOTIFYING => {
+                test_println!("-> state = NOTIFYING");
+                // f().notify();
+                // loom::hint::spin_loop();
+                return WaitResult::Notified;
+            }
+
+            Err(actual) => {
+                debug_assert!(actual == Self::PARKING || actual == Self::PARKING | Self::NOTIFYING);
+                return WaitResult::Wait;
+            }
+            Ok(_) => {}
+        }
+
+        test_println!("-> locked!");
+        let (panicked, prev_waiter) = match panic::catch_unwind(panic::AssertUnwindSafe(f)) {
+            Ok(new_waiter) => {
+                let new_waiter = test_dbg!(new_waiter);
+                let prev_waiter = self
+                    .waiter
+                    .with_mut(|waiter| unsafe { (*waiter).replace(new_waiter) });
+                (None, test_dbg!(prev_waiter))
+            }
+            Err(panic) => (Some(panic), None),
+        };
+
+        let result = match test_dbg!(self.lock.compare_exchange(
+            Self::PARKING,
+            Self::WAITING,
+            AcqRel,
+            Acquire
+        )) {
+            Ok(_) => {
+                let _ = panic::catch_unwind(move || drop(prev_waiter));
+
+                WaitResult::Wait
+            }
+            Err(actual) => {
+                test_println!("-> was notified; state={:#b}", actual);
+                let waiter = self.waiter.with_mut(|waiter| unsafe { (*waiter).take() });
+                // Reset to the WAITING state by clearing everything *except*
+                // the closed bits (which must remain set).
+                let state = test_dbg!(self
+                    .lock
+                    .fetch_and(Self::TX_CLOSED | Self::RX_CLOSED, AcqRel));
+                // The only valid state transition while we were parking is to
+                // add the TX_CLOSED bit.
+                debug_assert!(
+                    state == actual || state == actual | Self::TX_CLOSED,
+                    "state changed unexpectedly while parking!"
+                );
+
+                if let Some(prev_waiter) = prev_waiter {
+                    let _ = panic::catch_unwind(move || {
+                        prev_waiter.notify();
+                    });
+                }
+
+                if let Some(waiter) = waiter {
+                    debug_assert!(panicked.is_none());
+                    waiter.notify();
+                }
+
+                if state & Self::TX_CLOSED == Self::TX_CLOSED {
+                    WaitResult::TxClosed
+                } else {
+                    WaitResult::Notified
+                }
+            }
+        };
+
+        if let Some(panic) = panicked {
+            panic::resume_unwind(panic);
+        }
+
+        result
+    }
+
+    pub(crate) fn notify(&self) {
+        self.notify2(false)
+    }
+
+    pub(crate) fn close_tx(&self) {
+        self.notify2(true)
+    }
+
+    fn notify2(&self, close: bool) {
+        test_println!("notifying; close={:?};", close);
+        let bits = if close {
+            Self::NOTIFYING | Self::TX_CLOSED
+        } else {
+            Self::NOTIFYING
+        };
+        if test_dbg!(self.lock.fetch_or(bits, AcqRel)) == Self::WAITING {
+            // we have the lock!
+            let waiter = self.waiter.with_mut(|thread| unsafe { (*thread).take() });
+
+            self.lock.fetch_and(!Self::NOTIFYING, Release);
+
+            if let Some(waiter) = test_dbg!(waiter) {
+                waiter.notify();
+            }
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl Notify for thread::Thread {
+    fn notify(self) {
+        test_println!("NOTIFYING {:?} (from {:?})", self, thread::current());
+        self.unpark();
+    }
+}
+
+impl Notify for Waker {
+    fn notify(self) {
+        test_println!("WAKING TASK {:?} (from {:?})", self, thread::current());
+        self.wake();
+    }
+}
+
+impl<T: UnwindSafe> UnwindSafe for WaitCell<T> {}
+impl<T: RefUnwindSafe> RefUnwindSafe for WaitCell<T> {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::loom::{
+        self, future,
+        sync::atomic::{AtomicUsize, Ordering::Relaxed},
+        thread,
+    };
+    #[cfg(feature = "alloc")]
+    use alloc::sync::Arc;
+    use core::task::{Poll, Waker};
+
+    struct Chan {
+        num: AtomicUsize,
+        task: WaitCell<Waker>,
+    }
+
+    const NUM_NOTIFY: usize = 2;
+
+    async fn wait_on(chan: Arc<Chan>) {
+        futures_util::future::poll_fn(move |cx| {
+            let res = test_dbg!(chan.task.wait_with(|| cx.waker().clone()));
+
+            if NUM_NOTIFY == chan.num.load(Relaxed) {
+                return Poll::Ready(());
+            }
+
+            if res == WaitResult::Notified || res == WaitResult::TxClosed {
+                return Poll::Ready(());
+            }
+
+            Poll::Pending
+        })
+        .await
+    }
+
+    #[test]
+    #[cfg(feature = "alloc")]
+    fn basic_notification() {
+        loom::model(|| {
+            let chan = Arc::new(Chan {
+                num: AtomicUsize::new(0),
+                task: WaitCell::new(),
+            });
+
+            for _ in 0..NUM_NOTIFY {
+                let chan = chan.clone();
+
+                thread::spawn(move || {
+                    chan.num.fetch_add(1, Relaxed);
+                    chan.task.notify();
+                });
+            }
+
+            future::block_on(wait_on(chan));
+        });
+    }
+
+    #[test]
+    #[cfg(feature = "alloc")]
+    fn tx_close() {
+        loom::model(|| {
+            let chan = Arc::new(Chan {
+                num: AtomicUsize::new(0),
+                task: WaitCell::new(),
+            });
+
+            thread::spawn({
+                let chan = chan.clone();
+                move || {
+                    chan.num.fetch_add(1, Relaxed);
+                    chan.task.notify();
+                }
+            });
+
+            thread::spawn({
+                let chan = chan.clone();
+                move || {
+                    chan.num.fetch_add(1, Relaxed);
+                    chan.task.close_tx();
+                }
+            });
+
+            future::block_on(wait_on(chan));
+        });
+    }
+
+    #[test]
+    #[cfg(feature = "std")]
+    fn test_panicky_waker() {
+        use std::panic;
+        use std::ptr;
+        use std::task::{RawWaker, RawWakerVTable, Waker};
+
+        static PANICKING_VTABLE: RawWakerVTable =
+            RawWakerVTable::new(|_| panic!("clone"), |_| (), |_| (), |_| ());
+
+        let panicking = unsafe { Waker::from_raw(RawWaker::new(ptr::null(), &PANICKING_VTABLE)) };
+
+        loom::model(move || {
+            let chan = Arc::new(Chan {
+                num: AtomicUsize::new(0),
+                task: WaitCell::new(),
+            });
+
+            for _ in 0..NUM_NOTIFY {
+                let chan = chan.clone();
+
+                thread::spawn(move || {
+                    chan.num.fetch_add(1, Relaxed);
+                    chan.task.notify();
+                });
+            }
+
+            // Note: this panic should have no effect on the overall state of the
+            // waker and it should proceed as normal.
+            //
+            // A thread above might race to flag a wakeup, and a WAKING state will
+            // be preserved if this expected panic races with that so the below
+            // procedure should be allowed to continue uninterrupted.
+            let _ = panic::catch_unwind(|| chan.task.wait_with(|| panicking.clone()));
+
+            future::block_on(wait_on(chan));
+        });
+    }
+}