Linux 与 macOS：生产和开发的差别

    pub(crate) fn mask(self) -> Ready {
        match self {
            Interest::READABLE => Ready::READABLE | Ready::READ_CLOSED,
            Interest::WRITABLE => Ready::WRITABLE | Ready::WRITE_CLOSED,
            #[cfg(any(target_os = "linux", target_os = "android"))]
            Interest::PRIORITY => Ready::PRIORITY | Ready::READ_CLOSED,
            Interest::ERROR => Ready::ERROR,
            _ => Ready::EMPTY,
        }
    }

    pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
        self.io
            .registration()
            .async_io(Interest::READABLE | Interest::ERROR, || self.io.recv(buf))
            .await
    }

// A connected UDP socket sends to a closed port; then a TCP connect to a closed port.
use std::time::Duration;

use tokio::time::timeout;

#[tokio::main(flavor = "current_thread")]
async fn main() -> std::io::Result<()> {
  // A port nobody listens on: bind, read the port, close.
  let udp_port = std::net::UdpSocket::bind("127.0.0.1:0")?.local_addr()?.port();
  let tcp_port = std::net::TcpListener::bind("127.0.0.1:0")?.local_addr()?.port();

  let sock = tokio::net::UdpSocket::bind("127.0.0.1:0").await?;
  sock.connect(("127.0.0.1", udp_port)).await?;
  sock.send(b"ping").await?;
  let mut buf = [0u8; 16];
  match timeout(Duration::from_millis(500), sock.recv(&mut buf)).await {
    Ok(r) => println!("os={} udp recv: {r:?}", std::env::consts::OS),
    Err(_) => println!("os={} udp recv: still pending after 500ms", std::env::consts::OS),
  }

  let r = tokio::net::TcpStream::connect(("127.0.0.1", tcp_port)).await;
  println!("os={} tcp connect: {:?}", std::env::consts::OS, r.map(|_| ()));
  Ok(())
}

os=macos udp recv: Err(Os { code: 61, kind: ConnectionRefused, message: "Connection refused" })
os=macos tcp connect: Err(Os { code: 61, kind: ConnectionRefused, message: "Connection refused" })
os=linux udp recv: Err(Os { code: 111, kind: ConnectionRefused, message: "Connection refused" })
os=linux tcp connect: Err(Os { code: 111, kind: ConnectionRefused, message: "Connection refused" })

os=macos udp recv: Err(Os { code: 61, kind: ConnectionRefused, message: "Connection refused" })
os=macos tcp connect: Err(Os { code: 61, kind: ConnectionRefused, message: "Connection refused" })
os=linux udp recv: still pending after 500ms
os=linux tcp connect: Err(Os { code: 111, kind: ConnectionRefused, message: "Connection refused" })

        pub(crate) async fn connect_mio(sys: mio::net::TcpStream) -> io::Result<TcpStream> {
            let stream = TcpStream::new(sys)?;

            // Once we've connected, wait for the stream to be writable as
            // that's when the actual connection has been initiated. Once we're
            // writable we check for `take_socket_error` to see if the connect
            // actually hit an error or not.
            //
            // If all that succeeded then we ship everything on up.
            poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;

            if let Some(e) = stream.io.take_error()? {
                return Err(e);
            }

            Ok(stream)
        }

        pub fn max_io_events_per_tick(&mut self, capacity: usize) -> &mut Self {
            self.nevents = capacity;
            self
        }

// tokio 1.52.3 accepts max_io_events_per_tick(0): mio::Events::with_capacity(0).
use std::time::{Duration, Instant};

fn main() {
  let rt = tokio::runtime::Builder::new_current_thread()
    .enable_all()
    .max_io_events_per_tick(0)
    .build()
    .unwrap();
  let start = Instant::now();
  rt.block_on(async { tokio::time::sleep(Duration::from_millis(100)).await });
  println!("os={} slept 100ms, returned after {:?}", std::env::consts::OS, start.elapsed());
}

os=macos slept 100ms, returned after 100.893166ms

thread 'main' (1) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.52.3/src/runtime/io/driver.rs:196:23:
unexpected error when polling the I/O driver: Os { code: 22, kind: InvalidInput, message: "Invalid argument" }
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

// Two AsyncFd values for the same fd number.
use std::os::fd::{AsRawFd, RawFd};

struct Fd(RawFd);
impl AsRawFd for Fd {
  fn as_raw_fd(&self) -> RawFd {
    self.0
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let sock = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
  sock.set_nonblocking(true).unwrap();
  let fd = sock.as_raw_fd();
  let a = tokio::io::unix::AsyncFd::new(Fd(fd));
  let b = tokio::io::unix::AsyncFd::new(Fd(fd));
  println!("os={} first: {:?}", std::env::consts::OS, a.as_ref().map(|_| ()));
  println!("os={} second: {:?}", std::env::consts::OS, b.as_ref().map(|_| ()));
}

os=macos first: Ok(())
os=macos second: Ok(())
os=linux first: Ok(())
os=linux second: Err(Os { code: 17, kind: AlreadyExists, message: "File exists" })

// macOS only: AsyncFd on a kqueue fd. Registering READABLE | WRITABLE fails
// (EVFILT_WRITE on a kqueue fd is EINVAL), but the EVFILT_READ entry stays.
use std::os::fd::{AsRawFd, RawFd};
use std::time::Duration;

struct Fd(RawFd);
impl AsRawFd for Fd {
  fn as_raw_fd(&self) -> RawFd {
    self.0
  }
}

fn main() {
  let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
  rt.block_on(async {
    let inner = unsafe { libc::kqueue() };
    let r = tokio::io::unix::AsyncFd::new(Fd(inner));
    println!("AsyncFd::new(kqueue fd {inner}) = {:?}", r.as_ref().map(|_| ()));
    #[cfg(tokio_unstable)]
    let m = tokio::runtime::Handle::current().metrics();
    #[cfg(tokio_unstable)]
    println!("fd_registered={} ready_events={}", m.io_driver_fd_registered_count(), m.io_driver_ready_count());

    // Make the inner kqueue readable: trigger an EVFILT_USER event in it.
    let mut kev: libc::kevent = unsafe { std::mem::zeroed() };
    kev.ident = 1;
    kev.filter = libc::EVFILT_USER;
    kev.flags = libc::EV_ADD | libc::EV_CLEAR;
    kev.fflags = libc::NOTE_TRIGGER;
    let n = unsafe { libc::kevent(inner, &kev, 1, std::ptr::null_mut(), 0, std::ptr::null()) };
    println!("trigger inner kqueue: kevent = {n}");

    // Let the driver poll a few times.
    for _ in 0..3 {
      tokio::time::sleep(Duration::from_millis(20)).await;
    }
    #[cfg(tokio_unstable)]
    println!("fd_registered={} ready_events={}", m.io_driver_fd_registered_count(), m.io_driver_ready_count());
  });
  println!("done");
}

AsyncFd::new(kqueue fd 9) = Err(Os { code: 22, kind: InvalidInput, message: "Invalid argument" })
fd_registered=0 ready_events=0
trigger inner kqueue: kevent = 0
fd_registered=0 ready_events=1
done

AsyncFd::new(kqueue fd 9) = Err(Os { code: 22, kind: InvalidInput, message: "Invalid argument" })
fd_registered=0 ready_events=0
trigger inner kqueue: kevent = 0
=================================================================
==37470==ERROR: AddressSanitizer: heap-use-after-free on address 0x613000003a90 at pc 0x000102dba478 bp 0x00016d195be0 sp 0x00016d195bd8
READ of size 8 at 0x613000003a90 thread T0

for f in ~/.target/dx/app/debug/web/server ~/.target/dx/app/release/web/server; do
  echo "$f: $(grep -c -a 'already registered with a' $f) $(grep -c -a 'Only a single' $f)"
done

~/.target/dx/app/debug/web/server: 1 1
~/.target/dx/app/release/web/server: 0 0