tokio 的 I/O 驱动怎么调 mio

    pub(crate) fn new(nevents: usize) -> io::Result<(Driver, Handle)> {
        let poll = mio::Poll::new()?;
        #[cfg(not(target_os = "wasi"))]
        let waker = mio::Waker::new(poll.registry(), TOKEN_WAKEUP)?;
        let registry = poll.registry().try_clone()?;

        let driver = Driver {
            signal_ready: false,
            events: mio::Events::with_capacity(nevents),
            poll,
        };

        let (registrations, synced) = RegistrationSet::new();

        let handle = Handle {
            registry,
            registrations,
            synced: Mutex::new(synced),
            #[cfg(not(target_os = "wasi"))]
            waker,
            metrics: IoDriverMetrics::default(),
            // ...
        };

        Ok((driver, handle))
    }

    pub(crate) fn new(io: E) -> io::Result<Self> {
        PollEvented::new_with_interest(io, Interest::READABLE | Interest::WRITABLE)
    }

    pub(super) fn add_source(
        &self,
        source: &mut impl mio::event::Source,
        interest: Interest,
    ) -> io::Result<Arc<ScheduledIo>> {
        let scheduled_io = self.registrations.allocate(&mut self.synced.lock())?;
        let token = scheduled_io.token();

        // we should remove the `scheduled_io` from the `registrations` set if registering
        // the `source` with the OS fails. Otherwise it will leak the `scheduled_io`.
        if let Err(e) = self.registry.register(source, token, interest.to_mio()) {
            // safety: `scheduled_io` is part of the `registrations` set.
            unsafe {
                self.registrations
                    .remove(&mut self.synced.lock(), &scheduled_io)
            };

            return Err(e);
        }

        // TODO: move this logic to `RegistrationSet` and use a `CountedLinkedList`
        self.metrics.incr_fd_count();

        Ok(scheduled_io)
    }

    fn turn(&mut self, handle: &Handle, max_wait: Option<Duration>) {
        debug_assert!(!handle.registrations.is_shutdown(&handle.synced.lock()));

        handle.release_pending_registrations();

        let events = &mut self.events;

        // Block waiting for an event to happen, peeling out how many events
        // happened.
        match self.poll.poll(events, max_wait) {
            Ok(()) => {}
            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
            // ...
            Err(e) => panic!("unexpected error when polling the I/O driver: {e:?}"),
        }

        // Process all the events that came in, dispatching appropriately
        let mut ready_count = 0;
        for event in events.iter() {
            let token = event.token();

            if token == TOKEN_WAKEUP {
                // Nothing to do, the event is used to unblock the I/O driver
            } else if token == TOKEN_SIGNAL {
                self.signal_ready = true;
            } else {
                let ready = Ready::from_mio(event);
                let ptr = super::EXPOSE_IO.from_exposed_addr(token.0);

                // Safety: we ensure that the pointers used as tokens are not freed
                // until they are both deregistered from mio **and** we know the I/O
                // driver is not concurrently polling. The I/O driver holds ownership of
                // an `Arc<ScheduledIo>` so we can safely cast this to a ref.
                let io: &ScheduledIo = unsafe { &*ptr };

                io.set_readiness(Tick::Set, |curr| curr | ready);
                io.wake(ready);

                ready_count += 1;
            }
        }
        // ...
    }

impl<E: Source> Drop for PollEvented<E> {
    fn drop(&mut self) {
        if let Some(mut io) = self.io.take() {
            // Ignore errors
            let _ = self.registration.deregister(&mut io);
        }
    }
}

    pub(super) fn deregister_source(
        &self,
        registration: &Arc<ScheduledIo>,
        source: &mut impl Source,
    ) -> io::Result<()> {
        // Deregister the source with the OS poller **first**
        // Cleanup ALWAYS happens
        let os_result = self.registry.deregister(source);

        if self
            .registrations
            .deregister(&mut self.synced.lock(), registration)
        {
            self.unpark();
        }

        self.metrics.dec_fd_count();

        os_result // Return error after cleanup
    }

// A tokio echo server plus a blocking std client thread, in one process.
// Usage: echo [connections]   (default 1)
use std::io::{Read, Write};

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

fn main() -> std::io::Result<()> {
  let conns: usize = std::env::args().nth(1).map_or(1, |s| s.parse().unwrap());
  // Like the site (dioxus-server's block_on): multi-thread runtime, enable_all.
  // One worker keeps the trace readable.
  let rt = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(1)
    .enable_all()
    .build()?;
  rt.block_on(async move {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let client = std::thread::spawn(move || -> std::io::Result<String> {
      let mut last = String::new();
      for _ in 0..conns {
        let mut s = std::net::TcpStream::connect(addr)?;
        s.write_all(b"hello")?;
        let mut buf = [0u8; 5];
        s.read_exact(&mut buf)?;
        s.shutdown(std::net::Shutdown::Write)?;
        let n = s.read(&mut buf)?;
        last = format!("echoed {:?}, then read {n}", String::from_utf8_lossy(&buf));
      }
      Ok(last)
    });
    let mut tasks = Vec::new();
    for _ in 0..conns {
      let (mut sock, _) = listener.accept().await?;
      tasks.push(tokio::spawn(async move {
        let mut buf = [0u8; 64];
        loop {
          let n = sock.read(&mut buf).await.unwrap();
          if n == 0 {
            break;
          }
          sock.write_all(&buf[..n]).await.unwrap();
        }
      }));
    }
    for t in tasks {
      t.await.unwrap();
    }
    let last = client.join().unwrap()?;
    println!("os={} conns={conns} last client: {last}", std::env::consts::OS);
    Ok::<_, std::io::Error>(())
  })?;
  #[cfg(tokio_unstable)]
  {
    let m = rt.metrics();
    println!(
      "io_driver: fd_registered={} fd_deregistered={} ready_events={}",
      m.io_driver_fd_registered_count(),
      m.io_driver_fd_deregistered_count(),
      m.io_driver_ready_count()
    );
  }
  Ok(())
}

tokio = { version = "=1.52.3", features = ["full"] }
mio = { version = "=1.2.1", features = ["os-poll", "net", "os-ext"] }

epoll_create1(EPOLL_CLOEXEC)            = 3
eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)   = 4
epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
fcntl(3, F_DUPFD_CLOEXEC, 3)            = 5
fcntl(5, F_SETFD, FD_CLOEXEC)           = 0
socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0, [6, 7]) = 0
fcntl(6, F_DUPFD_CLOEXEC, 3)            = 8
fcntl(8, F_SETFD, FD_CLOEXEC)           = 0
epoll_ctl(5, EPOLL_CTL_ADD, 8, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
strace: Process 4 attached
[pid     3] socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, IPPROTO_IP) = 9
[pid     3] bind(9, {sa_family=AF_INET, sin_port=htons(0), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
[pid     3] listen(9, 128)              = 0
[pid     3] epoll_ctl(5, EPOLL_CTL_ADD, 9, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0xffff9b9d8c00}) = 0
[pid     4] epoll_pwait(3strace: Process 5 attached
 <unfinished ...>
[pid     5] socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 10
[pid     5] connect(10, {sa_family=AF_INET, sin_port=htons(45053), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
[pid     4] <... epoll_pwait resumed>, [{events=EPOLLIN, data=0xffff9b9d8c00}], 1024, -1, NULL, 8) = 1
[pid     5] sendto(10, "hello", 5, MSG_NOSIGNAL, NULL, 0) = 5
[pid     3] accept4(9 <unfinished ...>
[pid     5] recvfrom(10 <unfinished ...>
[pid     4] epoll_pwait(3 <unfinished ...>
[pid     3] <... accept4 resumed>, {sa_family=AF_INET, sin_port=htons(48136), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 11
[pid     3] epoll_ctl(5, EPOLL_CTL_ADD, 11, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0xffff9b9d8d80}) = 0
[pid     4] <... epoll_pwait resumed>, [{events=EPOLLIN|EPOLLOUT, data=0xffff9b9d8d80}], 1024, -1, NULL, 8) = 1
[pid     3] write(4, "\1\0\0\0\0\0\0\0", 8 <unfinished ...>
[pid     4] recvfrom(11 <unfinished ...>
[pid     3] <... write resumed>)        = 8
[pid     4] <... recvfrom resumed>, "hello", 64, 0, NULL, NULL) = 5
[pid     4] sendto(11, "hello", 5, MSG_NOSIGNAL, NULL, 0 <unfinished ...>
[pid     5] <... recvfrom resumed>, "hello", 5, 0, NULL, NULL) = 5
[pid     4] <... sendto resumed>)       = 5
[pid     5] shutdown(10, SHUT_WR <unfinished ...>
[pid     4] epoll_pwait(3 <unfinished ...>
[pid     5] <... shutdown resumed>)     = 0
[pid     4] <... epoll_pwait resumed>, [{events=EPOLLIN, data=0}, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP, data=0xffff9b9d8d80}], 1024, -1, NULL, 8) = 2
[pid     5] recvfrom(10 <unfinished ...>
[pid     4] recvfrom(11, "", 64, 0, NULL, NULL) = 0
[pid     4] epoll_ctl(5, EPOLL_CTL_DEL, 11, NULL) = 0
[pid     4] close(11 <unfinished ...>
[pid     5] <... recvfrom resumed>, "", 5, 0, NULL, NULL) = 0
[pid     4] <... close resumed>)        = 0
[pid     5] close(10)                   = 0
[pid     4] epoll_pwait(3 <unfinished ...>
[pid     5] +++ exited with 0 +++
[pid     3] write(1, "os=linux conns=1 last client: ec"..., 58os=linux conns=1 last client: echoed "hello", then read 0
) = 58
[pid     3] epoll_ctl(5, EPOLL_CTL_DEL, 9, NULL) = 0
[pid     3] close(9)                    = 0
[pid     3] write(4, "\1\0\0\0\0\0\0\0", 8) = 8
[pid     4] <... epoll_pwait resumed>, [{events=EPOLLIN, data=0}], 1024, -1, NULL, 8) = 1
[pid     4] close(3)                    = 0
[pid     4] close(8)                    = 0
[pid     4] close(5)                    = 0
[pid     4] close(4)                    = 0
[pid     4] +++ exited with 0 +++
+++ exited with 0 +++

os=linux conns=100 last client: echoed "hello", then read 0
 48.76    0.008801          22       400           recvfrom
 25.93    0.004680          15       301           epoll_pwait
 11.02    0.001989           9       200           sendto
  8.64    0.001560           7       199        99 accept4
  5.64    0.001018           4       204           epoll_ctl

os=macos conns=1 last client: echoed "hello", then read 0
io_driver: fd_registered=2 fd_deregistered=2 ready_events=5
os=macos conns=100 last client: echoed "hello", then read 0
io_driver: fd_registered=101 fd_deregistered=101 ready_events=500
os=linux conns=1 last client: echoed "hello", then read 0
io_driver: fd_registered=2 fd_deregistered=2 ready_events=3
os=linux conns=100 last client: echoed "hello", then read 0
io_driver: fd_registered=101 fd_deregistered=101 ready_events=300

// The echo connection from echo.rs, replayed on mio 1.2.1 directly (the way
// tokio registers: listener and stream both READABLE | WRITABLE), printing
// every event of every poll.
use std::io::{Read, Write};
use std::time::Duration;

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

const RW: Interest = Interest::READABLE.add(Interest::WRITABLE);

fn poll_once(poll: &mut Poll, events: &mut Events, step: &str) {
  poll.poll(events, Some(Duration::from_millis(100))).unwrap();
  let list: Vec<String> = events
    .iter()
    .map(|e| {
      let mut f = Vec::new();
      if e.is_readable() { f.push("R"); }
      if e.is_writable() { f.push("W"); }
      if e.is_read_closed() { f.push("RC"); }
      if e.is_write_closed() { f.push("WC"); }
      if e.is_error() { f.push("ERR"); }
      format!("{}:{}", e.token().0, f.join("|"))
    })
    .collect();
  println!("  {step:<28} {} event(s) {list:?}", list.len());
}

fn main() {
  println!("os={}", std::env::consts::OS);
  let mut poll = Poll::new().unwrap();
  let mut events = Events::with_capacity(1024);
  let mut l = TcpListener::bind("127.0.0.1:0".parse().unwrap()).unwrap();
  poll.registry().register(&mut l, Token(10), RW).unwrap();
  let mut c = std::net::TcpStream::connect(l.local_addr().unwrap()).unwrap();
  c.write_all(b"hello").unwrap();
  poll_once(&mut poll, &mut events, "connect + client writes");
  let (mut s, _) = l.accept().unwrap();
  poll.registry().register(&mut s, Token(11), RW).unwrap();
  poll_once(&mut poll, &mut events, "accepted, registered");
  let mut buf = [0u8; 64];
  let n = s.read(&mut buf).unwrap();
  s.write_all(&buf[..n]).unwrap();
  let mut back = [0u8; 5];
  c.read_exact(&mut back).unwrap();
  poll_once(&mut poll, &mut events, "echoed, client read it");
  c.shutdown(std::net::Shutdown::Write).unwrap();
  poll_once(&mut poll, &mut events, "client shutdown(Write)");
  println!("  read after FIN: {:?}", s.read(&mut buf).map_err(|e| e.kind()));
}

os=macos
  connect + client writes      1 event(s) ["10:R"]
  accepted, registered         2 event(s) ["11:W", "11:R"]
  echoed, client read it       1 event(s) ["11:W"]
  client shutdown(Write)       1 event(s) ["11:R|RC"]
  read after FIN: Ok(0)
os=linux
  connect + client writes      1 event(s) ["10:R"]
  accepted, registered         1 event(s) ["11:R|W"]
  echoed, client read it       0 event(s) []
  client shutdown(Write)       1 event(s) ["11:R|W|RC"]
  read after FIN: Ok(0)

[pid     3] accept4(9 <unfinished ...>
[pid     4] sendto(10, "hello", 5, MSG_NOSIGNAL, NULL, 0 <unfinished ...>
[pid     3] <... accept4 resumed>, {sa_family=AF_INET, sin_port=htons(44108), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 11
[pid     3] epoll_ctl(5, EPOLL_CTL_ADD, 11, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0xffffb3fd8c00} <unfinished ...>
[pid     4] <... sendto resumed>)       = 5
[pid     3] <... epoll_ctl resumed>)    = 0
[pid     4] recvfrom(10 <unfinished ...>
[pid     3] epoll_pwait(3, [{events=EPOLLIN|EPOLLOUT, data=0xffffb3fd8c00}], 1024, -1, NULL, 8) = 1
[pid     3] recvfrom(11, "hello", 64, 0, NULL, NULL) = 5
[pid     3] sendto(11, "hello", 5, MSG_NOSIGNAL, NULL, 0) = 5
[pid     4] <... recvfrom resumed>, "hello", 5, 0, NULL, NULL) = 5
[pid     3] epoll_pwait(3, [{events=EPOLLIN|EPOLLOUT|EPOLLRDHUP, data=0xffffb3fd8c00}], 1024, -1, NULL, 8) = 1
[pid     3] recvfrom(11 <unfinished ...>
[pid     4] recvfrom(10 <unfinished ...>
[pid     3] <... recvfrom resumed>, "", 64, 0, NULL, NULL) = 0
[pid     3] epoll_ctl(5, EPOLL_CTL_DEL, 11, NULL) = 0
[pid     3] close(11)                   = 0

[pid     3] accept4(9 <unfinished ...>
[pid     4] sendto(10, "hello", 5, MSG_NOSIGNAL, NULL, 0 <unfinished ...>
[pid     3] <... accept4 resumed>, {sa_family=AF_INET, sin_port=htons(53254), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 11
[pid     4] <... sendto resumed>)       = 5
[pid     3] epoll_ctl(5, EPOLL_CTL_ADD, 11, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0xffffa9419b80} <unfinished ...>
[pid     4] recvfrom(10 <unfinished ...>
[pid     3] <... epoll_ctl resumed>)    = 0
[pid     3] recvfrom(11, "hello", 64, 0, NULL, NULL) = 5
[pid     3] sendto(11, "hello", 5, MSG_NOSIGNAL, NULL, 0) = 5
[pid     4] <... recvfrom resumed>, "hello", 5, 0, NULL, NULL) = 5
[pid     3] epoll_pwait(3, [{events=EPOLLOUT, data=0xffffa9419b80}], 1024, -1, NULL, 8) = 1
[pid     3] epoll_pwait(3, [{events=EPOLLIN|EPOLLOUT|EPOLLRDHUP, data=0xffffa9419b80}], 1024, -1, NULL, 8) = 1
[pid     4] recvfrom(10 <unfinished ...>
[pid     3] recvfrom(11, "", 64, 0, NULL, NULL) = 0
[pid     3] epoll_ctl(5, EPOLL_CTL_DEL, 11, NULL) = 0
[pid     3] close(11)                   = 0

echo_ct152 1: accept4 epoll_ctl epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_ct152 2: accept4 epoll_ctl epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_ct152 3: accept4 epoll_ctl epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_ct154 1: accept4 epoll_ctl recvfrom sendto epoll_pwait epoll_pwait recvfrom epoll_ctl close 
echo_ct154 2: accept4 epoll_ctl recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_ct154 3: accept4 epoll_ctl recvfrom sendto epoll_pwait epoll_pwait recvfrom epoll_ctl close 

echo_late152 1: accept4 epoll_ctl epoll_pwait epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_late154 1: accept4 epoll_ctl recvfrom epoll_pwait epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_late152 2: accept4 epoll_ctl epoll_pwait epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_late154 2: accept4 epoll_ctl recvfrom epoll_pwait epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_late152 3: accept4 epoll_ctl epoll_pwait epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 
echo_late154 3: accept4 epoll_ctl recvfrom epoll_pwait epoll_pwait recvfrom sendto epoll_pwait recvfrom epoll_ctl close 

[pid     3] accept4(9, {sa_family=AF_INET, sin_port=htons(41144), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 11
[pid     3] epoll_ctl(5, EPOLL_CTL_ADD, 11, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0xffffa6e43b80}) = 0
[pid     3] recvfrom(11, 0xffffa6e43d58, 64, 0, NULL, NULL) = -1 EAGAIN (Resource temporarily unavailable)
[pid     3] epoll_pwait(3, [{events=EPOLLOUT, data=0xffffa6e43b80}], 1024, -1, NULL, 8) = 1
[pid     3] epoll_pwait(3 <unfinished ...>
[pid     3] <... epoll_pwait resumed>, [{events=EPOLLIN|EPOLLOUT, data=0xffffa6e43b80}], 1024, -1, NULL, 8) = 1
[pid     3] recvfrom(11 <unfinished ...>
[pid     3] <... recvfrom resumed>, "hello", 64, 0, NULL, NULL) = 5
[pid     3] sendto(11, "hello", 5, MSG_NOSIGNAL, NULL, 0) = 5
[pid     3] epoll_pwait(3, [{events=EPOLLIN|EPOLLOUT|EPOLLRDHUP, data=0xffffa6e43b80}], 1024, -1, NULL, 8) = 1
[pid     3] recvfrom(11, "", 64, 0, NULL, NULL) = 0
[pid     3] epoll_ctl(5, EPOLL_CTL_DEL, 11, NULL) = 0
[pid     3] close(11)                   = 0