Rust in Everything
基于 mio 1.2.4

一个事件循环

主线位置：两条主线从头到尾各走一遍。注册主线：register 落到 epoll_ctl(EPOLL_CTL_ADD)，deregister 落到 EPOLL_CTL_DEL；事件主线：poll 落到 epoll_pwait，事件里的 data 就是 Token，用户读写到 EAGAIN 再回去 poll。Waker 是事件主线的旁路，在 Linux 上是一次 eventfd 写。每一步背后的机制在 02、03 两部分，本页只看它们在系统调用层面长什么样。

sequenceDiagram
  participant C as 客户端线程
  participant S as 服务器线程
  participant K as 内核
  S->>K: 注册监听 fd 和 eventfd
  S->>K: epoll_pwait 阻塞
  C->>K: connect 两次
  K-->>S: data=0 可读
  S->>K: accept4 两次，第三次 EAGAIN
  S->>K: 注册 data=2、data=3
  C->>K: 发 19 字节
  K-->>S: data=2 可读
  S->>K: 读到 EAGAIN，写回 19 字节
  C->>K: shutdown 写方向
  K-->>S: data=2 可读加 RDHUP
  S->>K: 读到 0，DEL，close
  C->>K: eventfd 写 8 字节
  K-->>S: data=1 可读
  S->>K: DEL、close 其余 fd

上一页mio 解决什么问题讲了 mio 为什么只给「就绪加边沿触发」。这一页写一个最小的 echo 服务器，把这套语义从用户代码一直跟到系统调用。

程序

一个二进制里两个线程：主线程是 mio 的事件循环，另一个线程用阻塞的 std socket 当客户端，跑完自己退出，不需要外部的 nc。依赖只有一行：

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

use std::collections::HashMap;
use std::io::{self, ErrorKind, Read, Write};
use std::net::SocketAddr;
use std::sync::Arc;
use std::thread;

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

const LISTENER: Token = Token(0);
const WAKER: Token = Token(1);

struct Conn {
  stream: TcpStream,
  pending: Vec<u8>, // 读到了、还没写回去的字节
  want_write: bool, // 当前注册里有没有 WRITABLE
}

fn main() -> io::Result<()> {
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(128);

  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
  poll.registry().register(&mut listener, LISTENER, Interest::READABLE)?;
  let waker = Arc::new(Waker::new(poll.registry(), WAKER)?);

  let addr = listener.local_addr()?;
  let client = {
    let waker = Arc::clone(&waker);
    thread::spawn(move || run_client(addr, &waker))
  };

  let mut conns: HashMap<Token, Conn> = HashMap::new();
  let mut next_token = 2;
  let mut running = true;

  while running {
    poll.poll(&mut events, None)?;
    println!("[server] poll returned {} event(s)", events.iter().count());

    for event in events.iter() {
      match event.token() {
        LISTENER => loop {
          // 边沿触发：一直 accept 到 WouldBlock
          match listener.accept() {
            Ok((mut stream, peer)) => {
              let token = Token(next_token);
              next_token += 1;
              poll.registry().register(&mut stream, token, Interest::READABLE)?;
              println!("[server] accepted {peer} as {token:?}");
              conns.insert(token, Conn { stream, pending: Vec::new(), want_write: false });
            }
            Err(e) if e.kind() == ErrorKind::WouldBlock => break,
            Err(e) => return Err(e),
          }
        },
        WAKER => {
          println!("[server] woken, shutting down");
          running = false;
        }
        token => {
          let Some(conn) = conns.get_mut(&token) else { continue };
          let closed = if event.is_readable() { read_all(conn)? } else { false };
          let blocked = flush(conn)?;
          println!(
            "[server] {token:?} readable={} writable={} read_closed={} closed={closed}",
            event.is_readable(),
            event.is_writable(),
            event.is_read_closed(),
          );
          if closed {
            let mut conn = conns.remove(&token).unwrap();
            poll.registry().deregister(&mut conn.stream)?;
            println!("[server] {token:?} deregistered");
            // conn 在这里 drop，fd 被 close
          } else if blocked != conn.want_write {
            // 写不完才关心可写；写完了只关心可读
            let interest = if blocked { Interest::READABLE | Interest::WRITABLE } else { Interest::READABLE };
            poll.registry().reregister(&mut conn.stream, token, interest)?;
            conn.want_write = blocked;
          }
        }
      }
    }
  }

  let mut rest: Vec<_> = conns.into_iter().collect();
  rest.sort_by_key(|(token, _)| *token);
  for (token, mut conn) in rest {
    poll.registry().deregister(&mut conn.stream)?;
    println!("[server] {token:?} deregistered at shutdown");
  }
  poll.registry().deregister(&mut listener)?;
  drop(listener);

  let remaining = client.join().unwrap()?;
  drop(remaining);
  println!("[server] done");
  Ok(())
}

/// 读到 WouldBlock 为止；返回对端是否已关闭。
fn read_all(conn: &mut Conn) -> io::Result<bool> {
  let mut buf = [0u8; 4096];
  loop {
    match conn.stream.read(&mut buf) {
      Ok(0) => return Ok(true),
      Ok(n) => conn.pending.extend_from_slice(&buf[..n]),
      Err(e) if e.kind() == ErrorKind::WouldBlock => return Ok(false),
      Err(e) if e.kind() == ErrorKind::Interrupted => continue,
      Err(e) => return Err(e),
    }
  }
}

/// 把 pending 写回去；返回是否因 WouldBlock 没写完。
fn flush(conn: &mut Conn) -> io::Result<bool> {
  while !conn.pending.is_empty() {
    match conn.stream.write(&conn.pending) {
      Ok(n) => {
        conn.pending.drain(..n);
      }
      Err(e) if e.kind() == ErrorKind::WouldBlock => return Ok(true),
      Err(e) if e.kind() == ErrorKind::Interrupted => continue,
      Err(e) => return Err(e),
    }
  }
  Ok(false)
}

/// 客户端线程：用阻塞的 std socket，连两条，各发一句、收回显，
/// 半关闭第一条并等服务器关掉它，然后唤醒服务器；
/// 第二条交回主线程，等服务器关完再关。
fn run_client(addr: SocketAddr, waker: &Waker) -> io::Result<std::net::TcpStream> {
  let mut c0 = std::net::TcpStream::connect(addr)?;
  let mut c1 = std::net::TcpStream::connect(addr)?;
  for (i, c) in [&mut c0, &mut c1].into_iter().enumerate() {
    let msg = format!("hello from client {i}");
    c.write_all(msg.as_bytes())?;
    let mut buf = vec![0u8; msg.len()];
    c.read_exact(&mut buf)?;
    println!("[client] got back {:?}", String::from_utf8_lossy(&buf));
  }
  // 半关闭 c0，等服务器那一侧读到 EOF、关掉连接，这边再读到 0
  c0.shutdown(std::net::Shutdown::Write)?;
  let n = c0.read(&mut [0u8; 16])?;
  println!("[client] c0 read {n} byte(s): server closed it; waking server");
  waker.wake()?;
  Ok(c1)
}

几处安排：

  • Token 由用户分配。 Token(0) 给监听 socket，Token(1) 给 Waker，连接从 2 开始递增。mio 不检查 token 是否重复，也不保存 token 到 socket 的映射，所以要自己维护一个 HashMap<Token, Conn>，事件来了按 token 查。
  • 每个事件都处理到 WouldBlock。 监听 socket 上 accept 到 WouldBlock，连接上 read_all 读到 WouldBlock 或读到 0。这是上一页引的「Draining readiness」约定。
  • 缓冲是用户的。 mio 的 write 写不完时只返回写了多少或 WouldBlock，剩下的字节存在 pending 里；只有这时才 reregister 加上 WRITABLE，写完再改回只关心可读。这个程序里每次只回 19 字节，写从来不会阻塞，下面的 strace 里也就没有 EPOLL_CTL_MOD。
  • Waker 放进 Arc。 Waker 是 Send + Sync 的，客户端线程拿一份引用，wake() 之后服务器的 poll 返回一个 Token(1) 的事件，主循环据此退出。
  • 客户端的顺序是确定的。 客户端每发一句都等回显，第一条连接用 shutdown(Write) 半关闭，再等服务器关掉它（读到 0）之后才 wake()。这样服务器一定先处理完 EOF 再处理唤醒，每次运行的事件序列都一样。

跑起来

macOS（M2 Max、macOS、rustc 1.99.0），cargo run --release --bin echo：

[server] poll returned 1 event(s)
[server] accepted 127.0.0.1:56781 as Token(2)
[server] accepted 127.0.0.1:56782 as Token(3)
[server] poll returned 1 event(s)
[server] Token(2) readable=true writable=false read_closed=false closed=false
[client] got back "hello from client 0"
[server] poll returned 1 event(s)
[server] Token(3) readable=true writable=false read_closed=false closed=false
[client] got back "hello from client 1"
[server] poll returned 1 event(s)
[server] Token(2) readable=true writable=false read_closed=true closed=true
[server] Token(2) deregistered
[client] c0 read 0 byte(s): server closed it; waking server
[server] poll returned 1 event(s)
[server] woken, shutting down
[server] Token(3) deregistered at shutdown
[server] done

Linux（Linux 6.12.28（Apple container 里的 Alpine，aarch64 musl 静态链接）、rustc 1.99.0），同一份源码交叉编译成 aarch64-unknown-linux-musl，在容器里跑：

[server] poll returned 1 event(s)
[server] accepted 127.0.0.1:33044 as Token(2)
[server] accepted 127.0.0.1:33050 as Token(3)
[server] poll returned 1 event(s)
[server] Token(2) readable=true writable=false read_closed=false closed=false
[client] got back "hello from client 0"
[server] poll returned 1 event(s)
[server] Token(3) readable=true writable=false read_closed=false closed=false
[client] got back "hello from client 1"
[server] poll returned 1 event(s)
[server] Token(2) readable=true writable=false read_closed=true closed=true
[server] Token(2) deregistered
[client] c0 read 0 byte(s): server closed it; waking server
[server] poll returned 1 event(s)
[server] woken, shutting down
[server] Token(3) deregistered at shutdown
[server] done

两个平台各跑了 3 次，除了客户端的端口号，6 次输出逐行相同。同一份程序在 kqueue 和 epoll 上得到同样的五次 poll、同样的事件。

第一次 poll 只返回 1 个事件，却 accept 了两个连接：两次 connect 在服务器醒来之前都完成了，监听 socket 只有一次「从不可读变成可读」，边沿触发只报一次。如果服务器在这个事件上只 accept 一次，第二个连接就会一直留在队列里，直到又有新连接进来。

Linux 上的系统调用

用 strace 跑同一个二进制，每个线程单独一个输出文件（-ff -o），这样两个线程的调用不会交错成 <unfinished ...> / <... resumed>：

strace -ff -o /w/tr/r1 \
  -e trace=epoll_create1,epoll_ctl,epoll_pwait,epoll_wait,eventfd2,accept4,socket,bind,listen,connect,read,write,close,setsockopt,shutdown,recvfrom,sendto \
  /w/echo

跑了 3 次。除了端口号和栈上缓冲区的地址，3 次的服务器线程文件逐行相同，客户端线程文件也逐行相同。下面是第 1 次。write(1, ...) 是程序自己的 println!，留着当路标。

服务器线程（主线程）的文件一共 49 行，下面分五段贴出，合起来就是全文。

建 Poll、监听 socket 和 Waker

epoll_create1(EPOLL_CLOEXEC)            = 3
socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, IPPROTO_IP) = 4
setsockopt(4, SOL_SOCKET, SO_REUSEADDR, [1], 4) = 0
bind(4, {sa_family=AF_INET, sin_port=htons(0), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
listen(4, 128)                          = 0
epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)   = 5
epoll_ctl(3, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
  • 第 1 行是 Poll::new()。Linux 上 Selector::new 只有一个系统调用 epoll_create1(libc::EPOLL_CLOEXEC)，得到 fd 3。Poll 里的 Registry 里的 Selector 就是这个 fd。
  • 第 2 到 5 行是 TcpListener::bind。bind 的文档列了四步：建 socket、设 SO_REUSEADDR、bind、listen，四个系统调用一一对应。socket 的类型里直接带了 SOCK_NONBLOCK|SOCK_CLOEXEC（new_socket 在 Linux 上这样做），不需要事后 fcntl。backlog 128 和 std 相同。细节在 TCP。
  • 第 6 行是 register(&mut listener, Token(0), Interest::READABLE)，注册主线的终点。Registry::register 转给 IoSource 的 register（debug 下先查 SelectorId），再转给 Selector::register，一次 epoll_ctl(EPOLL_CTL_ADD)。events=EPOLLIN|EPOLLRDHUP|EPOLLET 是 interests_to_epoll 对 READABLE 的翻译，data=0 就是 Token(0)，原样存进 epoll_event 的 u64。这条路上每一层做了什么，见 Poll 与 Registry 和 Source 与 IoSource。
  • 第 7、8 行是 Waker::new(registry, Token(1))。Linux 上的 Waker::new 先 eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK)（libc 的 eventfd 在内核里是 eventfd2），再把它当作一个普通 fd 以 Interest::READABLE 注册进同一个 epoll，所以它的标志和 socket 一样，也带着 EPOLLRDHUP，data=0x1 是 Token(1)。

epoll 版的 register 源码里只有一个 syscall!，mio 也没有在用户态记下任何「注册表」。

第一次 poll：一个事件，两个连接

epoll_pwait(3, [{events=EPOLLIN, data=0}], 128, -1, NULL, 8) = 1
write(1, "[server] poll returned 1 event(s"..., 34) = 34
accept4(4, {sa_family=AF_INET, sin_port=htons(46582), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 8
epoll_ctl(3, EPOLL_CTL_ADD, 8, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x2}) = 0
write(1, "[server] accepted 127.0.0.1:4658"..., 46) = 46
accept4(4, {sa_family=AF_INET, sin_port=htons(46584), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 9
epoll_ctl(3, EPOLL_CTL_ADD, 9, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x3}) = 0
write(1, "[server] accepted 127.0.0.1:4658"..., 46) = 46
accept4(4, 0xffffee3b81c8, [128], SOCK_CLOEXEC|SOCK_NONBLOCK) = -1 EAGAIN (Resource temporarily unavailable)
  • 第 9 行是 poll.poll(&mut events, None)。源码里 Selector::select 调的是 libc::epoll_wait，strace 看到的却是 epoll_pwait(..., NULL, 8)：aarch64 的 Linux 没有 epoll_wait 这个系统调用（libc 的 aarch64 常量表里只有 SYS_epoll_pwait），musl 用信号掩码为 NULL 的 epoll_pwait 实现它，最后的 8 是信号集的字节数。参数 128 是 Events::with_capacity(128) 的容量，-1 是 timeout = None（select 里的 .unwrap_or(-1)）。返回 1，事件里 data=0，event.token() 就是把它原样读回来，见 Token、Interest 与 Event。
  • 第 11 到 17 行是 LISTENER 分支里的 accept 循环：两次 accept4 成功（fd 8、9），每次之后紧跟一次 EPOLL_CTL_ADD，第三次 EAGAIN，循环退出。accept4 的最后一个参数 SOCK_CLOEXEC|SOCK_NONBLOCK 来自 accept，新连接一出生就是非阻塞的。
📝NOTE
冷知识：Linux 上 accept 出来的 fd 不继承非阻塞

accept 有两条路：Linux、FreeBSD 等用 accept4(..., SOCK_CLOEXEC | SOCK_NONBLOCK)；macOS 等没有 accept4 的平台用 accept，注释说「Luckily BSD (derived) OSs inherit the non-blocking flag from the listener, so we just have to set CLOEXEC」。x86 Android 是例外，它走 accept 那条路，还要多一次 fcntl(F_SETFL, O_NONBLOCK)，代码旁边挂着两个 issue。

#1445（issue）报告 x86 Android 上程序被 seccomp 杀掉：「the seccomp profile forbids syscall 364 which maps to accept4」。修复把 x86 Android 挪到 accept 那条路上。挪过去之后，#1450（issue）发现这条路假设新 fd 会继承监听 socket 的阻塞属性：「This... seems to not be the case on Android, as after collecting a few strace outputs, I discovered that recvfrom() calls were blocking where they very much shouldn't have been」。于是加了 #[cfg] 里那次 F_SETFL。两处修复都在 0.7.9 发布，CHANGELOG 0.7.9：「Use accept(2) on x86 Android instead of accept4(2)」。

Android 用的是 Linux 内核，Linux 的 accept 不把监听 socket 的 O_NONBLOCK 带给新 fd。上面第 11、14 行的 SOCK_NONBLOCK 就是在 accept4 这一次调用里把这件事做掉。

读到 EAGAIN，写回去

epoll_pwait(3, [{events=EPOLLIN, data=0x2}], 128, -1, NULL, 8) = 1
write(1, "[server] poll returned 1 event(s"..., 34) = 34
recvfrom(8, "hello from client 0", 4096, 0, NULL, NULL) = 19
recvfrom(8, 0xffffee3b7200, 4096, 0, NULL, NULL) = -1 EAGAIN (Resource temporarily unavailable)
sendto(8, "hello from client 0", 19, MSG_NOSIGNAL, NULL, 0) = 19
write(1, "[server] Token(2) readable=true "..., 78) = 78
epoll_pwait(3, [{events=EPOLLIN, data=0x3}], 128, -1, NULL, 8) = 1
write(1, "[server] poll returned 1 event(s"..., 34) = 34
recvfrom(9, "hello from client 1", 4096, 0, NULL, NULL) = 19
recvfrom(9, 0xffffee3b7200, 4096, 0, NULL, NULL) = -1 EAGAIN (Resource temporarily unavailable)
sendto(9, "hello from client 1", 19, MSG_NOSIGNAL, NULL, 0) = 19
write(1, "[server] Token(3) readable=true "..., 78) = 78
  • 第 18 行：Token(2) 可读。events=EPOLLIN，没有 EPOLLOUT，因为注册时只要了 READABLE。
  • 第 20、21 行是 read_all：第一次 recvfrom 读到 19 字节，第二次 EAGAIN，循环退出。mio 的 TcpStream::read 是 do_io(|inner| inner.read(buf))，epoll 上 do_io 什么也不加，直接调 std 的 read；std 在 Linux 上用 recv，aarch64 上 recv 在内核里是 recvfrom。
  • 第 22 行是 flush：一次 sendto 写完 19 字节（MSG_NOSIGNAL 是 std 加的）。没写不完，就没有 reregister，也就没有 EPOLL_CTL_MOD。
  • 第 24 到 29 行是 Token(3) 上同样的一轮。

两个连接的事件分在两次 poll 里，是因为客户端每发一句都等回显，第二句发出时第一轮已经处理完了。

对端半关闭：RDHUP、读到 0、deregister

epoll_pwait(3, [{events=EPOLLIN|EPOLLRDHUP, data=0x2}], 128, -1, NULL, 8) = 1
write(1, "[server] poll returned 1 event(s"..., 34) = 34
recvfrom(8, "", 4096, 0, NULL, NULL)    = 0
write(1, "[server] Token(2) readable=true "..., 76) = 76
epoll_ctl(3, EPOLL_CTL_DEL, 8, NULL)    = 0
write(1, "[server] Token(2) deregistered\n", 31) = 31
close(8)                                = 0
  • 第 30 行：客户端 shutdown(SHUT_WR) 之后，Token(2) 的事件是 EPOLLIN|EPOLLRDHUP。EPOLLIN 让 is_readable() 为真（读到 EOF 也算可读），EPOLLRDHUP 让 is_read_closed() 为真，对应输出里的 read_closed=true。注册时默认带 EPOLLRDHUP，就是为了能报告这一位，见 epoll。
  • 第 32 行：recvfrom 返回 0，read_all 返回 closed = true。
  • 第 34 行是 deregister：Selector::deregister 是 epoll_ctl(EPOLL_CTL_DEL, fd, NULL)。
  • 第 36 行是 conn 被 drop：TcpStream 里的 std socket 关闭 fd 8。epoll 和 kqueue 上 mio 的 socket 类型没有自己的 Drop，drop 只是关 fd，不会顺手 deregister。
📝NOTE
没有蠢问题

问： Linux 上 fd 关闭时 epoll 会自动把它从关注集合里删掉，第 34 行那次 EPOLL_CTL_DEL 是不是多余的？

答： 在这个程序里删不删结果一样，但不能依赖它。epoll 跟踪的是 fd 背后那个「打开的文件」，只有指向它的所有 fd 都关了才会自动移除（epoll(7) 的 Q6）；fd 被 dup 过、或者 fork 出去的子进程还拿着，close 之后事件照样会来，带着一个已经被你复用给别的连接的 token。Registry::deregister 的文档给的保证是：显式调用之后，只要和之后的 poll 之间有 happens-before 关系，这个 token 不会再出现。这条保证只对显式 deregister 成立。别的平台上不 deregister 会怎样，见 Poll 与 Registry。

被 Waker 叫醒，收尾

epoll_pwait(3, [{events=EPOLLIN, data=0x1}], 128, -1, NULL, 8) = 1
write(1, "[server] poll returned 1 event(s"..., 34) = 34
write(1, "[server] woken, shutting down\n", 30) = 30
epoll_ctl(3, EPOLL_CTL_DEL, 9, NULL)    = 0
write(1, "[server] Token(3) deregistered a"..., 43) = 43
close(9)                                = 0
epoll_ctl(3, EPOLL_CTL_DEL, 4, NULL)    = 0
close(4)                                = 0
close(7)                                = 0
write(1, "[server] done\n", 14)         = 14
close(5)                                = 0
close(3)                                = 0
+++ exited with 0 +++
  • 第 37 行：epoll_pwait 返回 data=0x1，即 Token(1)。这是客户端线程往 eventfd 写了 8 字节（见下一节）之后，eventfd 变成可读，epoll 报告出来的。服务器从头到尾没有读过这个 eventfd，计数器一直停在 1。多次 wake()、计数器溢出时怎样，见 Waker。
  • 第 40 到 44 行：退出循环后，剩下的 Token(3) 先 EPOLL_CTL_DEL 再关 fd 9；监听 socket 先 EPOLL_CTL_DEL 再关 fd 4。
  • 第 45 行 close(7) 是客户端交回来的第二条连接（客户端那一侧的 fd），在 join 之后 drop。
  • 第 47、48 行是 main 返回时局部变量按声明的逆序 drop：先是 waker（Arc 的最后一份，关 eventfd 5），最后是 poll（关 epoll fd 3）。

客户端线程

客户端线程的文件，16 行全文：

socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 6
connect(6, {sa_family=AF_INET, sin_port=htons(42171), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 7
connect(7, {sa_family=AF_INET, sin_port=htons(42171), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
sendto(6, "hello from client 0", 19, MSG_NOSIGNAL, NULL, 0) = 19
recvfrom(6, "hello from client 0", 19, 0, NULL, NULL) = 19
write(1, "[client] got back \"hello from cl"..., 40) = 40
sendto(7, "hello from client 1", 19, MSG_NOSIGNAL, NULL, 0) = 19
recvfrom(7, "hello from client 1", 19, 0, NULL, NULL) = 19
write(1, "[client] got back \"hello from cl"..., 40) = 40
shutdown(6, SHUT_WR)                    = 0
recvfrom(6, "", 16, 0, NULL, NULL)      = 0
write(1, "[client] c0 read 0 byte(s): serv"..., 60) = 60
write(5, "\1\0\0\0\0\0\0\0", 8)         = 8
close(6)                                = 0
+++ exited with 0 +++
  • socket 只带 SOCK_CLOEXEC，没有 SOCK_NONBLOCK：这是 std 的阻塞 socket，connect、recvfrom 都在内核里等。
  • shutdown(6, SHUT_WR) 之后，recvfrom 返回 0：服务器那一侧关掉了 fd 8。
  • write(5, "\1\0\0\0\0\0\0\0", 8) 就是 waker.wake()。wake 把 1u64.to_ne_bytes() 写进 eventfd，aarch64 是小端，所以是一个 1 加七个 0。fd 5 是主线程建的那个 eventfd，同一个进程里的线程共享 fd 表。
  • close(6) 是 c0 在函数返回时 drop。c1（fd 7）被返回给主线程，所以这里没有 close(7)。

用 strace -f（不分文件）再跑 3 次，看两个线程的先后：每次都是服务器的 close(8) 在客户端那次 recvfrom 返回 0 之前，客户端的 write(5, ...) 在服务器第五次 epoll_pwait 返回之前。开头的时序图就是按这个顺序画的。

对照两条主线

用户代码系统调用（Linux）主线机制在哪一页
Poll::new()epoll_create1(EPOLL_CLOEXEC)—Poll 与 Registry
TcpListener::bindsocket、setsockopt、bind、listen—TCP
register(.., Token(n), READABLE)epoll_ctl(ADD, fd, {EPOLLIN|EPOLLRDHUP|EPOLLET, data=n})注册epoll
Waker::neweventfd2 加一次 epoll_ctl(ADD)注册Waker
poll.poll(&mut events, None)epoll_pwait(epfd, .., 128, -1, NULL, 8)事件epoll
event.token()读回 data事件Token、Interest 与 Event
listener.accept()accept4(.., SOCK_CLOEXEC|SOCK_NONBLOCK)，直到 EAGAIN事件TCP
stream.read / writerecvfrom / sendto，直到 EAGAIN事件Source 与 IoSource
deregisterepoll_ctl(DEL, fd, NULL)注册Poll 与 Registry
waker.wake()write(eventfd, 1u64, 8)事件（旁路）Waker

除了 bind 和 Waker::new，每一行都是一次 mio 调用对应一次系统调用（accept 和 read 是循环里的每一次）。在跟踪的这些系统调用里，没有哪一次是 mio 为了维护自己的状态而发出的。

macOS 上换成了什么

macOS 上没有跟踪系统调用，下面按源码说：

  • Poll::new() 是 kqueue()。
  • register 是一次 kevent，changelist 里每个方向一项：这个程序只要 READABLE，所以只有一项 EVFILT_READ，标志 EV_CLEAR | EV_RECEIPT | EV_ADD，udata 是 token。上一页的动脑筋里，READABLE | WRITABLE 时是两项、事件也分两条。
  • poll 是 kevent(kq, NULL, 0, events, capacity, timeout)：changelist 为空，只取事件。
  • socket 之后要多两次 fcntl 设 O_NONBLOCK 和 FD_CLOEXEC，还要 setsockopt(SO_NOSIGPIPE)（new_socket 里 Darwin 的分支）；accept 用普通的 accept 加一次 fcntl(F_SETFD, FD_CLOEXEC)，非阻塞从监听 socket 继承。
  • Waker 不建 eventfd。macOS 上的 Waker::new 用 Selector::try_clone 复制一份 kqueue 的 fd，wake() 在这个 kqueue 上提交一个带 NOTE_TRIGGER 的 EVFILT_USER 事件，也是一次 kevent。

细节在 kqueue 和 Waker。用户代码一个字没改，poll 返回的事件序列也和 Linux 一样，这就是上一页说的公共语义。

小结

  • 一个 mio 事件循环：自己分配 Token，自己维护 token 到连接的表，每个事件都读写到 WouldBlock，写不完的数据自己存、需要时 reregister 加 WRITABLE。
  • Linux 上一次 register 是一次 epoll_ctl(EPOLL_CTL_ADD)，标志 EPOLLIN|EPOLLRDHUP|EPOLLET，data 就是 token；deregister 是 EPOLL_CTL_DEL；Poll::new 是 epoll_create1(EPOLL_CLOEXEC)。
  • 一次 poll 是一次 epoll_pwait（aarch64 没有 epoll_wait，musl 用 epoll_pwait 实现），容量和超时原样传下去。
  • 一个监听 socket 的可读事件可能对应多个连接：实测一次事件 accept4 成功两次、第三次 EAGAIN。新连接由 accept4(SOCK_NONBLOCK) 直接设成非阻塞，Linux 的 accept 不继承这个标志。
  • 对端半关闭时事件是 EPOLLIN|EPOLLRDHUP，is_read_closed() 为真，读返回 0。
  • wake() 是往 eventfd 写 8 字节的 1，服务器不用读它。
  • macOS 上系统调用换成 kqueue / kevent / EVFILT_USER，用户代码和事件序列不变：两个平台各 3 次运行，输出除端口外逐行相同。

💬 关联讨论

发起讨论
加载中...