poll(2) 后备

主线位置：两条主线都在这里改了形状。注册主线走到 selector 就停在用户态：register 不进内核，只是往一个 Vec<pollfd> 里追加一项，再用内部唤醒通道打断可能正在等待的线程。事件主线上，poll 每次把整张表交给 poll(2)（strace 里是 ppoll），报出一个事件就把对应的关注位摘掉；用户读写碰到 WouldBlock 时，do_io 把关注位放回去。

stateDiagram-v2
  state "关注位在 pollfd 里" as Armed
  state "已报出，关注位摘掉" as Fired
  state "已从表里删掉" as Gone
  [*] --> Armed: register
  Armed --> Fired: poll 报告就绪
  Fired --> Fired: 读写没到 WouldBlock
  Fired --> Armed: do_io 遇到 WouldBlock
  Fired --> Armed: 手动 reregister
  Fired --> Gone: POLLHUP 或 POLLERR
  Armed --> Gone: deregister 或 drop
  Gone --> [*]

代码在 sys/unix/selector/poll.rs（676 行）和 registered_io_source.rs（151 行）。文件开头写着它「基于 polling crate 里的实现」。IoSourceState 的两种形态、do_io 的代码在 Source 与 IoSource，poll(2) 后备下 Waker 的事件数在 Waker；这一页讲 selector 自己：表怎么存、一次 poll 怎么走、别的线程改表时发生什么，以及这些设计在系统调用上的代价。

💡TIP

打个比方：一个拿着名单巡楼的值班员

值班员手里有一张房间名单（poll_fds），每一项写着房号和要盯的事（events：有人敲门、能送东西进去）。他每次巡楼都把整张名单从头看一遍（一次 poll(2) 的代价和表长成正比）。发现 302 有人敲门，他报告给前台（Events 里多一个事件），然后在名单上把 302 的「敲门」划掉（poll_fd.events &= !poll_fd.revents），不然下一圈还会报同一件事。前台的人去 302 把门口的人接完，确认门口没人了（读到 WouldBlock），才回来把「敲门」重新写上（do_io 里的 reregister）。

巡楼时名单在值班员手里（select 持有 fds 锁）。别人要往名单上加房间，得先按铃（往 notify_waker 写一下）让他停下，把名单交出来，改完再还给他。

比方到这里为止：真实的值班员每圈都从头开始计时（超时被重置，见后文「跨线程修改」），而且铃声和住户叫人用的是同一个铃（用户 Waker 也走 notify_waker），这两点比方里的值班员不会有，正是这个实现容易出问题的地方。

谁在用它

selector 由 sys/unix/mod.rs 里 mod selector; 前面的四个 cfg_attr 选出来，最后一个指向 poll.rs：

    #[cfg_attr(any(
        mio_unsupported_force_poll_poll,
        target_os = "aix",
        target_os = "espidf",
        target_os = "nuttx",
        target_os = "fuchsia",
        target_os = "haiku",
        target_os = "hermit",
        target_os = "hurd",
        target_os = "nto",
        target_os = "vita",
        target_os = "cygwin",
        target_os = "wasi",
        target_os = "horizon"
    ), path = "selector/poll.rs")]

也就是 AIX、ESP-IDF、NuttX、Fuchsia、Haiku、Hermit、Hurd、QNX（nto）、Vita、Cygwin、Horizon，以及 WASI preview 2 及以后（preview 1 在更上一层就分到了 sys/wasip1）。其余三个 cfg_attr 都带 not(mio_unsupported_force_poll_poll)，所以设了这个 cfg，Linux 和 macOS 也会用 poll.rs。每个平台的 selector 和 Waker 合成的总表、这两个 cfg 在 README 里的说明，见源码地图。

在 Linux 上强制使用它，把 cfg 加进 RUSTFLAGS，和交叉编译的链接器参数写在一起：

RUSTFLAGS="-C linker=rust-lld -C linker-flavor=ld.lld --cfg mio_unsupported_force_poll_poll" \
  cargo build --release --target aarch64-unknown-linux-musl --bin rearm

另一个标志 mio_unsupported_force_waker_pipe 换的是 sys/unix/waker/ 选哪个文件。在 poll(2) 后备下，那个文件不当用户的 Waker 用，而是 selector 内部的唤醒通道 notify_waker（下一节）：Linux 上默认是 eventfd，加上 --cfg mio_unsupported_force_waker_pipe 就变成 pipe；kqueue 平台设了 force_poll_poll 时自动换成 pipe，因为 kqueue(2) 的 waker 不能和 poll(2) 一起用。

本页 Linux 的实验都在 Linux 6.12.28（Apple container 里的 Alpine，aarch64 musl 静态链接）、rustc 1.99.0 上跑，release 构建；macOS 的在 M2 Max、macOS、rustc 1.99.0 上跑。aarch64 没有 poll 系统调用，musl 的 poll() 用 ppoll 实现，所以 strace 里看到的是 ppoll，超时以 timespec 显示。每个程序的第一行 ppoll([{fd=0, ...}, {fd=1, ...}, {fd=2, ...}], 3, ...) 是 Rust 标准库启动时检查标准输入输出，和 mio 无关。

数据结构

Selector 只有一个字段 state: Arc<SelectorState>，try_clone 就是 Arc::clone。Registry::try_clone、Waker::new、每个注册过的 IoSource 手里的 selector 都是同一个 SelectorState。

SelectorState（文档注释删掉了）：

struct SelectorState {
    fds: Mutex<Fds>,
    pending_removal: Mutex<Vec<RawFd>>,
    pending_wake_token: Mutex<Option<Token>>,
    notify_waker: WakerInternal,
    waiting_operations: AtomicUsize,
    operations_complete: Condvar,
    #[cfg(debug_assertions)]
    #[allow(dead_code)]
    id: usize,
}

字段	存什么	谁用
`fds`	`poll_fds: Vec<PollFd>`（交给 `poll(2)` 的数组）和 `fd_data: HashMap<RawFd, FdData>`	`select` 全程持锁；`register` 等修改要拿这把锁
`pending_removal`	在 `poll` 进行中被注销的 fd	`select` 返回前用它过滤掉这些 fd 的事件
`pending_wake_token`	最近一次 `Waker::wake` 的 token	`select` 被叫醒时合成一个事件
`notify_waker`	内部唤醒通道（eventfd、pipe，或 horizon / WASI 上的 `AtomicBool`）	修改 `fds` 前叫醒 `select`；用户 `Waker` 也借用它
`waiting_operations`	正在等 `fds` 锁的修改个数	不为 0 时 `select` 先让路
`operations_complete`	条件变量，配合 `fds` 锁	修改做完后通知 `select` 继续
`id`	debug 下的 `SelectorId` 编号	见 Source 与 IoSource

Fds 的注释说「第一个 fd 总是存在，用来通知 poller」：SelectorState::new 先建 notify_waker，再把它的 fd 以 POLLIN 放进 poll_fds[0]（horizon / WASI 的单线程版本没有 fd，poll_fds 从空开始）。之后每注册一个 fd，poll_fds 末尾多一项，fd_data 里多一条 FdData：它在 poll_fds 里的下标、token，以及和 IoSourceState 共享的 RegistrationRecord（后面讲关闭时用到）。

和 epoll / kqueue 不同，Token 不进内核：pollfd 里没有放用户数据的地方，token 存在 fd_data 里，事件回来时按 fd 查表取出。

关注集合到 pollfd.events 的翻译在 interests_to_poll：

`Interest`	`events` 里的位
`READABLE`	`POLLIN \| POLLRDHUP`（`POLLRDHUP` 只在 Linux 上有，其他平台定义成 0）
`WRITABLE`	`POLLOUT`
`PRIORITY`	`POLLPRI`

返回的事件是 Event { token, events: PollFlagInt }，events 就是内核填的 revents。is_readable 看 POLLIN 或 POLLPRI，is_writable 看 POLLOUT，is_error 看 POLLERR，is_read_closed 看 POLLHUP 或 POLLRDHUP，is_aio / is_lio 恒为 false。

register：不进内核

register_internal 做三件事：

        let mut pending_removal = self.pending_removal.lock().unwrap();
        if let Some(idx) = pending_removal.iter().position(|&pending| pending == fd) {
            pending_removal.swap_remove(idx);
        }
        drop(pending_removal);

        self.modify_fds(|fds| {
            if fds.fd_data.contains_key(&fd) {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    "I/O source already registered this `Registry` \
                    (an old file descriptor might have been closed without deregistration)",
                ));
            }

            let poll_fds_index = fds.poll_fds.len();
            let record = Arc::new(RegistrationRecord::new());
            fds.fd_data.insert(
                // ...
            );

            fds.poll_fds.push(PollFd(libc::pollfd {
                fd,
                events: interests_to_poll(interests),
                revents: 0,
            }));

            Ok(record)
        })

把这个 fd 从 pending_removal 里拿掉。源码注释举的例子是「register(1)、deregister(1)、register(1)，然后才 poll」：pending_removal 只在 poll 时清，不拿掉的话新注册的 fd 1 的事件会被当成已注销的过滤掉。
同一个 fd 已经在表里就返回 AlreadyExists。错误信息里的括号说出了最常见的原因：旧 fd 没注销就关了，号码被新的 socket 复用。
追加一项。reregister 也走 modify_fds，按下标把 events 整个覆盖成新的关注集合，并更新 token。

另外 debug 下 register_internal 拒绝注册 notify_waker 自己的 fd（InvalidInput）。

内核在 register 时什么都不知道，所以 epoll 在注册时就能报的错（比如 fd 无效），这里要到 poll 才露出来，形式也不是错误，见下面这个框。

📝NOTE

没有蠢问题

问：用 SourceFd 注册一个根本没打开的 fd，register 会失败吗？

答：在 epoll 上会，在 poll(2) 后备上不会，而且之后每次 poll 都立刻返回。src/bin/bad_fd.rs 注册 fd 999，再 poll 三次，每次超时 200ms：

use std::time::{Duration, Instant};

use mio::unix::SourceFd;
use mio::{Events, Interest, Poll, Token};

fn main() -> std::io::Result<()> {
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  let fd = 999; // not open
  println!("register fd 999: {:?}", poll.registry().register(&mut SourceFd(&fd), Token(9), Interest::READABLE));
  for round in 1..=3 {
    let start = Instant::now();
    poll.poll(&mut events, Some(Duration::from_millis(200)))?;
    let got: Vec<String> = events.iter().map(|e| format!("{:?} r={} err={}", e.token(), e.is_readable(), e.is_error())).collect();
    println!("poll {round}: {got:?} after {}ms", start.elapsed().as_millis());
  }
  Ok(())
}

Linux，epoll：

register fd 999: Err(Os { code: 9, kind: Uncategorized, message: "Bad file descriptor" })
poll 1: [] after 210ms
poll 2: [] after 201ms
poll 3: [] after 208ms

Linux，强制 poll(2)：

register fd 999: Ok(())
poll 1: ["Token(9) r=false err=false"] after 0ms
poll 2: ["Token(9) r=false err=false"] after 0ms
poll 3: ["Token(9) r=false err=false"] after 0ms

strace 里每次都是 ppoll(... {fd=999, events=POLLIN|POLLRDHUP}], 2, {tv_sec=0, tv_nsec=200000000}, NULL, 8) = 1 ([{fd=999, revents=POLLNVAL}], ...)。内核对无效 fd 回 POLLNVAL。mio 只把 POLLHUP | POLLERR 当作关闭，POLLNVAL 不在其中，所以这一项不会被内部注销；摘关注位的 events &= !revents 也摘不掉它，因为 POLLNVAL 本来就不在 events 里。结果是一个所有 is_* 都为 false 的事件，每次 poll 都立刻出现。各跑 3 次，结果相同。mio 自己的 socket 类型 drop 时会注销，碰不到这种情况；自己管 fd、用 SourceFd 的代码，关 fd 之前必须先 deregister。

select：一次 poll 怎么走

select 的骨架（trace! 和部分注释删掉了）：

    pub fn select(&self, events: &mut Events, mut timeout: Option<Duration>) -> io::Result<()> {
        events.clear();

        let mut fds = self.fds.lock().unwrap();
        // ...
        let mut closed_raw_fds = Vec::new();

        loop {
            // Complete all current operations.
            loop {
                if self.waiting_operations.load(Ordering::SeqCst) == 0 {
                    break;
                }

                fds = self.operations_complete.wait(fds).unwrap();
            }

            if self.notify_waker.woken() {
                timeout = Some(Duration::from_secs(0))
            }

            // Perform the poll.
            let num_events = poll(&mut fds.poll_fds, timeout)?;

            if num_events == 0 {
                return Ok(());
            }

            // ...

            // Store the events if there were any.
            if num_fd_events > 0 {
                let fds = &mut *fds;

                events.reserve(num_fd_events);

                // Add synthetic events we picked up from calls to wake()
                if let Some(pending_wake_token) = pending_wake_token {
                    events.push(Event {
                        token: pending_wake_token,
                        events: waker_events,
                    });
                }

                for fd_data in fds.fd_data.values_mut() {
                    let PollFd(poll_fd) = &mut fds.poll_fds[fd_data.poll_fds_index];

                    if pending_removal.contains(&poll_fd.fd) {
                        // Fd was removed while poll was running
                        continue;
                    }

                    if poll_fd.revents != 0 {
                        // Store event
                        events.push(Event {
                            token: fd_data.token,
                            events: poll_fd.revents,
                        });

                        if poll_fd.revents & (libc::POLLHUP | libc::POLLERR) != 0 {
                            pending_removal.push(poll_fd.fd);
                            closed_raw_fds.push(poll_fd.fd);
                        }

                        // Remove the interest which just got triggered the IoSourceState's do_io
                        // wrapper used with this selector will add back the interest using
                        // reregister.
                        poll_fd.events &= !poll_fd.revents;

                        // ...
                        if events.len() == num_fd_events {
                            break;
                        }
                    }
                }

                break; // No more polling.
            }

            // If we didn't break above it means we got woken up internally (for example for adding an fd), so we poll again.
        }

        drop(fds);
        let _ = self.deregister_all(&closed_raw_fds);

        Ok(())
    }

按顺序：

整个 select 持有 fds 锁，包括阻塞在 poll(2) 里的那段时间。 内核要读写 poll_fds 这块内存，等待期间不能让别人改它。代价是别的线程的 register / reregister / deregister 都得先把这个线程叫醒，见「跨线程修改」。
进 poll(2) 之前，如果有修改在排队（waiting_operations 不为 0），先在条件变量上等，等的时候锁是放开的。
poll(2) 返回 0 就是超时，直接返回。返回值不为 0 时，先看 poll_fds[0]：它有 revents 说明被内部通道叫醒过（notified），要先 ack_and_reset 把通道读空，否则水平触发的 poll(2) 下次会立刻返回；这时如果 pending_wake_token 里有 token，就合成一个用户 Waker 的事件。
真正的 fd 事件个数是 num_fd_events。它为 0，说明这次只是被内部叫醒（有人要改表），回到循环开头，用原来的 timeout 再等一次。
否则遍历 fd_data，收集 revents != 0 的项，跳过 pending_removal 里的 fd；碰到 POLLHUP / POLLERR 的记下来，返回前统一注销；然后把报出的位从 events 里摘掉。凑够 num_fd_events 个就提前退出遍历，注释称之为「Minor optimization」。

Events 的容量在这里不算数

epoll 和 kqueue 把 events.capacity() 当作 maxevents 交给内核，所以容量就是一次 poll 的上限。这里的 sys::Events 是 Vec<Event>，select 先 events.reserve(num_fd_events) 再 push，有多少就给多少。Poll::poll 文档里「At most events.capacity() events will be returned」在 poll(2) 后备上不成立。另一个差别是顺序：事件按 fd_data（一个 HashMap）的遍历顺序产生，和注册顺序、fd 大小都无关。

src/bin/capacity.rs：5 个服务端 socket 在第一次 poll 前都已可读，Events::with_capacity(2)，一直 poll 到没有事件为止：

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

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

fn main() -> std::io::Result<()> {
  let l = std::net::TcpListener::bind("127.0.0.1:0")?;
  let mut poll = Poll::new()?;
  let mut keep = vec![];
  for i in 0..5 {
    let mut c = std::net::TcpStream::connect(l.local_addr()?)?;
    let (s, _) = l.accept()?;
    s.set_nonblocking(true)?;
    let mut s = TcpStream::from_std(s);
    c.write_all(b"x")?; // all five are readable before the first poll
    poll.registry().register(&mut s, Token(i), Interest::READABLE)?;
    keep.push((s, c));
  }
  std::thread::sleep(Duration::from_millis(50));
  let mut events = Events::with_capacity(2);
  println!("Events::with_capacity(2), capacity()={}", events.capacity());
  for round in 1.. {
    poll.poll(&mut events, Some(Duration::from_millis(100)))?;
    let got: Vec<usize> = events.iter().map(|e| e.token().0).collect();
    println!("  poll #{round}: tokens {got:?}, capacity()={}", events.capacity());
    if got.is_empty() { break; }
  }
  Ok(())
}

Linux，epoll（3 次输出相同）：

Events::with_capacity(2), capacity()=2
  poll #1: tokens [0, 1], capacity()=2
  poll #2: tokens [2, 3], capacity()=2
  poll #3: tokens [4], capacity()=2
  poll #4: tokens [], capacity()=2

Linux，强制 poll(2)，3 次：

Events::with_capacity(2), capacity()=2
  poll #1: tokens [2, 3, 1, 4, 0], capacity()=5
  poll #2: tokens [], capacity()=5

Events::with_capacity(2), capacity()=2
  poll #1: tokens [4, 2, 3, 1, 0], capacity()=5
  poll #2: tokens [], capacity()=5

Events::with_capacity(2), capacity()=2
  poll #1: tokens [2, 4, 3, 0, 1], capacity()=5
  poll #2: tokens [], capacity()=5

一次拿全 5 个，Events 自己长到了 5，每次的顺序都不一样（HashMap 的哈希种子每个进程随机）。

用水平触发模拟边沿触发

poll(2) 是水平触发：只要 socket 里还有数据，每次都报。mio 的公共语义是边沿触发（见 mio 解决什么问题），上层（比如 tokio）按「报一次，读到 WouldBlock 之前不会再报」来写。模拟靠两步：

摘：select 报出一个事件后执行 poll_fd.events &= !poll_fd.revents。报了 POLLIN，就把 POLLIN 从这个 fd 的关注里去掉，下次 poll(2) 不再看它。POLLRDHUP 没报出来就还留着。
放回：mio 的 socket 类型的读写都经过 IoSource::do_io，poll(2) 后备上它调 registered_io_source.rs 的 do_io：闭包返回 WouldBlock 时，用注册时记下的 fd、token、interest 调一次 selector.reregister。代码和 InternalState 的字段在 Source 与 IoSource。

reregister 是整个覆盖：

    pub fn reregister(&self, fd: RawFd, token: Token, interests: Interest) -> io::Result<()> {
        self.modify_fds(|fds| {
            let data = fds.fd_data.get_mut(&fd).ok_or(io::ErrorKind::NotFound)?;
            data.token = token;
            let poll_fds_index = data.poll_fds_index;
            fds.poll_fds[poll_fds_index].0.events = interests_to_poll(interests);

            Ok(())
        })
    }

所以一次读到 WouldBlock，放回的是注册时的全部关注，不只是读。这一点在本页末尾的动脑筋里会看到后果。

strace：摘掉和放回

src/bin/rearm.rs：注册一个 listener，连一次、accept 到 WouldBlock；对端写 3 字节，读一次（读到 3 字节，没到 WouldBlock），poll(0)；再读一次（WouldBlock），poll(0)。每一步之前打印一行 == 标记，标准输出重定向掉，strace 里的 write(1, "== ...") 就是分隔线。

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

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

fn main() -> std::io::Result<()> {
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
  println!("== register listener");
  poll.registry().register(&mut listener, Token(0), Interest::READABLE)?;

  let mut client = std::net::TcpStream::connect(listener.local_addr()?)?;
  println!("== poll");
  poll.poll(&mut events, None)?;
  println!("== accept");
  let (mut stream, _) = listener.accept()?;
  println!("== register stream");
  poll.registry().register(&mut stream, Token(1), Interest::READABLE)?;
  println!("== accept again");
  assert_eq!(listener.accept().unwrap_err().kind(), ErrorKind::WouldBlock);

  client.write_all(b"abc")?;
  println!("== poll");
  poll.poll(&mut events, None)?;
  let mut buf = [0u8; 16];
  println!("== read");
  let n = stream.read(&mut buf)?;
  println!("== poll(0) after reading {n} bytes, before WouldBlock");
  poll.poll(&mut events, Some(Duration::ZERO))?;
  println!("== read again");
  assert_eq!(stream.read(&mut buf).unwrap_err().kind(), ErrorKind::WouldBlock);
  println!("== poll(0) after WouldBlock");
  poll.poll(&mut events, Some(Duration::ZERO))?;
  Ok(())
}

强制 poll(2)，strace -e trace=ppoll,eventfd2,accept4,read,write,futex,recvfrom,sendto,connect（3 次除端口和栈地址外逐行相同）：

ppoll([{fd=0, events=0}, {fd=1, events=0}, {fd=2, events=0}], 3, {tv_sec=0, tv_nsec=0}, NULL, 8) = 0 (Timeout)
eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)   = 3
write(1, "== register listener\n", 21)  = 21
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
futex(0xffff92cf92e0, FUTEX_WAKE_PRIVATE, 1) = 0
connect(5, {sa_family=AF_INET, sin_port=htons(39675), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
write(1, "== poll\n", 8)                = 8
ppoll([{fd=3, events=POLLIN}, {fd=4, events=POLLIN|POLLRDHUP}], 2, NULL, NULL, 8) = 1 ([{fd=4, revents=POLLIN}])
write(1, "== accept\n", 10)             = 10
accept4(4, {sa_family=AF_INET, sin_port=htons(52044), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 6
write(1, "== register stream\n", 19)    = 19
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
futex(0xffff92cf92e0, FUTEX_WAKE_PRIVATE, 1) = 0
write(1, "== accept again\n", 16)       = 16
accept4(4, 0xfffff83437c8, [128], SOCK_CLOEXEC|SOCK_NONBLOCK) = -1 EAGAIN (Resource temporarily unavailable)
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
futex(0xffff92cf92e0, FUTEX_WAKE_PRIVATE, 1) = 0
sendto(5, "abc", 3, MSG_NOSIGNAL, NULL, 0) = 3
write(1, "== poll\n", 8)                = 8
ppoll([{fd=3, events=POLLIN}, {fd=4, events=POLLIN|POLLRDHUP}, {fd=6, events=POLLIN|POLLRDHUP}], 3, NULL, NULL, 8) = 1 ([{fd=6, revents=POLLIN}])
write(1, "== read\n", 8)                = 8
recvfrom(6, "abc", 16, 0, NULL, NULL)   = 3
write(1, "== poll(0) after reading 3 bytes"..., 52) = 52
ppoll([{fd=3, events=POLLIN}, {fd=4, events=POLLIN|POLLRDHUP}, {fd=6, events=POLLRDHUP}], 3, {tv_sec=0, tv_nsec=0}, NULL, 8) = 0 (Timeout)
write(1, "== read again\n", 14)         = 14
recvfrom(6, 0xfffff8343990, 16, 0, NULL, NULL) = -1 EAGAIN (Resource temporarily unavailable)
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
futex(0xffff92cf92e0, FUTEX_WAKE_PRIVATE, 1) = 0
write(1, "== poll(0) after WouldBlock\n", 28) = 28
ppoll([{fd=3, events=POLLIN}, {fd=4, events=POLLIN|POLLRDHUP}, {fd=6, events=POLLIN|POLLRDHUP}], 3, {tv_sec=0, tv_nsec=0}, NULL, 8) = 0 (Timeout)
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
futex(0xffff92cf92e0, FUTEX_WAKE_PRIVATE, 1) = 0
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
futex(0xffff92cf92e0, FUTEX_WAKE_PRIVATE, 1) = 0
+++ exited with 0 +++

逐段看：

两次 register、两次放回、结尾的两次注销，各是一组 write(3) / read(3) / futex，一共 6 组。每组都是一次 modify_fds：先往 eventfd 写 1 想叫醒正在 poll 的线程，拿到锁后自己把它读掉，最后的 futex(..., FUTEX_WAKE_PRIVATE, 1) 是 Condvar::notify_one。这个程序只有一个线程，没人在 poll，三个系统调用全是白做的，下一节解释为什么不能省。
摘：第 3 次 ppoll（不算第一行标准库那次）时 fd 6 的 events 只剩 POLLRDHUP。上一次 ppoll 报出了 POLLIN，select 把它摘了；第一次读拿到 3 字节，没到 WouldBlock，没人放回。
放回：recvfrom 返回 EAGAIN 之后紧跟一组 write(3) / read(3) / futex，第 4 次 ppoll 时 fd 6 又是 POLLIN|POLLRDHUP。listener（fd 4）也一样：第一次 ppoll 报出 POLLIN 后被摘，accept4 的 EAGAIN 把它放回，所以第 2 次 ppoll 里它是完整的。
结尾：stream 和 listener drop 时 InternalState 的 Drop 各调一次 deregister，又是两组。

同一个程序在 epoll 上（同样的过滤条件加 epoll_*）：

ppoll([{fd=0, events=0}, {fd=1, events=0}, {fd=2, events=0}], 3, {tv_sec=0, tv_nsec=0}, NULL, 8) = 0 (Timeout)
epoll_create1(EPOLL_CLOEXEC)            = 3
write(1, "== register listener\n", 21)  = 21
epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
connect(5, {sa_family=AF_INET, sin_port=htons(45043), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
write(1, "== poll\n", 8)                = 8
epoll_pwait(3, [{events=EPOLLIN, data=0}], 8, -1, NULL, 8) = 1
write(1, "== accept\n", 10)             = 10
accept4(4, {sa_family=AF_INET, sin_port=htons(60598), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 6
write(1, "== register stream\n", 19)    = 19
epoll_ctl(3, EPOLL_CTL_ADD, 6, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
write(1, "== accept again\n", 16)       = 16
accept4(4, 0xffffd989b298, [128], SOCK_CLOEXEC|SOCK_NONBLOCK) = -1 EAGAIN (Resource temporarily unavailable)
sendto(5, "abc", 3, MSG_NOSIGNAL, NULL, 0) = 3
write(1, "== poll\n", 8)                = 8
epoll_pwait(3, [{events=EPOLLIN, data=0x1}], 8, -1, NULL, 8) = 1
write(1, "== read\n", 8)                = 8
recvfrom(6, "abc", 16, 0, NULL, NULL)   = 3
write(1, "== poll(0) after reading 3 bytes"..., 52) = 52
epoll_pwait(3, [], 8, 0, NULL, 8)       = 0
write(1, "== read again\n", 14)         = 14
recvfrom(6, 0xffffd989b3a0, 16, 0, NULL, NULL) = -1 EAGAIN (Resource temporarily unavailable)
write(1, "== poll(0) after WouldBlock\n", 28) = 28
epoll_pwait(3, [], 8, 0, NULL, 8)       = 0
+++ exited with 0 +++

register 是一次 epoll_ctl，WouldBlock 之后什么也不发生，fd 关闭时内核自己把它从 epoll 里拿掉。epoll 的细节在 epoll。

不读到 WouldBlock：这里比 epoll 更严

epoll 和 kqueue 的「边」是新数据到达这个动作。一次事件之后没读完，只要对端再写，内核还会再报。poll(2) 后备的「边」是 do_io 看到 WouldBlock：没读到 WouldBlock，关注位就一直是摘掉的，对端再写多少都不报。

src/bin/partial_read.rs，用 mio 的 TcpStream（读写经过 do_io），每步之后 poll 100ms 并打印事件数：

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

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

fn main() -> std::io::Result<()> {
  let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
  let mut peer = std::net::TcpStream::connect(listener.local_addr()?)?;
  let (s, _) = 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)?;
  let mut buf = [0u8; 4];
  let mut count = |poll: &mut Poll| -> std::io::Result<usize> {
    poll.poll(&mut events, Some(Duration::from_millis(100)))?;
    Ok(events.iter().count())
  };

  peer.write_all(b"12345678")?;
  println!("1. peer writes 8 bytes        -> poll: {} event(s)", count(&mut poll)?);
  stream.read_exact(&mut buf)?;
  println!("2. read 4, leave 4 unread     -> poll: {} event(s)", count(&mut poll)?);
  peer.write_all(b"abcdefgh")?;
  println!("3. peer writes 8 more         -> poll: {} event(s)", count(&mut poll)?);
  let mut total = 0;
  loop {
    match stream.read(&mut buf) {
      Ok(n) => total += n,
      Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
      Err(e) => return Err(e),
    }
  }
  println!("4. read {total:>2} until WouldBlock -> poll: {} event(s)", count(&mut poll)?);
  peer.write_all(b"x")?;
  println!("5. peer writes 1 byte         -> poll: {} event(s)", count(&mut poll)?);
  Ok(())
}

Linux epoll 和 macOS kqueue 输出相同（各 3 次）：

1. peer writes 8 bytes        -> poll: 1 event(s)
2. read 4, leave 4 unread     -> poll: 0 event(s)
3. peer writes 8 more         -> poll: 1 event(s)
4. read 12 until WouldBlock -> poll: 0 event(s)
5. peer writes 1 byte         -> poll: 1 event(s)

Linux，强制 poll(2)（3 次）：

1. peer writes 8 bytes        -> poll: 1 event(s)
2. read 4, leave 4 unread     -> poll: 0 event(s)
3. peer writes 8 more         -> poll: 0 event(s)
4. read 12 until WouldBlock -> poll: 0 event(s)
5. peer writes 1 byte         -> poll: 1 event(s)

差别只在第 3 行。一个在 Linux 上靠「对端总会再写点什么」侥幸跑通的程序，换到 poll(2) 平台就会卡住。tokio 的 PollEvented::poll_read 有一个「读不满就当作已读空」的优化，cfg 里专门排除了 mio_unsupported_force_poll_poll，原因就是这个：在这里少读一次 WouldBlock，关注位就回不来。来历见 tokio 卷从 TcpStream::read 到挂起。

不经过 do_io 的 SourceFd 永远得不到放回，读到 WouldBlock 也没用，见 Source 与 IoSource 文末的动脑筋。这个问题在 1.2.4 之后有人报为 issue #2018（「poll(2) selector: SourceFd gets only the first event」），截至本卷写作时未关闭。

📝NOTE

冷知识：最初的 poll(2) 后备方案打算直接改成水平触发

poll(2) 后备最早的方案是 PR #1602「Poll selector」，没有合并。它的说明列出了和 polling crate 实现的差别，其中一条是「mio (and tokio) used to assume edge triggering, but poll is level triggered (may be a breaking change!)」；Waker 还要新增一个 did_wake 方法，「needs to be called after a task has been woken up in order to reset the event source」，并且需要给 tokio 打补丁。

真正合并的是 PR #1687「Initial support for poll-based backend」，首次发布于 0.8.9。说明里写它「largely based off #1602 with improvements to fix tests and provide a more consistent implementation」，列出的改动包括「Closed fds are now automatically removed as is the case with the Windows implementation」和「Waker is now a pass through for the internal waker that the poll Selector needs」。它的第一个提交（「Implement poll-based backend」）里已经有本节的两步：select 里的 poll_fd.events &= !poll_fd.revents，和 IoSourceState::do_io 在 WouldBlock 时 reregister。于是对用户来说语义没变，依然是边沿触发；改变的代价落在 mio 内部：每次 WouldBlock 都多一次 modify_fds。

跨线程修改：先把正在 poll 的线程叫醒

epoll 和 kqueue 允许一个线程阻塞在 epoll_wait / kevent 里，另一个线程同时 epoll_ctl。这里不行：select 等待期间一直持有 fds 锁。所有修改都走 modify_fds：

    fn modify_fds<T>(&self, f: impl FnOnce(&mut Fds) -> T) -> T {
        self.waiting_operations.fetch_add(1, Ordering::SeqCst);

        // Wake up the current caller of `wait` if there is one.
        let sent_notification = self.notify_waker.wake().is_ok();

        let mut fds = self.fds.lock().unwrap();

        // If there was no caller of `wait` our notification was not removed from the pipe.
        if sent_notification {
            self.notify_waker.ack_and_reset();
        }

        let res = f(&mut *fds);

        if self.waiting_operations.fetch_sub(1, Ordering::SeqCst) == 1 {
            self.operations_complete.notify_one();
        }

        res
    }

sequenceDiagram
  participant P as 轮询线程
  participant N as notify_waker
  participant H as 注册线程
  P->>P: 持锁进入 poll(2)
  H->>H: 等待计数加 1
  H->>N: wake 写 1
  N-->>P: fd 0 可读，返回
  P->>N: 读空通道
  P->>P: 无真事件，放锁等待
  H->>N: 拿锁后再读一次
  H->>H: 改表
  H->>P: 计数归零 notify_one
  P->>P: 用原超时再 poll

modify_fds 不知道此刻有没有线程在 poll，所以总是先写一次通道；拿到锁之后，不管轮询线程读没读过，自己再读一次，保证通道是空的（否则下一次 poll(2) 会被一个早已过时的通知立刻叫醒）。waiting_operations 让 select 在被叫醒后先让路，不至于又抢在修改之前拿回锁、再进 poll(2)。上一节 strace 里单线程程序的每组 write(3) / read(3) / futex 就是这一套的固定开销。

实验：别的线程在 poll 期间注册

src/bin/cross_register.rs：主线程 poll(Some(300ms))，辅助线程 100ms 后用克隆的 Registry 注册一个没人连的 listener：

use std::thread;
use std::time::{Duration, Instant};

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

fn main() -> std::io::Result<()> {
  let mut poll = Poll::new()?;
  let registry = poll.registry().try_clone()?;
  let start = Instant::now();
  let helper = thread::spawn(move || -> std::io::Result<TcpListener> {
    thread::sleep(Duration::from_millis(100));
    let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
    // nobody connects: the listener never becomes ready
    registry.register(&mut listener, Token(7), Interest::READABLE)?;
    println!("[{:>3}ms] helper: registered", start.elapsed().as_millis());
    Ok(listener)
  });

  let mut events = Events::with_capacity(8);
  poll.poll(&mut events, Some(Duration::from_millis(300)))?;
  println!("[{:>3}ms] main: poll(300ms) returned {} event(s)", start.elapsed().as_millis(), events.iter().count());
  let _listener = helper.join().unwrap()?;
  Ok(())
}

三种情况各跑 3 次，主线程返回的时刻：

平台	第 1 次	第 2 次	第 3 次
macOS（kqueue）	302ms	302ms	301ms
Linux（epoll）	304ms	304ms	304ms
Linux（强制 `poll(2)`）	412ms	420ms	416ms

其中一次强制 poll(2) 的完整输出：

[110ms] helper: registered
[412ms] main: poll(300ms) returned 0 event(s)

poll(2) 后备上，300ms 的超时变成了「最后一次被打断之后再等 300ms」。select 被内部叫醒、发现没有真事件时回到循环开头，timeout 还是调用者给的原值，没有减去已经等掉的时间。strace（-f -e trace=ppoll,eventfd2,read,write,futex，3 次除地址、剩余时间和毫秒数外相同）：

ppoll([{fd=0, events=0}, {fd=1, events=0}, {fd=2, events=0}], 3, {tv_sec=0, tv_nsec=0}, NULL, 8) = 0 (Timeout)
eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)   = 3
strace: Process 4 attached
[pid     3] ppoll([{fd=3, events=POLLIN}], 1, {tv_sec=0, tv_nsec=300000000}, NULL, 8 <unfinished ...>
[pid     4] write(3, "\1\0\0\0\0\0\0\0", 8) = 8
[pid     3] <... ppoll resumed>)        = 1 ([{fd=3, revents=POLLIN}], left {tv_sec=0, tv_nsec=189269125})
[pid     4] futex(0xffffb64e1270, FUTEX_WAIT_BITSET_PRIVATE, 2, NULL, FUTEX_BITSET_MATCH_ANY <unfinished ...>
[pid     3] read(3, "\1\0\0\0\0\0\0\0", 8) = 8
[pid     3] futex(0xffffb64e1270, FUTEX_WAKE_PRIVATE, 1) = 1
[pid     4] <... futex resumed>)        = 0
[pid     3] futex(0xffffb64e12e0, FUTEX_WAIT_BITSET_PRIVATE, 0, NULL, FUTEX_BITSET_MATCH_ANY <unfinished ...>
[pid     4] read(3, 0xffffb64e06c0, 8)  = -1 EAGAIN (Resource temporarily unavailable)
[pid     4] futex(0xffffb64e12e0, FUTEX_WAKE_PRIVATE, 1 <unfinished ...>
[pid     3] <... futex resumed>)        = 0
[pid     4] <... futex resumed>)        = 1
[pid     3] futex(0xffffb64e1270, FUTEX_WAIT_BITSET_PRIVATE, 2, NULL, FUTEX_BITSET_MATCH_ANY <unfinished ...>
[pid     4] futex(0xffffb64e1270, FUTEX_WAKE_PRIVATE, 1 <unfinished ...>
[pid     3] <... futex resumed>)        = -1 EAGAIN (Resource temporarily unavailable)
[pid     4] <... futex resumed>)        = 0
[pid     3] ppoll([{fd=3, events=POLLIN}, {fd=4, events=POLLIN|POLLRDHUP}], 2, {tv_sec=0, tv_nsec=300000000}, NULL, 8 <unfinished ...>
[pid     4] write(1, "[111ms] helper: registered\n", 27) = 27
[pid     4] futex(0xffffb64e0ac0, FUTEX_WAKE_PRIVATE, 1) = 0
[pid     4] +++ exited with 0 +++
<... ppoll resumed>)                    = 0 (Timeout)
futex(0xffffb64e1270, FUTEX_WAKE_PRIVATE, 1) = 0
write(1, "[414ms] main: poll(300ms) return"..., 46) = 46
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
futex(0xffffb64e12e0, FUTEX_WAKE_PRIVATE, 1) = 0
+++ exited with 0 +++

和时序图一一对应：辅助线程（pid 4）写 eventfd，然后阻塞在 fds 锁的 futex 上；主线程的 ppoll 返回（还剩约 189ms），读掉通知，在条件变量（另一个 futex 地址）上等并放锁；辅助线程拿到锁，自己的 ack_and_reset 读到 EAGAIN（已经被主线程读空），改表，notify_one；主线程拿回锁，第二次 ppoll 带着 2 个 fd 和完整的 tv_nsec=300000000。最后一组 write(3) / read(3) 是 listener 在主线程 drop 时注销。

注册之外，reregister、deregister，以及别的线程上的 do_io 遇到 WouldBlock 时的放回，走的都是同一个 modify_fds。所以只要有一个线程阻塞在 poll 里，别的线程上每一次读写碰到 WouldBlock，都会打断它一次，并让它的超时从头算起。

和用户 Waker 共用一个通道

用户的 Waker 在这里不建自己的 fd，wake 把 token 放进 pending_wake_token，再写同一个 notify_waker（Waker 用 strace 看过它）。select 只在 notified 为真（poll_fds[0] 有 revents）时才把 pending_wake_token 变成事件。

modify_fds 拿到锁之后的 ack_and_reset 不分来源：如果此刻没有线程在 poll，它会把用户 wake() 写进去的那一下也读掉。token 还留在 pending_wake_token 里，但下一次 poll(2) 看不到通道可读，就不会去取它。src/bin/lost_wake.rs：先 wake()、再 register 一个 listener、再 poll 500ms：

use std::time::{Duration, Instant};

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

fn main() -> std::io::Result<()> {
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  let waker = Waker::new(poll.registry(), Token(10))?;

  waker.wake()?;
  let start = Instant::now();
  poll.poll(&mut events, Some(Duration::from_millis(500)))?;
  let got: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  println!("wake, poll               -> {got:?} after {}ms", start.elapsed().as_millis());

  waker.wake()?;
  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
  poll.registry().register(&mut listener, Token(1), Interest::READABLE)?;
  let start = Instant::now();
  poll.poll(&mut events, Some(Duration::from_millis(500)))?;
  let got: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  println!("wake, register, poll     -> {got:?} after {}ms", start.elapsed().as_millis());
  let start = Instant::now();
  poll.poll(&mut events, Some(Duration::from_millis(500)))?;
  let got: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  println!("poll again               -> {got:?} after {}ms", start.elapsed().as_millis());
  Ok(())
}

macOS 和 Linux epoll（各 3 次，只有毫秒数不同），其中一次：

wake, poll               -> [10] after 0ms
wake, register, poll     -> [10] after 0ms
poll again               -> [] after 508ms

Linux，强制 poll(2)（3 次，只有毫秒数不同），其中一次：

wake, poll               -> [10] after 0ms
wake, register, poll     -> [] after 510ms
poll again               -> [] after 509ms

strace（-e trace=ppoll,eventfd2,read,write）里看得很清楚：

write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
read(3, "\2\0\0\0\0\0\0\0", 8)          = 8
ppoll([{fd=3, events=POLLIN}, {fd=4, events=POLLIN|POLLRDHUP}], 2, {tv_sec=0, tv_nsec=500000000}, NULL, 8) = 0 (Timeout)

第一个 write 是 wake()，第二个是 register 里的 modify_fds，接着那次 read 读出 2，把两次通知一起清掉；ppoll 睡满 500ms。pending_wake_token 里的 Token(10) 要等到下一次有东西写通道时才会出现。前提是「没有线程正在 poll」：如果轮询线程在 poll(2) 里，它会先被叫醒、先读通道、先取走 token，事件不会丢。

pending_wake_token 是 Option<Token>，多个 Waker 时后一个覆盖前一个（Waker 的两个 Waker 实验）。#1687 里有一个提交把它从 Vec<Token> 改成了 Option<Token>，标题是「poll: Simplify and optimize pending_wake_tokens to use a singular Token」，没有更多说明。

关闭的 fd：内部注销

select 碰到 POLLHUP 或 POLLERR 的 fd，把它记进 closed_raw_fds，放锁之后调 deregister_all 删掉：从 fd_data 移除，poll_fds.swap_remove(下标)，再把被换过来的那一项在 fd_data 里的下标改对；同时对它的 RegistrationRecord 调 mark_unregistered。这个记录和 IoSourceState 共享，它的注释说明了用途：内部注销之后，IoSourceState 不要再注销一次，「技术上 deregister 是幂等的，但调试时会看到不成对的 register / deregister 调用和多余的 NotFound 错误」。

不过记录只在 InternalState 的 Drop 里被检查。用户主动调的 deregister 和 reregister 不看它，照样去 selector 里找这个 fd。src/bin/hup.rs：UnixStream::pair()，注册一端，drop 另一端：

use std::time::Duration;

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

fn main() -> std::io::Result<()> {
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  let (mut a, b) = UnixStream::pair()?;
  poll.registry().register(&mut a, Token(1), Interest::READABLE)?;
  drop(b);
  poll.poll(&mut events, Some(Duration::from_millis(100)))?;
  for e in &events {
    println!("poll 1: {:?} readable={} read_closed={}", e.token(), e.is_readable(), e.is_read_closed());
  }
  poll.poll(&mut events, Some(Duration::from_millis(100)))?;
  println!("poll 2: {} event(s)", events.iter().count());
  println!("reregister: {:?}", poll.registry().reregister(&mut a, Token(1), Interest::READABLE));
  println!("deregister: {:?}", poll.registry().deregister(&mut a));
  Ok(())
}

Linux，epoll（3 次相同）：

poll 1: Token(1) readable=true read_closed=true
poll 2: 0 event(s)
reregister: Ok(())
deregister: Ok(())

Linux，强制 poll(2)（3 次相同）：

poll 1: Token(1) readable=true read_closed=true
poll 2: 0 event(s)
reregister: Err(Kind(NotFound))
deregister: Err(Kind(NotFound))

strace 里第一次 ppoll 返回 revents=POLLIN|POLLHUP|POLLRDHUP，紧接着一组 write(3) / read(3) / futex（内部注销），第二次 ppoll 只剩 {fd=3, events=POLLIN} 一项；程序结束 drop a 时没有再注销。对端关闭之后，poll(2) 后备上的 reregister / deregister 会返回 NotFound，epoll 上不会，在 drop 路径之外处理这两个错误的代码要留意这个差别。

pending_removal 处理的是另一种时序：poll(2) 正在等时，别的线程注销了某个 fd。deregister_all 先把 fd 放进 pending_removal，再走 modify_fds；select 这一轮如果已经拿到了这个 fd 的 revents，就按 pending_removal 把它跳过，不把已注销的 token 交给用户。

超时与重试

poll 这个辅助函数把 Duration 换成毫秒：

        #[cfg(target_pointer_width = "32")]
        const MAX_SAFE_TIMEOUT: u128 = 1789569;
        #[cfg(not(target_pointer_width = "32"))]
        const MAX_SAFE_TIMEOUT: u128 = libc::c_int::MAX as u128;

        let timeout = timeout
            .map(|to| {
                // `Duration::as_millis` truncates, so round up. This avoids
                // turning sub-millisecond timeouts into a zero timeout, unless
                // the caller explicitly requests that by specifying a zero
                // timeout.
                let to_ms = to
                    .checked_add(Duration::from_nanos(999_999))
                    .unwrap_or(to)
                    .as_millis();
                cmp::min(MAX_SAFE_TIMEOUT, to_ms) as libc::c_int
            })
            .unwrap_or(-1);

不足 1ms 的超时向上取整，None 变成 -1（永远等）。
MAX_SAFE_TIMEOUT 在 32 位上是 libuv 用的 1789569（注释：2.6.37 之前的内核在 32 位上把大于 LONG_MAX / CONFIG_HZ 的超时当成无限），64 位上是 c_int::MAX。epoll 版本曾经删掉这个常量，导致 1.0.0 到 1.2.0 的约 24.8 天溢出问题；poll.rs 一直保留着它，所以没有受影响。来龙去脉在 Poll 与 Registry。
poll(2) 返回 EAGAIN 时重试（注释：「poll returns EAGAIN if we can retry it」），其他错误包括 EINTR 原样返回。

超时只在进 poll(2) 前算一次。上一节看到，被内部唤醒之后重新进 poll(2) 用的还是这个值；horizon / WASI 的单线程通道没有 fd，woken() 为真时直接把超时设成 0。

1.2.3 修的 UnixListener（#1998）

1.2.3 的 CHANGELOG：「Fixed an issue with the UDS listener not reregistering readiness correctly when using the poll(2) implementation」。PR #1998「Fix UDS listener re-arming with poll」只改了一行，在 src/net/uds/listener.rs 里：

     pub fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
-        sys::uds::listener::accept(&self.inner)
+        self.inner.do_io(sys::uds::listener::accept)
     }

1.2.2 及以前，UnixListener::accept 直接调 sys 层的 accept，不经过 do_io。在 epoll / kqueue 上 do_io 本来就是空的，没有区别；在 poll(2) 后备上，第一个连接的事件报出后 POLLIN 被摘掉，accept 读到 WouldBlock 也没人放回，之后再来多少连接都不会有事件。TcpListener::accept 一直是 self.inner.do_io(...)，所以 TCP 没有这个问题。

PR 说明写的是在 AIX（默认用 poll(2)）上通过 tokio 遇到的：「a first connection being accepted, being drained up to the point of hitting WouldBlock, and then subsequent connections are never accepted and instead stall」，并用 mio_unsupported_force_poll_poll 在 Linux 上复现；修法是「follow the pattern from the TcpListener」。PR 同时加了回归测试 unix_listener_reregisters_after_would_block。合并提交首次发布于 1.2.3。

复现：1.2.2 和 1.2.4 对照

src/bin/uds_rearm.rs：注册一个 UnixListener，两个客户端先后连接；每次连接后 poll 500ms，然后 accept 到 WouldBlock。同一份代码放进两个 crate，一个依赖 mio = "=1.2.2"，一个依赖 mio = "=1.2.4"：

use std::io::ErrorKind;
use std::time::Duration;

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

fn main() -> std::io::Result<()> {
  let path = std::env::temp_dir().join(format!("uds-rearm-{}.sock", std::process::id()));
  let mut listener = UnixListener::bind(&path)?;
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  poll.registry().register(&mut listener, Token(0), Interest::READABLE)?;

  for client in 1..=2 {
    let _c = std::os::unix::net::UnixStream::connect(&path)?;
    poll.poll(&mut events, Some(Duration::from_millis(500)))?;
    println!("client {client} connected -> poll: {} event(s)", events.iter().count());
    let mut accepted = 0;
    loop {
      match listener.accept() {
        Ok(_) => accepted += 1,
        Err(e) if e.kind() == ErrorKind::WouldBlock => break,
        Err(e) => return Err(e),
      }
    }
    println!("  accepted {accepted}, then WouldBlock");
  }
  std::fs::remove_file(&path)
}

mio 1.2.2，强制 poll(2)（3 次相同）：

client 1 connected -> poll: 1 event(s)
  accepted 1, then WouldBlock
client 2 connected -> poll: 0 event(s)
  accepted 1, then WouldBlock

mio 1.2.2 用 epoll、mio 1.2.4 用 epoll、mio 1.2.4 强制 poll(2)，三种都是（各 3 次）：

client 1 connected -> poll: 1 event(s)
  accepted 1, then WouldBlock
client 2 connected -> poll: 1 event(s)
  accepted 1, then WouldBlock

1.2.2 下第二个客户端其实已经连上了（程序不管有没有事件都去 accept，照样接到 1 个），只是 selector 不报。strace（-e trace=ppoll,eventfd2,accept4,read,write,connect），1.2.2 的关键几行：

accept4(3, {sa_family=AF_UNIX}, [110 => 2], SOCK_CLOEXEC|SOCK_NONBLOCK) = 6
accept4(3, 0xffffcaa483f0, [110], SOCK_CLOEXEC|SOCK_NONBLOCK) = -1 EAGAIN (Resource temporarily unavailable)
write(1, "  accepted 1, then WouldBlock\n", 30) = 30
connect(5, {sa_family=AF_UNIX, sun_path="/tmp/uds-rearm-3.sock"}, 24) = 0
ppoll([{fd=4, events=POLLIN}, {fd=3, events=POLLRDHUP}], 2, {tv_sec=0, tv_nsec=500000000}, NULL, 8) = 0 (Timeout)

1.2.4 的同一段：

accept4(3, {sa_family=AF_UNIX}, [110 => 2], SOCK_CLOEXEC|SOCK_NONBLOCK) = 6
accept4(3, 0xffffd71cc520, [110], SOCK_CLOEXEC|SOCK_NONBLOCK) = -1 EAGAIN (Resource temporarily unavailable)
write(4, "\1\0\0\0\0\0\0\0", 8)         = 8
read(4, "\1\0\0\0\0\0\0\0", 8)          = 8
write(1, "  accepted 1, then WouldBlock\n", 30) = 30
connect(5, {sa_family=AF_UNIX, sun_path="/tmp/uds-rearm-3.sock"}, 24) = 0
ppoll([{fd=4, events=POLLIN}, {fd=3, events=POLLIN|POLLRDHUP}], 2, {tv_sec=0, tv_nsec=500000000}, NULL, 8) = 1 ([{fd=3, revents=POLLIN}], left {tv_sec=0, tv_nsec=499999416})

（这里 listener 先创建，拿到 fd 3，内部 eventfd 是 fd 4。）差别正是 accept4 返回 EAGAIN 之后有没有那组 write(4) / read(4)：没有它，listener（fd 3）在第二次 ppoll 里只剩 POLLRDHUP，连接到了也看不见。1.2.4 的 src/net/ 里，绕过 do_io 直接调 sys 层的只剩构造函数（bind、connect、pair）和 only_v6 这类选项读取；这些类型本身见 UDP 与 Unix 域套接字。

代价

每个连接多少个系统调用

src/bin/echo_cost.rs 是一个最小的 echo 服务器：子进程里的客户端顺序建立 N 个连接，每个写 16 字节、读回 16 字节、等 2ms 后关闭；服务端用 mio 接受、读到 WouldBlock 或 EOF、回写、EOF 时 deregister。strace 只跟服务端进程（不加 -f）。

use std::collections::HashMap;
use std::io::{ErrorKind, Read, Write};

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

// client: N sequential connections, each sends 16 bytes, reads the echo, waits 2ms, closes
fn client(n: usize, addr: &str) -> std::io::Result<()> {
  for _ in 0..n {
    let mut c = std::net::TcpStream::connect(addr)?;
    c.write_all(&[b'x'; 16])?;
    let mut buf = [0u8; 16];
    c.read_exact(&mut buf)?;
    // keep the connection open briefly so the server sees WouldBlock before EOF
    std::thread::sleep(std::time::Duration::from_millis(2));
  }
  Ok(())
}

fn main() -> std::io::Result<()> {
  let args: Vec<String> = std::env::args().collect();
  if args[1] == "client" {
    return client(args[2].parse().unwrap(), &args[3]);
  }
  let n: usize = args[1].parse().unwrap();
  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(64);
  poll.registry().register(&mut listener, Token(0), Interest::READABLE)?;
  let mut child = std::process::Command::new(std::env::current_exe()?)
    .args(["client", &n.to_string(), &listener.local_addr()?.to_string()])
    .spawn()?;

  let mut conns: HashMap<usize, TcpStream> = HashMap::new();
  let (mut next, mut done, mut polls, mut read_wb) = (1, 0, 0, 0);
  while done < n {
    poll.poll(&mut events, None)?;
    polls += 1;
    for event in &events {
      if event.token() == Token(0) {
        loop {
          match listener.accept() {
            Ok((mut s, _)) => {
              poll.registry().register(&mut s, Token(next), Interest::READABLE)?;
              conns.insert(next, s);
              next += 1;
            }
            Err(e) if e.kind() == ErrorKind::WouldBlock => break,
            Err(e) => return Err(e),
          }
        }
        continue;
      }
      let key = event.token().0;
      let mut buf = [0u8; 64];
      loop {
        let s = conns.get_mut(&key).unwrap();
        match s.read(&mut buf) {
          Ok(0) => {
            poll.registry().deregister(s)?;
            conns.remove(&key);
            done += 1;
            break;
          }
          Ok(k) => s.write_all(&buf[..k])?,
          Err(e) if e.kind() == ErrorKind::WouldBlock => {
            read_wb += 1;
            break;
          }
          Err(e) => return Err(e),
        }
      }
    }
  }
  child.wait()?;
  eprintln!("connections={n} polls={polls} read_wouldblock={read_wb}");
  Ok(())
}

N=100，strace -c，只保留和 I/O 有关的行（grep -E 的模式顺带匹配到了 readlinkat）。epoll：

connections=100 polls=264 read_wouldblock=100
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
 11.89    0.002147           8       264           epoll_pwait
 10.48    0.001892           6       300       100 recvfrom
  9.93    0.001792           8       200       100 accept4
  9.36    0.001690          16       104           close
  8.11    0.001465          14       100           sendto
  7.45    0.001345           6       201           epoll_ctl
  0.70    0.000126          18         7           write
  0.00    0.000000           0         1           read
  0.00    0.000000           0         1           ppoll
  0.00    0.000000           0         1           readlinkat
------ ----------- ----------- --------- --------- ----------------
100.00    0.018054          14      1211       200 total

强制 poll(2)：

connections=100 polls=259 read_wouldblock=100
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
 16.45    0.003276          16       200       100 accept4
  7.82    0.001557           5       260           ppoll
  7.67    0.001528           3       409           write
  6.85    0.001364           3       402           futex
  6.71    0.001335           3       403           read
  6.23    0.001240          11       104           close
  5.88    0.001171           3       300       100 recvfrom
  5.75    0.001144          11       100           sendto
  0.00    0.000000           0         1           readlinkat
------ ----------- ----------- --------- --------- ----------------
100.00    0.019909           9      2211       200 total

N=100 和 N=200 各跑 3 次。除了等待调用（epoll_pwait / ppoll，取决于事件凑在一起的时机），其余计数 6 次完全一致，减掉启动时的固定部分，每个连接：

系统调用	epoll	强制 `poll(2)`	来自哪里
`accept4`	2	2	1 次成功，1 次 `EAGAIN`
`recvfrom`	3	3	16 字节、`EAGAIN`、EOF
`sendto`	1	1	回写
`close`	1	1	drop 连接
`epoll_ctl`	2	0	`ADD`、`DEL`
`write` / `read` / `futex`（内部通道）	0	4 / 4 / 4	`register`、listener 放回、连接放回、`deregister` 各一次 `modify_fds`
等待调用	约 2.5–2.8	约 2.4–2.8	随时机变化：epoll N=100 时 253–264 次、N=200 时 548–553 次；`poll(2)` 分别是 243–259 次和 546–557 次（程序打印的 `polls`）
合计（N=200 减 N=100，再除以 100）	11.9 左右	22 左右

poll(2) 后备每个连接多出的 10 个系统调用全是 modify_fds 的固定开销，register、放回、注销一视同仁，哪怕只有一个线程。客户端那 2ms 是为了让服务端在 EOF 之前先读到一次 WouldBlock：不加它，strace 拖慢了服务端，第二次 recvfrom 直接读到 EOF，「连接放回」那一次就没了（那组数据是 epoll 每连接 10 个、poll(2) 17 个）。不跑 strace 时，加不加这 2ms 都是每个连接一次 WouldBlock（程序打印的 read_wouldblock=100），所以有 2ms 的这一组更接近真实情况。

闲置的 fd 也要付钱

poll(2) 每次都把整张表交给内核，内核逐项检查；返回后 select 还要遍历 fd_data。epoll 的等待只和就绪的数量有关。src/bin/idle_cost.rs：注册 k 个从不收数据的 UDP socket，测 2000 次 poll(Some(0)) 的平均时间：

use std::time::{Duration, Instant};

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

// cost of one poll(0) with k idle registered sockets
fn main() -> std::io::Result<()> {
  let mut events = Events::with_capacity(64);
  for k in [1, 10, 100, 1000, 10000] {
    let mut poll = Poll::new()?;
    let mut socks = Vec::with_capacity(k);
    for i in 0..k {
      let mut s = UdpSocket::bind("127.0.0.1:0".parse().unwrap())?;
      poll.registry().register(&mut s, Token(i), Interest::READABLE)?;
      socks.push(s);
    }
    let rounds = 2000;
    let start = Instant::now();
    for _ in 0..rounds {
      poll.poll(&mut events, Some(Duration::ZERO))?;
    }
    let per = start.elapsed().as_nanos() / rounds;
    println!("{k:>6} idle sockets: {:>8.1} us per poll(0)", per as f64 / 1000.0);
  }
  Ok(())
}

epoll（3 次输出相同）：

     1 idle sockets:      0.2 us per poll(0)
    10 idle sockets:      0.2 us per poll(0)
   100 idle sockets:      0.2 us per poll(0)
  1000 idle sockets:      0.2 us per poll(0)
 10000 idle sockets:      0.2 us per poll(0)

强制 poll(2)，3 次：

     1 idle sockets:      0.2 us per poll(0)
    10 idle sockets:      0.4 us per poll(0)
   100 idle sockets:      3.0 us per poll(0)
  1000 idle sockets:     32.7 us per poll(0)
 10000 idle sockets:    656.0 us per poll(0)

     1 idle sockets:      0.2 us per poll(0)
    10 idle sockets:      0.4 us per poll(0)
   100 idle sockets:      3.0 us per poll(0)
  1000 idle sockets:     31.9 us per poll(0)
 10000 idle sockets:    615.3 us per poll(0)

     1 idle sockets:      0.2 us per poll(0)
    10 idle sockets:      0.4 us per poll(0)
   100 idle sockets:      2.9 us per poll(0)
  1000 idle sockets:     32.8 us per poll(0)
 10000 idle sockets:    642.8 us per poll(0)

这里没有事件，select 在 num_events == 0 时直接返回，不遍历 fd_data，量到的基本是内核扫描 pollfd 数组的时间：大致和 k 成正比，1 万个 fd 时一次空 poll 要 0.6ms 以上。

汇总

方面	epoll / kqueue	`poll(2)` 后备
`register`	一次 `epoll_ctl` / `kevent`	不进内核；`write` + `read` + `futex` 叫醒并让轮询线程让路
报出之后	什么都不做	摘掉关注位；`WouldBlock` 时再 `write` + `read` + `futex` 放回
一次 `poll`	和就绪数量有关	和注册数量成正比（内核扫数组，有事件时用户态再遍历 `HashMap`）
别的线程改关注	和 `poll` 并发	打断 `poll`，超时从头算
`Events::with_capacity`	一次最多取这么多	不限，`Vec` 自己扩容；顺序随机
对端关闭后 `reregister` / `deregister`	`Ok`	`NotFound`（已被内部注销）
`Poll` 能否交出一个 fd	能（`AsRawFd`）	不能（`impl AsRawFd for Poll` 的 cfg 排除了这些平台，见源码地图）

这些代价换来的是：AIX、Haiku、ESP-IDF 这类只有 poll(2) 的系统也能用 mio（以及建在它上面的 tokio），上层代码不用知道底下是水平触发。

❗IMPORTANT

动脑筋：同时关注读写，读到 WouldBlock 之后的那次 poll 能拿到什么？

src/bin/rw_rearm.rs：一条已连接、发送缓冲区空着（一直可写）的 TCP 连接，用 READABLE | WRITABLE 注册（tokio 的 PollEvented::new 就是这么注册的）。每一步之后 poll 100ms，打印每个事件的 r / w：

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

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

fn show(poll: &mut Poll, events: &mut Events, step: &str) -> std::io::Result<()> {
  poll.poll(events, Some(Duration::from_millis(100)))?;
  let got: Vec<String> = events.iter().map(|e| {
    format!("{}{}", if e.is_readable() { "r" } else { "" }, if e.is_writable() { "w" } else { "" })
  }).collect();
  println!("{step:<34} -> {got:?}");
  Ok(())
}

fn main() -> std::io::Result<()> {
  let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
  let mut peer = std::net::TcpStream::connect(listener.local_addr()?)?;
  let (s, _) = 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 | Interest::WRITABLE)?;
  show(&mut poll, &mut events, "1. register READABLE|WRITABLE")?;
  show(&mut poll, &mut events, "2. poll again")?;
  peer.write_all(b"x")?;
  show(&mut poll, &mut events, "3. peer writes 1 byte")?;
  let mut buf = [0u8; 16];
  stream.read(&mut buf)?;
  assert_eq!(stream.read(&mut buf).unwrap_err().kind(), ErrorKind::WouldBlock);
  show(&mut poll, &mut events, "4. read until WouldBlock")?;
  show(&mut poll, &mut events, "5. poll again")?;
  Ok(())
}

在 Linux epoll、macOS kqueue、Linux 强制 poll(2) 上，第 3 行和第 4 行各打印什么？

动脑筋答案