没读到 WouldBlock 就回去 poll，程序卡住

mio = { version = "=1.2.4", features = ["os-poll", "net", "os-ext"] }

use std::io::{self, ErrorKind, Read, Write};
use std::time::Duration;

use mio::net::{TcpListener, TcpStream};
use mio::{Events, Interest, Poll, Token};

// one poll with a 1s timeout: the guard that keeps a hung loop from hanging the program
fn wait(poll: &mut Poll, events: &mut Events) -> io::Result<usize> {
  poll.poll(events, Some(Duration::from_secs(1)))?;
  Ok(events.iter().count())
}

fn stream_case(drain: bool) -> io::Result<()> {
  println!("== stream: peer sends 8 bytes, server reads with a 4-byte buffer");
  let std_listener = std::net::TcpListener::bind("127.0.0.1:0")?;
  let mut peer = std::net::TcpStream::connect(std_listener.local_addr()?)?;
  let (s, _) = std_listener.accept()?;
  s.set_nonblocking(true)?;
  let mut stream = TcpStream::from_std(s);

  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  poll.registry().register(&mut stream, Token(1), Interest::READABLE)?;
  peer.write_all(b"abcdefgh")?;

  let mut got = 0;
  let mut buf = [0u8; 4];
  while got < 8 {
    let n = wait(&mut poll, &mut events)?;
    if n == 0 {
      println!("poll: 0 event(s) after 1s, server has {got}/8 bytes");
      let n = stream.read(&mut buf)?;
      println!("read anyway: {n} byte(s) {:?}", String::from_utf8_lossy(&buf[..n]));
      return Ok(());
    }
    println!("poll: {n} event(s)");
    loop {
      match stream.read(&mut buf) {
        Ok(0) => break,
        Ok(n) => {
          got += n;
          println!("  read {n} byte(s) {:?}", String::from_utf8_lossy(&buf[..n]));
        }
        Err(e) if e.kind() == ErrorKind::WouldBlock => {
          println!("  read: WouldBlock");
          break;
        }
        Err(e) => return Err(e),
      }
      if !drain {
        break;
      }
    }
  }
  println!("server has {got}/8 bytes");
  Ok(())
}

fn listener_case(drain: bool) -> io::Result<()> {
  println!("== listener: two clients connect before the first poll");
  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
  let addr = listener.local_addr()?;
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  poll.registry().register(&mut listener, Token(0), Interest::READABLE)?;
  let _c1 = std::net::TcpStream::connect(addr)?;
  let _c2 = std::net::TcpStream::connect(addr)?;
  // let both handshakes finish on the server side before the first poll
  std::thread::sleep(Duration::from_millis(100));

  let mut accepted = Vec::new();
  while accepted.len() < 2 {
    let n = wait(&mut poll, &mut events)?;
    if n == 0 {
      println!("poll: 0 event(s) after 1s, accepted {}/2", accepted.len());
      let r = listener.accept().map(|_| "a connection").map_err(|e| e.kind());
      println!("accept anyway: {r:?}");
      return Ok(());
    }
    println!("poll: {n} event(s)");
    loop {
      match listener.accept() {
        Ok((s, _)) => {
          println!("  accept: Ok");
          accepted.push(s);
        }
        Err(e) if e.kind() == ErrorKind::WouldBlock => {
          println!("  accept: WouldBlock");
          break;
        }
        Err(e) => return Err(e),
      }
      if !drain {
        break;
      }
    }
  }
  println!("accepted {}/2", accepted.len());
  Ok(())
}

fn main() -> io::Result<()> {
  let drain = std::env::args().nth(1).as_deref() == Some("drain");
  println!("os={} mode={}", std::env::consts::OS, if drain { "drain" } else { "once" });
  stream_case(drain)?;
  listener_case(drain)?;
  Ok(())
}

os=macos mode=once
== stream: peer sends 8 bytes, server reads with a 4-byte buffer
poll: 1 event(s)
  read 4 byte(s) "abcd"
poll: 0 event(s) after 1s, server has 4/8 bytes
read anyway: 4 byte(s) "efgh"
== listener: two clients connect before the first poll
poll: 1 event(s)
  accept: Ok
poll: 0 event(s) after 1s, accepted 1/2
accept anyway: Ok("a connection")

os=linux mode=once
== stream: peer sends 8 bytes, server reads with a 4-byte buffer
poll: 1 event(s)
  read 4 byte(s) "abcd"
poll: 0 event(s) after 1s, server has 4/8 bytes
read anyway: 4 byte(s) "efgh"
== listener: two clients connect before the first poll
poll: 1 event(s)
  accept: Ok
poll: 0 event(s) after 1s, accepted 1/2
accept anyway: Ok("a connection")

os=linux mode=drain
== stream: peer sends 8 bytes, server reads with a 4-byte buffer
poll: 1 event(s)
  read 4 byte(s) "abcd"
  read 4 byte(s) "efgh"
  read: WouldBlock
server has 8/8 bytes
== listener: two clients connect before the first poll
poll: 1 event(s)
  accept: Ok
  accept: Ok
  accept: WouldBlock
accepted 2/2

write(1, "os=linux mode=once\n", 19)    = 19
write(1, "== stream: peer sends 8 bytes, s"..., 65) = 65
accept4(3, {sa_family=AF_INET, sin_port=htons(53422), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC) = 5
epoll_ctl(6, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
epoll_pwait(6, [{events=EPOLLIN, data=0x1}], 8, 1000, NULL, 8) = 1
write(1, "poll: 1 event(s)\n", 17)      = 17
recvfrom(5, "abcd", 4, 0, NULL, NULL)   = 4
write(1, "  read 4 byte(s) \"abcd\"\n", 24) = 24
epoll_pwait(6, [], 8, 1000, NULL, 8)    = 0
write(1, "poll: 0 event(s) after 1s, serve"..., 48) = 48
recvfrom(5, "efgh", 4, 0, NULL, NULL)   = 4
write(1, "read anyway: 4 byte(s) \"efgh\"\n", 30) = 30
write(1, "== listener: two clients connect"..., 55) = 55
epoll_ctl(4, EPOLL_CTL_ADD, 3, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
epoll_pwait(4, [{events=EPOLLIN, data=0}], 8, 1000, NULL, 8) = 1
write(1, "poll: 1 event(s)\n", 17)      = 17
accept4(3, {sa_family=AF_INET, sin_port=htons(56786), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 7
write(1, "  accept: Ok\n", 13)          = 13
epoll_pwait(4, [], 8, 1000, NULL, 8)    = 0
write(1, "poll: 0 event(s) after 1s, accep"..., 40) = 40
accept4(3, {sa_family=AF_INET, sin_port=htons(56800), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 8
write(1, "accept anyway: Ok(\"a connection\""..., 34) = 34
+++ exited with 0 +++

    pub(crate) fn do_io<T, F, R>(&self, f: F, io: &T) -> io::Result<R>
    where
        F: FnOnce(&T) -> io::Result<R>,
    {
        // We don't hold state, so we can just call the function and
        // return.
        f(io)
    }