from_std 之后事件循环卡在 read 里

mio = { version = "=1.2.4", features = ["os-poll", "net", "os-ext"] }
libc = "=0.2.191"
tokio = { version = "=1.54.1", features = ["rt", "net", "macros", "io-util", "time"] }

use std::io::{self, ErrorKind, Read, Write};
use std::os::fd::AsRawFd;
use std::thread;
use std::time::Duration;

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

// O_NONBLOCK as the kernel sees it
fn is_nonblocking(fd: &impl AsRawFd) -> bool {
  let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) };
  flags & libc::O_NONBLOCK != 0
}

fn main() -> io::Result<()> {
  println!("os={}", std::env::consts::OS);
  let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
  let addr = listener.local_addr()?;
  let mut peer_a = std::net::TcpStream::connect(addr)?;
  let (a, _) = listener.accept()?; // forgot set_nonblocking
  let mut peer_b = std::net::TcpStream::connect(addr)?;
  let (b, _) = listener.accept()?;
  b.set_nonblocking(true)?;

  let mut conns = vec![TcpStream::from_std(a), TcpStream::from_std(b)];
  for (i, c) in conns.iter().enumerate() {
    println!("conn {i}: O_NONBLOCK={}", is_nonblocking(c));
  }
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  for (i, c) in conns.iter_mut().enumerate() {
    poll.registry().register(c, Token(i), Interest::READABLE)?;
  }

  peer_a.write_all(b"from a")?;
  peer_b.write_all(b"from b")?;
  thread::sleep(Duration::from_millis(100));

  // guard: peer b waits for an echo that never comes
  thread::spawn(move || {
    peer_b.set_read_timeout(Some(Duration::from_secs(1))).unwrap();
    let mut buf = [0u8; 16];
    let r = peer_b.read(&mut buf).map_err(|e| e.kind());
    println!("guard: peer b got no echo within 1s ({r:?}), exiting");
    std::process::exit(0);
  });

  poll.poll(&mut events, None)?;
  let ready: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  println!("poll: tokens {ready:?}");
  let mut buf = [0u8; 64];
  for t in ready {
    let c = &mut conns[t];
    loop {
      match c.read(&mut buf) {
        Ok(n) => {
          println!("conn {t}: read {n} byte(s), echo");
          c.write_all(&buf[..n])?;
        }
        Err(e) if e.kind() == ErrorKind::WouldBlock => {
          println!("conn {t}: WouldBlock");
          break;
        }
        Err(e) => return Err(e),
      }
    }
  }
  println!("all events handled");
  Ok(())
}

os=linux
conn 0: O_NONBLOCK=false
conn 1: O_NONBLOCK=true
poll: tokens [0, 1]
conn 0: read 6 byte(s), echo
guard: peer b got no echo within 1s (Err(WouldBlock)), exiting

3     epoll_ctl(8, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
3     epoll_ctl(8, EPOLL_CTL_ADD, 7, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
3     sendto(4, "from a", 6, MSG_NOSIGNAL, NULL, 0) = 6
3     sendto(6, "from b", 6, MSG_NOSIGNAL, NULL, 0) = 6
3     epoll_pwait(8, [{events=EPOLLIN, data=0}, {events=EPOLLIN, data=0x1}], 8, -1, NULL, 8) = 2
3     recvfrom(5, "from a", 64, 0, NULL, NULL) = 6
3     sendto(5, "from a", 6, MSG_NOSIGNAL, NULL, 0) = 6
3     recvfrom(5 <unfinished ...>
4     recvfrom(6, 0xffff81ebd8e0, 16, 0, NULL, NULL) = -1 EAGAIN (Resource temporarily unavailable)
4     exit_group(0 <unfinished ...>
3     <... recvfrom resumed> <unfinished ...>) = ?
4     <... exit_group resumed>)         = ?
4     +++ exited with 0 +++
3     +++ exited with 0 +++

    pub fn from_std(stream: net::TcpStream) -> TcpStream {
        TcpStream {
            inner: IoSource::new(stream),
        }
    }

let (std_stream, _) = std_listener.accept()?;
std_stream.set_nonblocking(true)?;
let stream = mio::net::TcpStream::from_std(std_stream);

pub(crate) fn check_socket_for_blocking<S: AsFd>(s: &S) -> crate::io::Result<()> {
    #[cfg(not(tokio_allow_from_blocking_fd))]
    {
        let sock = socket2::SockRef::from(s);

        debug_assert!(
            sock.nonblocking()?,
            "Registering a blocking socket with the tokio runtime is unsupported. \
            If you wish to do anyways, please add `--cfg tokio_allow_from_blocking_fd` to your \
            RUSTFLAGS. See github.com/tokio-rs/tokio/issues/7172 for details."
        );
    }

    Ok(())
}

use std::io::Write;
use std::time::Duration;

use tokio::io::AsyncReadExt;
use tokio::time::timeout;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  println!("os={} debug_assertions={}", std::env::consts::OS, cfg!(debug_assertions));
  let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
  let mut peer = std::net::TcpStream::connect(listener.local_addr().unwrap()).unwrap();
  let (s, _) = listener.accept().unwrap(); // forgot set_nonblocking
  peer.write_all(b"hello").unwrap();

  // guard: the runtime thread will be stuck in the kernel, so exit from another OS thread
  std::thread::spawn(|| {
    std::thread::sleep(Duration::from_secs(2));
    println!("guard: still blocked after 2s, exiting");
    std::process::exit(0);
  });

  let mut stream = tokio::net::TcpStream::from_std(s).unwrap();
  println!("from_std: Ok");
  // 64: the first read is short; 5: the first read fills the buffer exactly
  let len: usize = std::env::args().nth(1).map_or(64, |a| a.parse().unwrap());
  println!("buffer: {len} bytes");
  let mut buf = vec![0u8; len];
  for i in 1..=2 {
    match timeout(Duration::from_millis(500), stream.read(&mut buf)).await {
      Ok(r) => println!("read #{i}: {:?}", r.map_err(|e| e.kind())),
      Err(_) => println!("read #{i}: timed out after 500ms"),
    }
  }
}

os=macos debug_assertions=true

thread 'main' (2343511) panicked at src/bin/from_std_tokio.rs:22:20:
Registering a blocking socket with the tokio runtime is unsupported. If you wish to do anyways, please add `--cfg tokio_allow_from_blocking_fd` to your RUSTFLAGS. See github.com/tokio-rs/tokio/issues/7172 for details.
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

os=linux debug_assertions=false
from_std: Ok
buffer: 64 bytes
read #1: Ok(5)
read #2: timed out after 500ms
os=linux debug_assertions=false
from_std: Ok
buffer: 5 bytes
read #1: Ok(5)
guard: still blocked after 2s, exiting