ScheduledIo 与就绪状态

pub(crate) struct ScheduledIo {
    pub(super) linked_list_pointers: UnsafeCell<linked_list::Pointers<Self>>,

    /// Packs the resource's readiness and I/O driver latest tick.
    readiness: AtomicUsize,

    waiters: Mutex<Waiters>,
}

// The `ScheduledIo::readiness` (`AtomicUsize`) is packed full of goodness.
//
// | shutdown | driver tick | readiness |
// |----------+-------------+-----------|
// |   1 bit  |   15 bits   |  16 bits  |

const READINESS: bit::Pack = bit::Pack::least_significant(16);

const TICK: bit::Pack = READINESS.then(15);

const SHUTDOWN: bit::Pack = TICK.then(1);

pub(super) fn set_readiness(&self, tick_op: Tick, f: impl Fn(Ready) -> Ready) {
    let _ = self.readiness.fetch_update(AcqRel, Acquire, |curr| {
        // If the io driver is shut down, then you are only allowed to clear readiness.
        debug_assert!(SHUTDOWN.unpack(curr) == 0 || matches!(tick_op, Tick::Clear(_)));

        const MAX_TICK: usize = TICK.max_value() + 1;
        let tick = TICK.unpack(curr);

        let new_tick = match tick_op {
            // Trying to clear readiness with an old event!
            Tick::Clear(t) if tick != usize::from(t) => return None,
            Tick::Clear(t) => usize::from(t),
            Tick::Set => tick.wrapping_add(1) % MAX_TICK,
        };
        let ready = Ready::from_usize(READINESS.unpack(curr));
        // Keep the shutdown bit, so that a clear after shutdown (a
        // `WouldBlock` observed once the driver is gone) does not turn the
        // next wait into a wait for an event that will never come.
        let next = TICK.pack(new_tick, f(ready).as_usize());
        Some(SHUTDOWN.pack(SHUTDOWN.unpack(curr), next))
    });
}

pub(crate) fn clear_readiness(&self, event: ReadyEvent) {
    // This consumes the current readiness state **except** for closed
    // states. Closed states are excluded because they are final states.
    let mask_no_closed = event.ready - Ready::READ_CLOSED - Ready::WRITE_CLOSED;
    self.set_readiness(Tick::Clear(event.tick), |curr| curr - mask_no_closed);
}

#[derive(Debug, Default)]
struct Waiters {
    /// List of all current waiters.
    list: LinkedList<Waiter>,

    /// Waker used for `AsyncRead`.
    reader: Option<Waker>,

    /// Waker used for `AsyncWrite`.
    writer: Option<Waker>,
}

use std::io::Write;
use std::sync::Arc;
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tokio::time::sleep;

async fn pair() -> (Arc<TcpStream>, std::net::TcpStream) {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let peer = std::thread::spawn(move || std::net::TcpStream::connect(addr).unwrap());
  let (s, _) = listener.accept().await.unwrap();
  // accept 出来的流被假定可读；先 try_read 一次，拿到 WouldBlock，把就绪位清掉
  let mut b = [0u8; 8];
  assert_eq!(
    s.try_read(&mut b).unwrap_err().kind(),
    std::io::ErrorKind::WouldBlock
  );
  (Arc::new(s), peer.join().unwrap())
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  for mode in ["poll_read_ready", "readable"] {
    let (s, mut peer) = pair().await;
    let mut tasks = Vec::new();
    for name in ["A", "B"] {
      let s = s.clone();
      tasks.push(tokio::spawn(async move {
        if mode == "poll_read_ready" {
          std::future::poll_fn(|cx| s.poll_read_ready(cx)).await.unwrap();
        } else {
          s.readable().await.unwrap();
        }
        println!("  {name} 被唤醒");
      }));
      sleep(Duration::from_millis(10)).await; // 让 A 先挂起，B 后挂起
    }
    println!("{mode}：A、B 都在等，对端写入");
    peer.write_all(b"x").unwrap();
    sleep(Duration::from_millis(200)).await;
    for (name, t) in ["A", "B"].iter().zip(&tasks) {
      if !t.is_finished() {
        println!("  {name} 200ms 后还在等");
      }
    }
    for t in tasks {
      t.abort();
    }
  }
}

poll_read_ready：A、B 都在等，对端写入
  B 被唤醒
  A 200ms 后还在等
readable：A、B 都在等，对端写入
  B 被唤醒
  A 被唤醒

pub(super) fn wake(&self, ready: Ready) {
    let mut wakers = WakeList::new();

    let mut waiters = self.waiters.lock();

    // check for AsyncRead slot
    if ready.is_readable() {
        if let Some(waker) = waiters.reader.take() {
            wakers.push(waker);
        }
    }

    // check for AsyncWrite slot
    if ready.is_writable() {
        if let Some(waker) = waiters.writer.take() {
            wakers.push(waker);
        }
    }

    'outer: loop {
        let mut iter = waiters.list.drain_filter(|w| ready.satisfies(w.interest));

        while wakers.can_push() {
            match iter.next() {
                Some(waiter) => {
                    let waiter = unsafe { &mut *waiter.as_ptr() };

                    if let Some(waker) = waiter.waker.take() {
                        waiter.is_ready = true;
                        wakers.push(waker);
                    }
                }
                None => {
                    break 'outer;
                }
            }
        }

        drop(waiters);

        wakers.wake_all();

        // Acquire the lock again.
        waiters = self.waiters.lock();
    }

    // Release the lock before notifying
    drop(waiters);

    wakers.wake_all();
}

struct Readiness<'a> {
    scheduled_io: &'a ScheduledIo,

    state: State,

    /// Entry in the waiter `LinkedList`.
    waiter: UnsafeCell<Waiter>,
}

enum State {
    Init,
    Waiting,
    Done,
}

use std::io::Write;
use std::os::unix::net::UnixStream;
use std::time::Duration;
use tokio::io::unix::AsyncFd;
use tokio::time::timeout;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  for case in ["对端写 1 字节", "对端关闭写方向"] {
    let (a, mut b) = UnixStream::pair().unwrap();
    a.set_nonblocking(true).unwrap();
    // SAFETY: `a` 归 AsyncFd 所有，fd 在 AsyncFd drop 之前一直打开
    let fd = unsafe { AsyncFd::register(a) }.unwrap();
    if case == "对端写 1 字节" {
      b.write_all(b"x").unwrap();
    } else {
      b.shutdown(std::net::Shutdown::Write).unwrap();
    }
    println!("{case}：");
    for round in 0..3 {
      match timeout(Duration::from_millis(200), fd.readable()).await {
        Ok(guard) => {
          let mut guard = guard.unwrap();
          let r = guard.ready();
          println!(
            "  第 {round} 轮：返回，is_readable={} is_read_closed={}",
            r.is_readable(),
            r.is_read_closed()
          );
          guard.clear_ready(); // 不读数据，直接清位
        }
        Err(_) => println!("  第 {round} 轮：200ms 内没返回"),
      }
    }
  }
}

对端写 1 字节：
  第 0 轮：返回，is_readable=true is_read_closed=false
  第 1 轮：200ms 内没返回
  第 2 轮：200ms 内没返回
对端关闭写方向：
  第 0 轮：返回，is_readable=true is_read_closed=true
  第 1 轮：返回，is_readable=true is_read_closed=true
  第 2 轮：返回，is_readable=true is_read_closed=true