unbounded_spsc/

lib.rs

//! This library adapts the block-waiting `recv` mechanism from the Rust standard
//! library to an unbounded SPSC channel backed by `spsc`.
//!
//! # Delivery guarantee
//!
//! A successful [`Sender::send`] (`Ok(())`) does not guarantee that the corresponding
//! [`Receiver::recv`] will observe the message. If the receiver is dropped concurrently
//! with a send, the in-flight value may be silently discarded (it is destructed
//! properly when the underlying queue is dropped; it is not leaked). This is a
//! deliberate weakening relative to the std-library stream-flavor channel from which
//! this design was originally adapted.

#![feature(negative_impls)]

use bounded_spsc_queue as spsc;

use std::sync::atomic::Ordering;

mod blocking;
mod select;

const DISCONNECTED     : isize = isize::MIN;
#[cfg(test)]
const MAX_STEALS       : isize = 5;
#[cfg(not(test))]
const MAX_STEALS       : isize = 1 << 20;   // ~1 million
const INITIAL_CAPACITY : usize = 128;

pub struct Receiver <T> {
  consumer    : std::cell::UnsafeCell <spsc::Consumer <T>>,
  receive_new : std::sync::mpsc::Receiver <spsc::Consumer <T>>,
  inner       : std::sync::Arc <Inner>,
  steals      : std::cell::UnsafeCell <isize>
}

pub struct Sender <T> {
  producer : std::cell::UnsafeCell <spsc::Producer <T>>,
  send_new : std::sync::mpsc::Sender <spsc::Consumer <T>>,
  inner    : std::sync::Arc <Inner>
}

struct Inner {
  counter   : std::sync::atomic::AtomicIsize,
  connected : std::sync::atomic::AtomicBool,
  to_wake   : std::sync::atomic::AtomicPtr <blocking::Inner>
}

#[derive(Debug)]
pub struct Iter <'a, T > {
  rx : &'a Receiver <T>
}

#[derive(Debug)]
pub struct TryIter <'a, T > {
  rx : &'a Receiver <T>
}

#[derive(Debug)]
pub struct IntoIter <T> {
  rx : Receiver <T>
}

/// Sender disconnected, no further messages will ever be received.
#[derive(Clone,Copy,Debug,Eq,PartialEq)]
pub struct RecvError;

/// Receiver disconnected, message will never be deliverable.
#[derive(Clone,Copy,Eq,PartialEq)]
pub struct SendError <T> (pub T);

#[derive(Clone,Copy,Debug,Eq,PartialEq)]
pub enum TryRecvError {
  Empty,
  Disconnected
}

#[derive(Clone,Copy,Debug,Eq,PartialEq)]
pub enum RecvTimeoutError {
  Timeout,
  Disconnected
}

pub enum SelectionResult {
  SelSuccess,
  SelCanceled
}

impl <T> Receiver <T> {
  /// Non-blocking receive, returns `Err(TryRecvError::Empty)` if buffer was empty; will
  /// continue to receive pending messages from a disconnected channel until it is
  /// empty, at which point further calls to this function will return
  /// `Err(TryRecvError::Disconnected)`.
  #[expect(clippy::missing_panics_doc)]
  pub fn try_recv (&self) -> Result <T, TryRecvError> {
    match unsafe { (*self.consumer.get()).try_pop() } {
      Some (t) => unsafe {
        if MAX_STEALS < *self.steals.get() {
          match self.inner.counter.swap (0, Ordering::SeqCst) {
            DISCONNECTED => {
              self.inner.counter.store (DISCONNECTED, Ordering::SeqCst);
            }
            n => {
              let m = std::cmp::min (n, *self.steals.get());
              *self.steals.get() -= m;
              self.bump (n - m);
            }
          }
          // TODO: can this be changed to a debug assertion ?
          assert!(0 <= *self.steals.get());
        }
        *self.steals.get() += 1;
        Ok (t)
      },
      None => {
        match self.receive_new.try_recv() {
          Ok (new_consumer) => {
            unsafe { *self.consumer.get() = new_consumer; }
            self.try_recv()
          },
          Err (std::sync::mpsc::TryRecvError::Empty) => {
            match self.inner.counter.load (Ordering::SeqCst) {
              n if n != DISCONNECTED => Err (TryRecvError::Empty),
              _ => {
                match unsafe { (*self.consumer.get()).try_pop() } {
                  Some (t) => Ok (t),
                  None     => Err (TryRecvError::Disconnected)
                }
              }
            }
          },
          Err (std::sync::mpsc::TryRecvError::Disconnected) => {
            Err (TryRecvError::Disconnected)
          }
        }
      }
    }
  }

  /// Block waiting if no messages are pending in the buffer.
  pub fn recv (&self) -> Result <T, RecvError> {
    match self.try_recv() {
      Err (TryRecvError::Empty) => {}
      Err (TryRecvError::Disconnected) => return Err (RecvError),
      Ok  (t) => return Ok (t)
    }
    let (wait_token, signal_token) = blocking::tokens();
    if self.decrement (signal_token).is_ok() {
      wait_token.wait();
    }
    match self.try_recv() {
      Ok (t) => unsafe {
        *self.steals.get() -= 1;
        Ok (t)
      },
      Err (TryRecvError::Empty) => unreachable!(
        "woken thread should have found pending message"),
      Err (TryRecvError::Disconnected) => Err (RecvError)
    }
  }

  pub fn recv_timeout (&self, timeout : std::time::Duration)
    -> Result <T, RecvTimeoutError>
  {
    match self.try_recv() {
      Ok  (t)                          => Ok (t),
      Err (TryRecvError::Disconnected) => Err (RecvTimeoutError::Disconnected),
      Err (TryRecvError::Empty)
        => self.recv_max_until (std::time::Instant::now() + timeout)
    }
  }

  #[expect(mismatched_lifetime_syntaxes)]
  pub const fn iter (&self) -> Iter <T> {
    Iter {
      rx: self
    }
  }

  #[expect(mismatched_lifetime_syntaxes)]
  pub const fn try_iter (&self) -> TryIter <T> {
    TryIter {
      rx: self
    }
  }

  pub fn capacity (&self) -> usize {
    unsafe {
      (*self.consumer.get()).capacity()
    }
  }

  fn recv_max_until (&self, deadline : std::time::Instant)
    -> Result <T, RecvTimeoutError>
  {
    loop {
      match self.recv_deadline (deadline) {
        result @ Err (RecvTimeoutError::Timeout) => {
          if deadline <= std::time::Instant::now() {
            return result
          }
        },
        result => return result
      }
    }
  }

  /// This is the same as `recv` except with code for timeout.
  fn recv_deadline (&self, deadline : std::time::Instant)
    -> Result <T, RecvTimeoutError>
  {
    match self.try_recv() {
      Err (TryRecvError::Empty) => {}
      Err (TryRecvError::Disconnected)
        => return Err (RecvTimeoutError::Disconnected),
      Ok  (t) => return Ok (t)
    }
    let (wait_token, signal_token) = blocking::tokens();
    if self.decrement (signal_token).is_ok() {
      let timed_out = !wait_token.wait_max_until (deadline);
      if timed_out {
        // this boolean result is not used: `try_recv` is always called below
        let _has_data = self.abort_selection_();
      }
    }
    match self.try_recv() {
      Ok (t) => unsafe {
        *self.steals.get() -= 1;
        Ok (t)
      }
      Err (TryRecvError::Empty)        => Err (RecvTimeoutError::Timeout),
      Err (TryRecvError::Disconnected) => Err (RecvTimeoutError::Disconnected)
    }
  }

  fn decrement (&self, token : std::sync::Arc <blocking::Inner>)
    -> Result <(), std::sync::Arc <blocking::Inner>>
  {
    assert_eq!(self.inner.to_wake.load (Ordering::SeqCst), std::ptr::null_mut());
    // Consume `token` and stash its raw pointer in `to_wake`, leaking one strong
    // reference into the AtomicPtr. The leaked ref is reclaimed either by
    // `Inner::take_to_wake` (when a sender wakes us) or by `Arc::from_raw` below (when
    // we cancel before blocking).
    //
    // NOTE: We must use `Arc::into_raw` (not `Arc::as_ptr`) here. Using `as_ptr` would
    // leave the strong-count unchanged; `token` would then be dropped at function exit,
    // releasing the ref, while the dangling-looking pointer remained in `to_wake`. A
    // subsequent `take_to_wake` -> `Arc::from_raw` would over-decrement the strong
    // count and free the allocation while another `Arc` still pointed at it -- a
    // use-after-free / double-free that manifests as `tcache_thread_shutdown: unaligned
    // tcache chunk detected` under glibc.
    let ptr = std::sync::Arc::into_raw (token).cast_mut();
    self.inner.to_wake.store (ptr, Ordering::SeqCst);
    let steals = unsafe { std::ptr::replace (self.steals.get(), 0) };
    match self.inner.counter.fetch_sub (1 + steals, Ordering::SeqCst) {
      DISCONNECTED => {
        self.inner.counter.store (DISCONNECTED, Ordering::SeqCst);
      }
      n => {
        assert!(0 <= n);
        if n - steals <= 0 {
          // Sender will reclaim the leaked ref via `take_to_wake`.
          return Ok (())
        }
      }
    }
    self.inner.to_wake.store (std::ptr::null_mut(), Ordering::SeqCst);
    // Cancelled before blocking: no sender will read `to_wake`, so we reclaim the
    // leaked strong reference ourselves and return it. SAFETY: `ptr` was produced by
    // `Arc::into_raw` above and no other code path observed it (we just swapped null
    // back into `to_wake`).
    Err (unsafe { std::sync::Arc::from_raw (ptr) })
  }

  /////////////////////////////////////////////////////////////////////////////
  //  select functions
  /////////////////////////////////////////////////////////////////////////////

  fn can_recv_ (&self) -> bool {
    0 < unsafe { (*self.consumer.get()).size() }
  }

  fn start_selection_ (&self, token : std::sync::Arc <blocking::Inner>)
    -> SelectionResult
  {
    match self.decrement (token) {
      Ok  (()) => SelectionResult::SelSuccess,
      Err (_token) => {
        // undo decrement above
        let prev = self.bump (1);
        assert!(prev == DISCONNECTED || 0 <= prev);
        SelectionResult::SelCanceled
      }
    }
  }

  /// Returns true if receiver has data pending.
  fn abort_selection_ (&self) -> bool {
    let steals = 1;
    let prev = self.bump (steals + 1);
    if prev == DISCONNECTED {
      assert_eq!(self.inner.to_wake.load (Ordering::SeqCst),
        std::ptr::null_mut());
      true
    } else {
      let cur = prev + steals + 1;
      assert!(0 <= cur);
      if prev < 0 {
        drop (self.inner.take_to_wake());
      } else {
        while !self.inner.to_wake.load (Ordering::SeqCst).is_null() {
          std::thread::yield_now();
        }
      }
      unsafe {
        assert_eq!(*self.steals.get(), 0);
        *self.steals.get() = steals;
      }
      0 <= prev
    }
  }

  fn bump (&self, amt : isize) -> isize {
    match self.inner.counter.fetch_add (amt, Ordering::SeqCst) {
      DISCONNECTED => {
        self.inner.counter.store (DISCONNECTED, Ordering::SeqCst);
        DISCONNECTED
      }
      n => n
    }
  }

} // end impl Receiver

impl <T> std::fmt::Debug for Receiver <T> {
  fn fmt (&self, f : &mut std::fmt::Formatter) -> std::fmt::Result {
    write!(f, "Receiver {{ .. }}")
  }
}

impl <T> IntoIterator for Receiver <T> {
  type Item = T;
  type IntoIter = IntoIter <T>;
  fn into_iter (self) -> IntoIter <T> {
    IntoIter {
      rx: self
    }
  }
}

impl <'a, T> IntoIterator for &'a Receiver <T> {
  type Item = T;
  type IntoIter = Iter <'a, T>;
  fn into_iter (self) -> Iter <'a, T> {
    self.iter()
  }
}

impl <T> Drop for Receiver <T> {
  fn drop (&mut self) {
    // Gate future sends: once `connected` is observed `false`, `Sender::send` bails out
    // before pushing, so after this point at most ONE in-flight `send` (whose
    // `connected.load` already returned `true`) may still commit, after which no more
    // pushes ever happen. The drain loop below relies on this bound.
    self.inner.connected.store (false, Ordering::SeqCst);

    // Drain remaining messages so that:
    //   1. their destructors run *now*, while the receiver's `T` type is still in scope
    //      (the bounded queues' `Buffer::drop` will of course also drop them, but only
    //      when the last `Arc<Buffer<T>>` is released, which can be later); and
    //   2. the channel `counter` can be reconciled with our local `steals` and CAS'd to
    //      `DISCONNECTED`, which is what informs senders that the orphan path applies
    //      (see `Sender::send`).
    //
    // Because this crate grows the channel by allocating a fresh `bounded-spsc-queue`
    // and publishing the new `Consumer` via the `send_new` / `receive_new`
    // `std::sync::mpsc` side channel, draining requires walking that side channel too:
    // items pushed after a resize live in queues whose `Consumer` is still buffered in
    // `receive_new`, not in `*self.consumer.get()`. Without this side-channel walk, the
    // CAS below never observes `counter == steals` and the loop spins forever (this
    // manifested as a hang in `tests::recv_try_iter` when run alongside other tests).
    let mut steals = unsafe { *self.steals.get() };
    // Bound on consecutive `yield_now`s with no observable progress. If the sender is
    // parked on an unrelated channel rather than mid-send, the counter will never
    // converge on `steals` here, but the orphan items (if any) are still safe --
    // `Buffer::drop` will destruct them when the bounded queues' `Arc<Buffer<T>>`
    // refcounts hit zero.
    const MAX_IDLE_YIELDS : u32 = 32;
    let mut idle_yields = 0u32;
    while {
      let count = self.inner.counter.compare_exchange (
        steals, DISCONNECTED, Ordering::SeqCst, Ordering::SeqCst
      ).unwrap_or_else (|i| i);
      count != DISCONNECTED && count != steals
    } {
      // Drain the current consumer.
      let mut drained_here = 0;
      while let Some (_t) = unsafe { (*self.consumer.get()).try_pop() } {
        steals += 1;
        drained_here += 1;
      }
      // Try to pick up the next post-resize consumer from the side channel.
      match self.receive_new.try_recv() {
        Ok (new_consumer) => unsafe {
          *self.consumer.get() = new_consumer;
          idle_yields = 0;
        }
        Err (std::sync::mpsc::TryRecvError::Empty) => {
          if drained_here != 0 {
            idle_yields = 0;
          } else {
            idle_yields += 1;
            if idle_yields >= MAX_IDLE_YIELDS {
              // Give up: store DISCONNECTED unconditionally. The CAS in the loop
              // condition would otherwise spin forever (sender parked on another
              // channel, no further progress observable here).
              self.inner.counter.store (DISCONNECTED, Ordering::SeqCst);
              break;
            }
            std::thread::yield_now();
          }
        }
        Err (std::sync::mpsc::TryRecvError::Disconnected) => {
          // Sender is gone entirely; no further new consumers will arrive. Force
          // DISCONNECTED state and let `Buffer::drop` clean up the rest.
          self.inner.counter.store (DISCONNECTED, Ordering::SeqCst);
          break;
        }
      }
    }
  }
}

impl <T> Sender <T> {
  /// Non-blocking send.
  ///
  /// # Delivery guarantee
  ///
  /// Returning `Ok(())` indicates that the message was successfully enqueued, but it
  /// does not guarantee that the receiver will observe it: if the [`Receiver`] is
  /// dropped concurrently with this call, the message may be silently discarded. The
  /// orphaned value is destructed properly when the underlying queue is dropped; it is
  /// not leaked.
  //
  // This is a deliberate weakening relative to the std-library stream-flavor channel
  // from which this design was originally adapted. The std version returned the orphan
  // to the caller via `Err(SendError(t))`, but the underlying `bounded-spsc-queue` does
  // not expose a safe way to recover an already-pushed value from the producer side,
  // and the previous implementation that did so via `std::mem::transmute` was unsound
  // (see issues #1 and #4).
  #[expect(clippy::missing_panics_doc)]
  pub fn send (&self, t : T) -> Result <(), SendError <T>> {
    if self.inner.connected.load (Ordering::SeqCst) {
      match unsafe { (*self.producer.get()).try_push (t) } {
        None     => {}, // success
        Some (t) => {   // queue full
          let new_capacity = 2 * unsafe { (*self.producer.get()).capacity() };
          let (new_producer, new_consumer) = spsc::make (new_capacity);
          if self.send_new.send (new_consumer).is_err() {
            // Receiver may have dropped since `connected.load()`, treat as disconnected
            return Err (SendError (t));
          }
          unsafe { *self.producer.get() = new_producer; }
          match unsafe { (*self.producer.get()).try_push (t) } {
            None      => {}
            Some (_t) => unreachable!(
              "send on a newly created queue should always succeed")
          }
        }
      }
      // TODO: can we replace asserts with debug assertions ?
      match self.inner.counter.fetch_add (1, Ordering::SeqCst) {
        -1 => {
          self.inner.take_to_wake().signal();
        },
        -2 => {},
        DISCONNECTED => {
          self.inner.counter.store (DISCONNECTED, Ordering::SeqCst);
          // The receiver disconnected after we successfully pushed the message onto the
          // bounded queue. The value is now orphaned, but it is *not* leaked: it will
          // be properly dropped either by the receiver's drain loop in `<Receiver<T> as
          // Drop>::drop`, or by `bounded_spsc_queue::Buffer::<T>::drop`, which pops and
          // drops every remaining element when the last `Arc<Buffer<T>>` reference is
          // released.
          //
          // Historical note: the previous implementation here transmuted the producer
          // pointer into a `Consumer<T>` value in order to pop the orphan back and
          // return it via `Err(SendError(t))`. That transmute was unsound (issues #1
          // and #4): the transmuted "Consumer"'s internal `Arc<Buffer<T>>` field
          // pointed into the `Sender` stack frame, not a real heap allocation, so
          // `try_pop` performed an out-of-bounds read and the fake `Arc`'s `Drop` would
          // call `dealloc` on a non-allocated address. Recovering the value safely is
          // not possible without an upstream change to `bounded-spsc-queue` to expose a
          // producer-side pop or a `Consumer<T>` recovery primitive. See the `send`
          // doc-comment for the resulting (weakened) delivery guarantee.
        },
        n => {
          assert! (0 <= n);
        }
      }
      Ok (())
    } else {
      Err (SendError (t))
    }
  }
} // end impl Sender

impl <T> std::fmt::Debug for Sender <T> {
  fn fmt (&self, f : &mut std::fmt::Formatter) -> std::fmt::Result {
    write!(f, "Sender {{ .. }}")
  }
}

impl <T> Drop for Sender <T> {
  fn drop (&mut self) {
    self.inner.connected.store (false, Ordering::SeqCst);
    match self.inner.counter.swap (DISCONNECTED, Ordering::SeqCst) {
      DISCONNECTED => {}
      -1 => {
        self.inner.take_to_wake().signal();
      }
      n  => {
        assert!(0 <= n);
      }
    }
  }
}

impl Inner {
  fn take_to_wake (&self) -> std::sync::Arc <blocking::Inner> {
    let ptr = self.to_wake.swap (std::ptr::null_mut(), Ordering::SeqCst);
    assert!(!ptr.is_null());
    unsafe {
      std::sync::Arc::from_raw (ptr)
    }
  }
}

impl <T> Iterator for Iter <'_, T> {
  type Item = T;
  fn next (&mut self) -> Option <T> {
    self.rx.recv().ok()
  }
}

impl <T> Iterator for TryIter <'_, T> {
  type Item = T;
  fn next (&mut self) -> Option <T> {
    self.rx.try_recv().ok()
  }
}

impl <T> Iterator for IntoIter <T> {
  type Item = T;
  fn next (&mut self) -> Option <T> {
    self.rx.recv().ok()
  }
}

impl std::fmt::Display for RecvError {
  fn fmt (&self, f : &mut std::fmt::Formatter) -> std::fmt::Result {
    "receiving on a closed channel".fmt (f)
  }
}

impl std::error::Error for RecvError {
  fn description (&self) -> &'static str {
    "receiving on a closed channel"
  }

  fn cause (&self) -> Option <&dyn std::error::Error> {
    None
  }
}

impl <T> std::fmt::Debug for SendError <T> {
  fn fmt (&self, f : &mut std::fmt::Formatter) -> std::fmt::Result {
    "SendError(..)".fmt (f)
  }
}

impl <T> std::fmt::Display for SendError <T> {
  fn fmt (&self, f : &mut std::fmt::Formatter) -> std::fmt::Result {
    "sending on a closed channel".fmt (f)
  }
}

impl <T : Send> std::error::Error for SendError <T> {
  fn description (&self) -> &'static str {
    "sending on a closed channel"
  }

  fn cause (&self) -> Option <&dyn std::error::Error> {
    None
  }
}

impl std::fmt::Display for TryRecvError {
  fn fmt (&self, f : &mut std::fmt::Formatter) -> std::fmt::Result {
    match *self {
      TryRecvError::Empty        => "receiving on an empty channel".fmt (f),
      TryRecvError::Disconnected => "receiving on a closed channel".fmt (f)
    }
  }
}

impl std::error::Error for TryRecvError {
  fn description (&self) -> &str {
    match *self {
      TryRecvError::Empty        => "receiving on an empty channel",
      TryRecvError::Disconnected => "receiving on a closed channel"
    }
  }

  fn cause (&self) -> Option <&dyn std::error::Error> {
    None
  }
}

pub fn channel <T : 'static> () -> (Sender <T>, Receiver <T>) {
  let (producer, consumer) = spsc::make (INITIAL_CAPACITY);
  let (send_new, receive_new) = std::sync::mpsc::channel();
  let inner = std::sync::Arc::new (
    Inner {
      counter:   std::sync::atomic::AtomicIsize::new (0),
      connected: std::sync::atomic::AtomicBool::new (true),
      to_wake:   std::sync::atomic::AtomicPtr::new (std::ptr::null_mut())
    }
  );
  let sender    = Sender {
    producer: std::cell::UnsafeCell::new (producer),
    send_new,
    inner: inner.clone()
  };
  let receiver  = Receiver {
    consumer: std::cell::UnsafeCell::new (consumer),
    receive_new,
    steals: std::cell::UnsafeCell::new (0),
    inner
  };
  (sender, receiver)
}

#[cfg(test)]
mod tests {
  use super::*;

  pub(crate) fn stress_factor() -> usize {
    match std::env::var ("RUST_TEST_STRESS") {
      Ok  (val) => val.parse().unwrap(),
      Err (..)  => 1,
    }
  }

  #[test]
  fn smoke() {
    let (tx, rx) = channel::<i32>();
    tx.send (1).unwrap();
    assert_eq!(rx.recv().unwrap(), 1);
  }

  #[test]
  fn drop_full() {
    let (tx, _rx) = channel::<Box <isize>>();
    tx.send(Box::new (1)).unwrap();
  }

  // TODO: test failed on an unwrap
  #[test]
  fn smoke_threads() {
    let (tx, rx) = channel::<i32>();
    let _t = std::thread::spawn (move|| {
      println!("smoke threads sending...");
      tx.send (1).unwrap();
    });
    println!("smoke threads receiving...");
    assert_eq!(rx.recv().unwrap(), 1);
  }

  #[test]
  fn smoke_port_gone() {
    let (tx, rx) = channel::<i32>();
    drop (rx);
    assert!(tx.send (1).is_err());
  }

  #[test]
  fn smoke_shared_port_gone() {
    let (tx, rx) = channel::<i32>();
    drop (rx);
    assert!(tx.send (1).is_err())
  }

  #[test]
  fn port_gone_concurrent() {
    let (tx, rx) = channel::<i32>();
    let _t = std::thread::spawn (move|| {
      rx.recv().unwrap();
    });
    while tx.send (1).is_ok() {}
  }

  #[test]
  fn smoke_chan_gone() {
    let (tx, rx) = channel::<i32>();
    drop (tx);
    rx.recv().unwrap_err();
  }

  #[test]
  fn chan_gone_concurrent() {
    let (tx, rx) = channel::<i32>();
    let _t = std::thread::spawn (move|| {
      tx.send (1).unwrap();
      tx.send (1).unwrap();
    });
    while rx.recv().is_ok() {}
  }

  #[test]
  fn stress() {
    let (tx, rx) = channel::<i32>();
    let t = std::thread::spawn (move|| {
      for _ in 0..10000 { tx.send (1).unwrap(); }
    });
    for _ in 0..10000 {
      assert_eq!(rx.recv().unwrap(), 1);
    }
    t.join().ok().unwrap();
  }

  #[test]
  fn send_from_outside_runtime() {
    let (tx1, rx1) = channel::<bool>();
    let (tx2, rx2) = channel::<i32>();
    let t1 = std::thread::spawn (move|| {
      tx1.send (true).unwrap();
      for _ in 0..40 {
        assert_eq!(rx2.recv().unwrap(), 1);
      }
    });
    rx1.recv().unwrap();
    let t2 = std::thread::spawn (move|| {
      for _ in 0..40 {
        tx2.send (1).unwrap();
      }
    });
    t1.join().ok().unwrap();
    t2.join().ok().unwrap();
  }

  #[test]
  fn recv_from_outside_runtime() {
    let (tx, rx) = channel::<i32>();
    let t = std::thread::spawn (move|| {
      for _ in 0..40 {
        assert_eq!(rx.recv().unwrap(), 1);
      }
    });
    for _ in 0..40 {
      tx.send (1).unwrap();
    }
    t.join().ok().unwrap();
  }

  #[test]
  fn no_runtime() {
    let (tx1, rx1) = channel::<i32>();
    let (tx2, rx2) = channel::<i32>();
    let t1 = std::thread::spawn (move|| {
      assert_eq!(rx1.recv().unwrap(), 1);
      tx2.send (2).unwrap();
    });
    let t2 = std::thread::spawn (move|| {
      tx1.send (1).unwrap();
      assert_eq!(rx2.recv().unwrap(), 2);
    });
    t1.join().ok().unwrap();
    t2.join().ok().unwrap();
  }

  #[test]
  fn oneshot_single_thread_close_port_first() {
    // Simple test of closing without sending
    let (_tx, rx) = channel::<i32>();
    drop (rx);
  }

  #[test]
  fn oneshot_single_thread_close_chan_first() {
    // Simple test of closing without sending
    let (tx, _rx) = channel::<i32>();
    drop (tx);
  }

  #[test]
  fn oneshot_single_thread_send_port_close() {
    // Testing that the sender cleans up the payload if receiver is closed
    let (tx, rx) = channel::<Box <i32>>();
    drop (rx);
    assert!(tx.send (Box::new (0)).is_err());
  }

  #[test]
  fn oneshot_single_thread_recv_chan_close() {
    // Receiving on a closed chan will panic
    let res = std::thread::spawn (move|| {
      let (tx, rx) = channel::<i32>();
      drop (tx);
      rx.recv().unwrap();
    }).join();
    // What is our res?
    assert!(res.is_err());
  }

  #[test]
  fn oneshot_single_thread_send_then_recv() {
    let (tx, rx) = channel::<Box <i32>>();
    tx.send (Box::new (10)).unwrap();
    assert!(*rx.recv().unwrap() == 10);
  }

  #[test]
  fn oneshot_single_thread_try_send_open() {
    let (tx, rx) = channel::<i32>();
    tx.send (10).unwrap();
    assert!(rx.recv().unwrap() == 10);
  }

  #[test]
  fn oneshot_single_thread_try_send_closed() {
    let (tx, rx) = channel::<i32>();
    drop (rx);
    assert!(tx.send (10).is_err());
  }

  #[test]
  fn oneshot_single_thread_try_recv_open() {
    let (tx, rx) = channel::<i32>();
    tx.send (10).unwrap();
    assert!(rx.recv() == Ok (10));
  }

  #[test]
  fn oneshot_single_thread_try_recv_closed() {
    let (tx, rx) = channel::<i32>();
    drop (tx);
    rx.recv().unwrap_err();
  }

  #[test]
  fn oneshot_single_thread_peek_data() {
    let (tx, rx) = channel::<i32>();
    assert_eq!(rx.try_recv(), Err (TryRecvError::Empty));
    tx.send (10).unwrap();
    assert_eq!(rx.try_recv(), Ok (10));
  }

  #[test]
  fn oneshot_single_thread_peek_close() {
    let (tx, rx) = channel::<i32>();
    drop (tx);
    assert_eq!(rx.try_recv(), Err (TryRecvError::Disconnected));
    assert_eq!(rx.try_recv(), Err (TryRecvError::Disconnected));
  }

  #[test]
  fn oneshot_single_thread_peek_open() {
    let (_tx, rx) = channel::<i32>();
    assert_eq!(rx.try_recv(), Err (TryRecvError::Empty));
  }

  #[test]
  fn oneshot_multi_task_recv_then_send () {
    let (tx, rx) = channel::<Box <i32>>();
    let _t = std::thread::spawn (move|| {
      assert!(*rx.recv().unwrap() == 10);
    });

    tx.send (Box::new (10)).unwrap();
  }

  #[test]
  fn oneshot_multi_task_recv_then_close() {
    let (tx, rx) = channel::<Box <i32>>();
    let _t = std::thread::spawn (move|| {
      drop (tx);
    });
    let res = std::thread::spawn (move|| {
      assert!(*rx.recv().unwrap() == 10);
    }).join();
    assert!(res.is_err());
  }

  #[test]
  fn oneshot_multi_thread_close_stress() {
    for _ in 0..stress_factor() {
      let (tx, rx) = channel::<i32>();
      let _t = std::thread::spawn (move|| {
        drop (rx);
      });
      drop (tx);
    }
  }

  #[test]
  fn oneshot_multi_thread_send_close_stress() {
    for _ in 0..stress_factor() {
      let (tx, rx) = channel::<i32>();
      let _t = std::thread::spawn (move|| {
        drop (rx);
      });
      let _ = std::thread::spawn (move|| {
        tx.send (1).unwrap();
      }).join();
    }
  }

  #[test]
  fn oneshot_multi_thread_recv_close_stress() {
    for _ in 0..stress_factor() {
      let (tx, rx) = channel::<i32>();
      std::thread::spawn (move|| {
        let res = std::thread::spawn (move|| {
          rx.recv().unwrap();
        }).join();
        assert!(res.is_err());
      });
      let _t = std::thread::spawn (move|| {
        std::thread::spawn (move|| {
          drop (tx);
        });
      });
    }
  }

  #[test]
  fn oneshot_multi_thread_send_recv_stress() {
    for _ in 0..stress_factor() {
      let (tx, rx) = channel::<Box <isize>>();
      let _t = std::thread::spawn (move|| {
        tx.send (Box::new (10)).unwrap();
      });
      assert!(*rx.recv().unwrap() == 10);
    }
  }

  #[test]
  fn stream_send_recv_stress() {
    for _ in 0..stress_factor() {
      let (tx, rx) = channel();

      send (tx, 0);
      recv (rx, 0);

      fn send (tx: Sender<Box <i32>>, i: i32) {
        if i == 10 { return }

        std::thread::spawn (move|| {
          tx.send (Box::new (i)).unwrap();
          send (tx, i + 1);
        });
      }

      fn recv (rx: Receiver<Box <i32>>, i: i32) {
        if i == 10 { return }

        std::thread::spawn (move|| {
          assert!(*rx.recv().unwrap() == i);
          recv (rx, i + 1);
        });
      }
    }
  }

  #[test]
  fn oneshot_single_thread_recv_timeout() {
    let (tx, rx) = channel();
    tx.send (true).unwrap();
    assert_eq!(rx.recv_timeout (std::time::Duration::from_millis (1)), Ok (true));
    assert_eq!(rx.recv_timeout (std::time::Duration::from_millis (1)),
      Err (RecvTimeoutError::Timeout));
    tx.send (true).unwrap();
    assert_eq!(rx.recv_timeout (std::time::Duration::from_millis (1)), Ok (true));
  }

  #[test]
  fn stress_recv_timeout_two_threads() {
    let (tx, rx) = channel();
    let stress = stress_factor() + 100;
    let timeout = std::time::Duration::from_millis (100);

    std::thread::spawn (move || {
      for i in 0..stress {
        if i % 2 == 0 {
          std::thread::sleep (timeout * 2);
        }
        tx.send (1usize).unwrap();
      }
    });

    let mut recv_count = 0;
    loop {
      match rx.recv_timeout (timeout) {
        Ok (n) => {
          assert_eq!(n, 1usize);
          recv_count += 1;
        }
        Err (RecvTimeoutError::Timeout) => { }
        Err (RecvTimeoutError::Disconnected) => break
      }
    }

    assert_eq!(recv_count, stress);
  }

  #[test]
  fn recv_a_lot() {
    // Regression test that we don't run out of stack in scheduler context
    let (tx, rx) = channel();
    for _ in 0..10000 { tx.send (true).unwrap(); }
    for _ in 0..10000 { rx.recv().unwrap(); }
  }

  #[test]
  fn nested_recv_iter() {
    let (tx, rx) = channel::<i32>();
    let (total_tx, total_rx) = channel::<i32>();

    let _t = std::thread::spawn (move|| {
      let mut acc = 0;
      for x in rx.iter() {
        acc += x;
      }
      total_tx.send (acc).unwrap();
    });

    tx.send (3).unwrap();
    tx.send (1).unwrap();
    tx.send (2).unwrap();
    drop (tx);
    assert_eq!(total_rx.recv().unwrap(), 6);
  }

  #[test]
  fn recv_iter_break() {
    let (tx, rx) = channel::<i32>();
    let (count_tx, count_rx) = channel();

    let _t = std::thread::spawn (move|| {
      let mut count = 0;
      for x in rx.iter() {
        if count >= 3 {
          break;
        } else {
          count += x;
        }
      }
      count_tx.send (count).unwrap();
    });

    tx.send (2).unwrap();
    tx.send (2).unwrap();
    tx.send (2).unwrap();
    let _ = tx.send (2);
    drop (tx);
    assert_eq!(count_rx.recv().unwrap(), 4);
  }

  #[test]
  fn recv_try_iter() {
    let (request_tx, request_rx) = channel();
    let (response_tx, response_rx) = channel();

    // Request `x`s until we have `6`.
    let t = std::thread::spawn (move|| {
      let mut count = 0;
      loop {
        for x in response_rx.try_iter() {
          count += x;
          if count == 6 {
            return count;
          }
        }
        println!("test recv try iter send request...");
        request_tx.send (true).unwrap();
      }
    });

    for _ in request_rx.iter() {
      println!("test recv try iter send response...");
      if response_tx.send (2).is_err() {
        break;
      }
    }

    println!("test recv try iter join...");

    assert_eq!(t.join().unwrap(), 6);
  }

  #[test]
  fn recv_into_iter_owned() {
    let mut iter = {
      let (tx, rx) = channel::<i32>();
      tx.send (1).unwrap();
      tx.send (2).unwrap();

      rx.into_iter()
    };
    assert_eq!(iter.next().unwrap(), 1);
    assert_eq!(iter.next().unwrap(), 2);
    assert!(iter.next().is_none());
  }

  #[test]
  fn recv_into_iter_borrowed() {
    let (tx, rx) = channel::<i32>();
    tx.send (1).unwrap();
    tx.send (2).unwrap();
    drop (tx);
    let mut iter = (&rx).into_iter();
    assert_eq!(iter.next().unwrap(), 1);
    assert_eq!(iter.next().unwrap(), 2);
    assert!(iter.next().is_none());
  }

  // TODO: test failed unwrap on RecvError
  #[test]
  fn try_recv_states() {
    let (tx1, rx1) = channel::<i32>();
    let (tx2, rx2) = channel::<bool>();
    let (tx3, rx3) = channel::<bool>();
    let _t = std::thread::spawn (move|| {
      rx2.recv().unwrap();
      tx1.send (1).unwrap();
      tx3.send (true).unwrap();
      rx2.recv().unwrap();
      drop (tx1);
      tx3.send (true).unwrap();
    });

    assert_eq!(rx1.try_recv(), Err (TryRecvError::Empty));
    tx2.send (true).unwrap();
    rx3.recv().unwrap();
    assert_eq!(rx1.try_recv(), Ok (1));
    assert_eq!(rx1.try_recv(), Err (TryRecvError::Empty));
    tx2.send (true).unwrap();
    rx3.recv().unwrap();
    assert_eq!(rx1.try_recv(), Err (TryRecvError::Disconnected));
  }

  #[test]
  fn issue_32114() {
    let (tx, _) = channel();
    let _ = tx.send (123);
    assert_eq!(tx.send (123), Err (SendError (123)));
  }

  #[test]
  fn zero_size() {
    let (tx, rx) = channel::<()>();
    tx.send (()).unwrap();
    let () = rx.recv().unwrap();
  }

  #[test]
  fn race_disconnect_does_not_corrupt_sender_or_abort() {
    for _ in 0..200 {
      let (tx, rx) = channel::<Box<u64>>();
      let h = std::thread::spawn(move || {
        for _ in 0..10_000 {
          let _ = tx.send(Box::new(0xDEAD_BEEF));
        }
      });
      drop(rx);
      h.join().unwrap();
    }
  }
}