oneshot、watch、broadcast、Notify

struct Inner<T> {
    /// Manages the state of the inner cell.
    state: AtomicUsize,

    /// The value. This is set by `Sender` and read by `Receiver`. The state of
    /// the cell is tracked by `state`.
    value: UnsafeCell<Option<T>>,

    /// The task to notify when the receiver drops without consuming the value.
    // ...
    tx_task: Task,

    /// The task to notify when the value is sent.
    // ...
    rx_task: Task,
}

struct Task(UnsafeCell<MaybeUninit<Waker>>);

struct Shared<T> {
    /// The most recent value.
    value: RwLock<T>,

    /// The current version.
    // ...
    state: AtomicState,

    /// Tracks the number of `Receiver` instances.
    ref_count_rx: AtomicUsize,

    /// Tracks the number of `Sender` instances.
    ref_count_tx: AtomicUsize,

    /// Notifies waiting receivers that the value changed.
    notify_rx: big_notify::BigNotify,

    /// Notifies any task listening for `Receiver` dropped events.
    notify_tx: Notify,
}

loop {
    // In order to avoid a race condition, we first request a notification,
    // **then** check the current value's version. If a new version exists,
    // the notification request is dropped.
    let notified = shared.notify_rx.notified();

    if let Some(ret) = maybe_changed(shared, version) {
        return ret;
    }

    notified.await;
    // loop around again in case the wake-up was spurious
}

pub(super) struct BigNotify {
    #[cfg(not(all(not(loom), feature = "sync", any(feature = "rt", feature = "macros"))))]
    next: std::sync::atomic::AtomicUsize,
    inner: [Notify; 8],
}

use std::time::Duration;
use tokio::sync::watch;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let (tx, mut rx) = watch::channel(0);
  for i in 1..=3 {
    tx.send(i).unwrap();
  }
  rx.changed().await.unwrap();
  println!("after 3 sends: changed() ok, value = {}", *rx.borrow_and_update());
  let r = tokio::time::timeout(Duration::from_millis(100), rx.changed()).await;
  println!("second changed(): {}", if r.is_ok() { "ready" } else { "timed out" });
  drop(tx);
  println!("after drop(tx): changed() = {:?}", rx.changed().await);
}

after 3 sends: changed() ok, value = 3
second changed(): timed out
after drop(tx): changed() = Err(RecvError(()))

struct Shared<T> {
    /// slots in the channel.
    buffer: Box<[Mutex<Slot<T>>]>,

    /// Mask a position -> index.
    mask: usize,

    /// Tail of the queue. Includes the rx wait list.
    tail: Mutex<Tail>,
    // ...
}

struct Tail {
    /// Next position to write to.
    pos: u64,

    /// Number of active receivers.
    rx_cnt: usize,

    /// True if the channel is closed.
    closed: bool,

    /// Receivers waiting for a value.
    waiters: LinkedList<Waiter>,
}

struct Slot<T> {
    /// Remaining number of receivers that are expected to see this value.
    // ...
    rem: usize,

    /// Uniquely identifies the `send` stored in the slot.
    pos: u64,

    /// The value being broadcast.
    // ...
    val: Option<T>,
}

// At this point, the receiver has lagged behind the sender by
// more than the channel capacity. The receiver will attempt to
// catch up by skipping dropped messages and setting the
// internal cursor to the **oldest** message stored by the
// channel.
let next = tail.pos.wrapping_sub(self.shared.buffer.len() as u64);

let missed = next.wrapping_sub(self.next);

drop(tail);

// The receiver is slow but no values have been missed
if missed == 0 {
    self.next = self.next.wrapping_add(1);

    return Ok(RecvGuard { slot });
}

self.next = next;

return Err(TryRecvError::Lagged(missed));

use tokio::sync::broadcast;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  // capacity 3 is rounded up to 4
  let (tx, mut rx) = broadcast::channel::<u32>(3);
  for i in 0..10 {
    tx.send(i).unwrap();
  }
  println!("rx.len() = {}", rx.len());
  loop {
    match rx.try_recv() {
      Ok(v) => println!("recv {v}"),
      Err(e) => {
        println!("err {e:?}");
        if matches!(e, broadcast::error::TryRecvError::Empty) {
          break;
        }
      }
    }
  }
}

rx.len() = 10
err Lagged(6)
recv 6
recv 7
recv 8
recv 9
err Empty

pub struct Notify {
    // `state` uses 2 bits to store one of `EMPTY`,
    // `WAITING` or `NOTIFIED`. The rest of the bits
    // are used to store the number of times `notify_waiters`
    // was called.
    //
    // Throughout the code there are two assumptions:
    // - state can be transitioned *from* `WAITING` only if
    //   `waiters` lock is held
    // - number of times `notify_waiters` was called can
    //   be modified only if `waiters` lock is held
    state: AtomicUsize,
    waiters: Mutex<LinkedList<Waiter>>,
}

// Load the current state
let mut curr = self.state.load(Acquire);

// If the state is `EMPTY`, transition to `NOTIFIED` and return.
while let EMPTY | NOTIFIED = get_state(curr) {
    // The compare-exchange from `NOTIFIED` -> `NOTIFIED` is intended. A
    // happens-before synchronization must happen between this atomic
    // operation and a task calling `notified().await`.
    let new = set_state(curr, NOTIFIED);
    let res = self.state.compare_exchange(curr, new, AcqRel, Acquire);

    match res {
        // No waiters, no further work to do
        Ok(_) => return,
        Err(actual) => {
            curr = actual;
        }
    }
}

// There are waiters, the lock must be acquired to notify.
let mut waiters = self.waiters.lock();

if matches!(get_state(curr), EMPTY | NOTIFIED) {
    // There are no waiting tasks. All we need to do is increment the
    // number of times this method was called.
    atomic_inc_num_notify_waiters_calls(&self.state);
    return;
}

// Increment the number of times this method was called
// and transition to empty.
let new_state = set_state(inc_num_notify_waiters_calls(curr), EMPTY);
self.state.store(new_state, Release);

pub fn notified(&self) -> Notified<'_> {
    // we load the number of times notify_waiters
    // was called and store that in the future.
    let state = self.state.load(Acquire);
    Notified {
        notify: self,
        state: State::Init,
        notify_waiters_calls: get_num_notify_waiters_calls(state),
        waiter: Waiter::new(),
    }
}

use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::timeout;

async fn wait(n: &Notify) -> &'static str {
  match timeout(Duration::from_millis(100), n.notified()).await {
    Ok(()) => "woken",
    Err(_) => "timed out",
  }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let n = Notify::new();
  n.notify_one();
  println!("1. notify_one, then notified().await: {}", wait(&n).await);

  let n = Notify::new();
  n.notify_one();
  n.notify_one();
  println!("2. notify_one x2, first wait: {}", wait(&n).await);
  println!("   second wait: {}", wait(&n).await);

  let n = Notify::new();
  n.notify_waiters();
  println!("3. notify_waiters, then notified().await: {}", wait(&n).await);

  let n = Notify::new();
  let f = n.notified(); // created, never polled
  n.notify_waiters();
  let r = timeout(Duration::from_millis(100), f).await;
  println!("4. notified() created, notify_waiters, then await: {}", if r.is_ok() { "woken" } else { "timed out" });

  // 5. two un-enabled futures, notify_one: the permit goes to whoever polls first
  let n = Notify::new();
  let a = n.notified();
  let mut b = std::pin::pin!(n.notified());
  b.as_mut().enable();
  n.notify_one();
  let ra = timeout(Duration::from_millis(100), a).await.is_ok();
  let rb = timeout(Duration::from_millis(100), b).await.is_ok();
  println!("5. a not enabled, b enabled, notify_one: a woken={ra}, b woken={rb}");
}

1. notify_one, then notified().await: woken
2. notify_one x2, first wait: woken
   second wait: timed out
3. notify_waiters, then notified().await: timed out
4. notified() created, notify_waiters, then await: woken
5. a not enabled, b enabled, notify_one: a woken=false, b woken=true

rx1: Err(Lagged(12)) then Ok(12)
rx2: Ok(12) then Ok(13)