Rust in Everything
基于 mio 1.2.4

Waker

主线位置：事件主线的旁路。一个线程阻塞在 Poll::poll（epoll_wait / kevent）里，另一个线程调 Waker::wake，让内核给前者送来一个带 waker token 的可读事件，poll 就返回了。它不处理任何 I/O，唯一的作用是让 poll 提前返回。

flowchart TB
  newnode["Waker::new"] --> chk{"debug：has_waker"}
  chk -->|已是 true| boom["panic"]
  chk -->|false| plat{平台}
  plat -->|Linux| efd["eventfd 注册进 epoll"]
  plat -->|"macOS / BSD"| kq["复制 kqueue 句柄"]
  plat -->|"poll(2) 后备"| pl["复制 selector"]
  plat -->|其他 Unix| pipenode["pipe 读端注册"]
  wake["wake()"] --> w1["write 8 字节"]
  wake --> w2["kevent NOTE_TRIGGER"]
  wake --> w3["记 token，叫醒 poll"]
  wake --> w4["write 1 字节"]

代码在 waker.rs（96 行，几乎全是文档）和 sys/unix/waker/ 下的四个文件，加上 kqueue、poll(2) selector 里各一小段。Poll 和 Registry 的结构在 Poll 与 Registry。tokio 的 I/O 驱动只建一个 Waker，token 是 TOKEN_WAKEUP（Token(0)），它怎么用见 tokio 卷反应器的「两个保留的 token」。

一个薄壳

Waker 只包着一个 sys::Waker：

#[derive(Debug)]
pub struct Waker {
    inner: sys::Waker,
}

impl Waker {
    /// Create a new `Waker`.
    pub fn new(registry: &Registry, token: Token) -> io::Result<Waker> {
        #[cfg(debug_assertions)]
        registry.register_waker();
        sys::Waker::new(registry.selector(), token).map(|inner| Waker { inner })
    }

    /// Wake up the [`Poll`] associated with this `Waker`.
    ///
    /// [`Poll`]: struct.Poll.html
    pub fn wake(&self) -> io::Result<()> {
        self.inner.wake()
    }
}

wake 只要 &self，Waker 是 Send + Sync，跨线程共享就用 Arc<Waker>（文档的例子就是这么写的）。它不实现 event::Source，也不经过 IoSource：各平台实现在 new 里直接调 selector。文档对它有两条约定：事件只在 Waker 还活着时保证送达；一个 Poll 只能有一个活跃的 Waker，多个时「会发生什么是未指定的」。horizon 和 WASI 上 lib.rs 不导出 Waker。

一个 Poll 一个 Waker

Waker::new 在 debug 下先调 Registry::register_waker：

    pub(crate) fn register_waker(&self) {
        assert!(
            !self.has_waker.swap(true, Ordering::AcqRel),
            "Only a single `Waker` can be active per `Poll` instance"
        );
    }

has_waker 是 Registry 里的一个 Arc<AtomicBool>，只在 all(debug_assertions, not(any(target_os = "wasi", target_os = "horizon"))) 下存在。Registry::try_clone 克隆的是这个 Arc，所以从同一个 Poll 克隆出来的所有注册器共用一个标志。违反时是 assert!，即 panic，不是返回错误；release 下整段不存在，不检查。

还有一点源码里看得出来：Waker 没有实现 Drop，没有任何地方把 has_waker 改回 false。所以「活跃的」实际上是「曾经建过的」：drop 掉第一个 Waker 再建一个，debug 下照样 panic。mio 的测试 using_multiple_wakers_panics 和紧随其后的 ..._different_cloned_registries 只测了「同时存在两个」。

实验：建第二个 Waker

实验 crate 的依赖：

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

src/bin/two_wakers.rs：建 waker1（Token(10)），在 catch_unwind 里建 waker2（Token(11)）。panic 了就 drop waker1 再建一个；没 panic 就做几组 wake 再 poll。

use std::panic::{catch_unwind, AssertUnwindSafe};
use std::time::Duration;

use mio::{Events, Poll, Token, Waker};

fn tokens(poll: &mut Poll, events: &mut Events) -> Vec<usize> {
  poll.poll(events, Some(Duration::from_millis(100))).unwrap();
  events.iter().map(|e| e.token().0).collect()
}

fn main() {
  println!("debug_assertions = {}", cfg!(debug_assertions));
  let mut poll = Poll::new().unwrap();
  let mut events = Events::with_capacity(8);
  let w1 = Waker::new(poll.registry(), Token(10)).unwrap();
  println!("waker1 (token 10) created");

  let second = catch_unwind(AssertUnwindSafe(|| Waker::new(poll.registry(), Token(11))));
  let w2 = match second {
    Ok(Ok(w2)) => {
      println!("waker2 (token 11) created");
      w2
    }
    Ok(Err(e)) => panic!("unexpected error {e}"),
    Err(_) => {
      println!("waker2: panicked");
      drop(w1);
      println!("waker1 dropped, trying again");
      let third = catch_unwind(AssertUnwindSafe(|| Waker::new(poll.registry(), Token(12))));
      println!("waker3: {}", if third.is_err() { "panicked" } else { "created" });
      return;
    }
  };

  w1.wake().unwrap();
  println!("w1.wake()            -> tokens {:?}", tokens(&mut poll, &mut events));
  w2.wake().unwrap();
  println!("w2.wake()            -> tokens {:?}", tokens(&mut poll, &mut events));
  w1.wake().unwrap();
  w2.wake().unwrap();
  println!("w1.wake(); w2.wake() -> tokens {:?}", tokens(&mut poll, &mut events));
  w2.wake().unwrap();
  w1.wake().unwrap();
  println!("w2.wake(); w1.wake() -> tokens {:?}", tokens(&mut poll, &mut events));
}

M2 Max、macOS、rustc 1.99.0，debug 构建，标准输出和标准错误合在一起：

debug_assertions = true
waker1 (token 10) created

thread 'main' (2010332) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/mio-1.2.4/src/poll.rs:727:9:
Only a single `Waker` can be active per `Poll` instance
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
waker2: panicked
waker1 dropped, trying again

thread 'main' (2010332) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/mio-1.2.4/src/poll.rs:727:9:
Only a single `Waker` can be active per `Poll` instance
waker3: panicked

Linux 6.12.28（Apple container 里的 Alpine，aarch64 musl 静态链接）、rustc 1.99.0 的 debug 构建，标准输出的四行和上面相同，标准错误里是同样的两条 panic 消息（poll.rs:727:9，线程号是 1）。drop 掉 waker1 之后的 waker3 也 panic 了，印证了上面「标志从不复位」。

release 构建两个 Waker 都建成功，之后几组 wake 的结果在 Linux 和 macOS 上不一样，留作文末的动脑筋。

📝NOTE
冷知识：从克隆的 Registry 建第二个 Waker，曾经查不出来

检查本身是 #1329（0.7.5）加的，那时标志是每个平台 Selector 里的一个 AtomicBool，Selector::try_clone 把当时的值复制进新的 AtomicBool。#1698 指出：「目前 AtomicBool 在 Selector 上，Selector::try_clone 时被克隆，但之后就不再同步了。所以你可以 try_clone 这个 Registry 再注册第二个 Waker。」#1706（0.8.9）把标志从各平台的 Selector 挪到平台无关的 Registry 上，换成 Arc<AtomicBool>，提交说明是「并且在所有 Registry 之间正确共享」，同时加了上面那个 using_multiple_wakers_panics_different_cloned_registries 测试。

各平台的实现

sys::Waker 是哪个文件，由 sys/unix/mod.rs 里一组 #[cfg_attr(..., path = "waker/...")] 决定（下表的「平台」一列照抄 cfg）。用 poll(2) 的平台例外：那里的 sys::Waker 是 selector/poll.rs 里自己定义的类型，cfg 选出来的文件只当它内部的唤醒通道。

实现文件平台wake() 做什么
eventfdwaker/eventfd.rsandroid、espidf、nuttx、fuchsia、hermit、illumos、linux（除非 --cfg mio_unsupported_force_waker_pipe）write 8 字节的 1u64
selectorwaker/selector.rsdragonfly、freebsd、netbsd、openbsd、Apple 各系统（kqueue，EVFILT_USER）；solaris（event ports，port_send）。都要求没设 mio_unsupported_force_waker_pipe 和 mio_unsupported_force_poll_poll调 Selector::wake(token)
pipewaker/pipe.rsaix、haiku、hurd、nto、redox、vita、cygwin；设了 force_poll_poll 的 kqueue 平台和 solaris；或设了 force_waker_pipewrite 1 字节
poll(2) selector 的selector/poll.rs所有用 poll(2) selector 的平台把 token 记进 pending_wake_token，再用上面某一种叫醒 poll
single_threadedwaker/single_threaded.rshorizon、WASI preview 2（只作 poll(2) selector 的内部通道）置一个 AtomicBool，下次 poll 用 0 超时

几个容易看漏的地方：

  • redox 用 epoll selector，但 waker 是 pipe。
  • kqueue 的 waker 和 poll(2) 不能一起用（cfg 旁的注释：「kqueue(2) based waker doesn't work with poll(2)」），所以 macOS 上设了 force_poll_poll 时内部通道换成 pipe。Linux 上设了它，内部通道仍是 eventfd（eventfd 那一行的 cfg 不排除 force_poll_poll），下面的实验会看到。
  • single_threaded 的文档说它用于「只有一个控制线程的系统」，不可能去叫醒一个已经阻塞在 select 里的线程，只能让下一次 select 不阻塞。

Linux：eventfd

Waker::new 建一个非阻塞的 eventfd，按 READABLE 注册进 epoll；wake 往里写 1：

    pub(crate) fn new(selector: &Selector, token: Token) -> io::Result<Waker> {
        let waker = Waker::new_unregistered()?;
        selector.register(waker.fd.as_raw_fd(), token, Interest::READABLE)?;
        Ok(waker)
    }

    pub(crate) fn new_unregistered() -> io::Result<Waker> {
        #[cfg(not(target_os = "espidf"))]
        let flags = libc::EFD_CLOEXEC | libc::EFD_NONBLOCK;
        // ...
        let fd = syscall!(eventfd(0, flags))?;
        let file = unsafe { File::from_raw_fd(fd) };
        Ok(Waker { fd: file })
    }

    #[allow(clippy::unused_io_amount)] // Don't care about partial writes.
    pub(crate) fn wake(&self) -> io::Result<()> {
        // The epoll emulation on some illumos systems currently requires
        // the eventfd to be read before an edge-triggered read event is
        // generated.
        // See https://www.illumos.org/issues/16700.
        #[cfg(target_os = "illumos")]
        self.reset()?;

        let buf: [u8; 8] = 1u64.to_ne_bytes();
        match (&self.fd).write(&buf) {
            Ok(_) => Ok(()),
            Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => {
                // Writing only blocks if the counter is going to overflow.
                // So we'll reset the counter to 0 and wake it again.
                self.reset()?;
                self.wake()
            }
            Err(err) => Err(err),
        }
    }

eventfd 是内核里的一个 64 位计数器：写 8 字节就是加上这个数，读 8 字节返回当前值并清零（文件开头的注释也这么说）。值得问的是：谁来读？ 整个文件里只有 reset 读它，而 reset 只在三处被调用：

  1. illumos 上每次 wake 之前（绕开那边 epoll 模拟的 bug）；
  2. write 返回 WouldBlock，也就是计数器再加就要溢出时；
  3. ack_and_reset，注释写着「Only used by the poll(2) implementation」。

所以在 Linux 的 epoll 路径上，计数器在正常运行中从来不被读。这没有问题：注册时带了 EPOLLET，每次 write 都会让 eventfd 唤醒它的等待队列，epoll 就把它重新放进就绪列表，不管计数器之前是不是已经大于 0。用户在 poll 之后也不需要（也没办法）去读这个 fd。

实验：wake 三次，poll 几次

src/bin/multi_wake.rs：连续 wake 三次再 poll；不 wake 再 poll 一次（0 超时）；再 wake 一次、poll；最后让另一个线程 100ms 后 wake，主线程 poll(None)。Linux 上顺便从 /proc/self/fdinfo 读出 eventfd 的计数（读 fdinfo 不会动计数器）。带参数 quiet 时不读 fdinfo，用来跑 strace。

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

use mio::{Events, Poll, Token, Waker};

// Linux only: the eventfd counter, read from /proc/self/fdinfo (no read(2) on the eventfd).
#[cfg(target_os = "linux")]
fn eventfd_counts() -> String {
  let mut out = Vec::new();
  for entry in std::fs::read_dir("/proc/self/fd").unwrap() {
    let entry = entry.unwrap();
    let Ok(link) = std::fs::read_link(entry.path()) else { continue };
    if link.to_string_lossy() != "anon_inode:[eventfd]" {
      continue;
    }
    let fd = entry.file_name().to_string_lossy().into_owned();
    let info = std::fs::read_to_string(format!("/proc/self/fdinfo/{fd}")).unwrap();
    let count = info.lines().find_map(|l| l.strip_prefix("eventfd-count:")).unwrap().trim().to_string();
    out.push(format!("fd {fd} count=0x{count}"));
  }
  out.join(", ")
}

#[cfg(not(target_os = "linux"))]
fn eventfd_counts() -> String {
  "n/a".to_string()
}

fn main() {
  let quiet = std::env::args().nth(1).as_deref() == Some("quiet");
  let mut log: Vec<String> = Vec::new();
  let mut note = |s: String| log.push(s);
  let counts = || if quiet { String::from("-") } else { eventfd_counts() };

  let mut poll = Poll::new().unwrap();
  let mut events = Events::with_capacity(8);
  let waker = Arc::new(Waker::new(poll.registry(), Token(10)).unwrap());
  note(format!("after Waker::new: eventfd {}", counts()));

  for _ in 0..3 {
    waker.wake().unwrap();
  }
  note(format!("after 3 x wake(): eventfd {}", counts()));
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  let t: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  note(format!("poll #1 (100ms): {} event(s) {t:?}, readable={}", t.len(), events.iter().all(|e| e.is_readable())));
  note(format!("after poll #1: eventfd {}", counts()));
  poll.poll(&mut events, Some(Duration::ZERO)).unwrap();
  note(format!("poll #2 (0ms, no wake): {} event(s)", events.iter().count()));
  waker.wake().unwrap();
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  note(format!("wake() once more, poll #3: {} event(s)", events.iter().count()));
  note(format!("after poll #3: eventfd {}", counts()));

  let w = Arc::clone(&waker);
  let start = Instant::now();
  let handle = std::thread::spawn(move || {
    std::thread::sleep(Duration::from_millis(100));
    w.wake().unwrap();
  });
  poll.poll(&mut events, None).unwrap();
  let t: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  note(format!("poll #4 (None), woken from another thread: {t:?} after >= 100ms: {}", start.elapsed() >= Duration::from_millis(100)));
  handle.join().unwrap();
  for line in log {
    println!("{line}");
  }
}

Linux，release：

after Waker::new: eventfd fd 4 count=0x0
after 3 x wake(): eventfd fd 4 count=0x3
poll #1 (100ms): 1 event(s) [10], readable=true
after poll #1: eventfd fd 4 count=0x3
poll #2 (0ms, no wake): 0 event(s)
wake() once more, poll #3: 1 event(s)
after poll #3: eventfd fd 4 count=0x4
poll #4 (None), woken from another thread: [10] after >= 100ms: true

strace -f -e trace=epoll_create1,epoll_ctl,epoll_pwait,eventfd2,read,write,ppoll,poll,pipe2，带 quiet 参数：

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
eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)   = 4
epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0xa}) = 0
write(4, "\1\0\0\0\0\0\0\0", 8)         = 8
write(4, "\1\0\0\0\0\0\0\0", 8)         = 8
write(4, "\1\0\0\0\0\0\0\0", 8)         = 8
epoll_pwait(3, [{events=EPOLLIN, data=0xa}], 8, 100, NULL, 8) = 1
epoll_pwait(3, [], 8, 0, NULL, 8)       = 0
write(4, "\1\0\0\0\0\0\0\0", 8)         = 8
epoll_pwait(3, [{events=EPOLLIN, data=0xa}], 8, 100, NULL, 8) = 1
strace: Process 4 attached
[pid     3] epoll_pwait(3 <unfinished ...>
[pid     4] write(4, "\1\0\0\0\0\0\0\0", 8 <unfinished ...>
[pid     3] <... epoll_pwait resumed>, [{events=EPOLLIN, data=0xa}], 8, -1, NULL, 8) = 1
[pid     4] <... write resumed>)        = 8
[pid     4] +++ exited with 0 +++
write(1, "after Waker::new: eventfd -\n", 28) = 28
write(1, "after 3 x wake(): eventfd -\n", 28) = 28
write(1, "poll #1 (100ms): 1 event(s) [10]"..., 48) = 48
write(1, "after poll #1: eventfd -\n", 25) = 25
write(1, "poll #2 (0ms, no wake): 0 event("..., 35) = 35
write(1, "wake() once more, poll #3: 1 eve"..., 38) = 38
write(1, "after poll #3: eventfd -\n", 25) = 25
write(1, "poll #4 (None), woken from anoth"..., 69) = 69
+++ exited with 0 +++

第一行 ppoll 是 Rust 标准库启动时检查 0、1、2 号 fd 是否打开，与 mio 无关；最后几行 write(1, ...) 是程序结尾统一打印的输出。中间就是 Waker 的全部：

  • 创建：eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)，然后一次 epoll_ctl(EPOLL_CTL_ADD)，标志和普通 socket 完全一样（EPOLLIN|EPOLLRDHUP|EPOLLET），data=0xa 就是 Token(10)。
  • wake：一次 write(4, "\1\0\0\0\0\0\0\0", 8)，小端的 1u64。
  • 三次 wake 一个事件：epoll 的就绪列表里一个 fd 只有一项，三次写入合成一个 EPOLLIN。
  • 没有 read：整个过程里 read 一次都没出现，fdinfo 里的计数从 3 涨到 4，poll 之后也不归零。poll #2 没有 wake 就没有事件，poll #3 前一次新的 write 又产生了一个事件，说明不清零也不妨碍下一次唤醒。
  • 跨线程：线程 3 阻塞在 epoll_pwait(..., -1, ...) 里，线程 4 的 write 让它返回。（mio 调的是 epoll_wait，strace 里是 epoll_pwait，那是 musl 在 aarch64 上的实现方式，见 epoll。）

macOS（release）的输出除了 fdinfo 显示 n/a，事件数完全相同：3 次 wake 后 1 个事件，不 wake 时 0 个，再 wake 1 个，跨线程唤醒成功。三种配置（macOS、Linux、Linux 强制 poll(2)）各跑 3 次，输出不变。

📝NOTE
没有蠢问题

问： 计数器从来不清零，一直加下去不会溢出吗？

答： eventfd 的计数上限是 0xfffffffffffffffe。写入会让它超过上限时，非阻塞的 eventfd 返回 EAGAIN，也就是 wake 里的 WouldBlock 分支：先 reset 读一次清零，再重试。每次 wake 只加 1，要 wake 约 1.8×10¹⁹ 次才会走到这个分支，所以实际上计数器就是一直涨。这个分支存在是为了正确，不是因为会经常发生。pipe 版本的同一个分支更现实：pipe 缓冲区满了（Linux 默认 64 KiB，即 65536 次没被读走的 wake）就会 WouldBlock，它调 empty 把 pipe 读空再写。

poll(2) 后备：在这里清零

用 RUSTFLAGS="--cfg mio_unsupported_force_poll_poll" 重新编译同一个程序。Linux，release：

after Waker::new: eventfd fd 3 count=0x0
after 3 x wake(): eventfd fd 3 count=0x3
poll #1 (100ms): 1 event(s) [10], readable=true
after poll #1: eventfd fd 3 count=0x0
poll #2 (0ms, no wake): 0 event(s)
wake() once more, poll #3: 1 event(s)
after poll #3: eventfd fd 3 count=0x0
poll #4 (None), woken from another thread: [10] after >= 100ms: true

strace（同样的参数）：

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(3, "\1\0\0\0\0\0\0\0", 8)         = 8
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
ppoll([{fd=3, events=POLLIN}], 1, {tv_sec=0, tv_nsec=100000000}, NULL, 8) = 1 ([{fd=3, revents=POLLIN}], left {tv_sec=0, tv_nsec=99999291})
read(3, "\3\0\0\0\0\0\0\0", 8)          = 8
ppoll([{fd=3, events=POLLIN}], 1, {tv_sec=0, tv_nsec=0}, NULL, 8) = 0 (Timeout)
write(3, "\1\0\0\0\0\0\0\0", 8)         = 8
ppoll([{fd=3, events=POLLIN}], 1, {tv_sec=0, tv_nsec=100000000}, NULL, 8) = 1 ([{fd=3, revents=POLLIN}], left {tv_sec=0, tv_nsec=99999500})
read(3, "\1\0\0\0\0\0\0\0", 8)          = 8
strace: Process 4 attached
[pid     3] ppoll([{fd=3, events=POLLIN}], 1, NULL, NULL, 8 <unfinished ...>
[pid     4] write(3, "\1\0\0\0\0\0\0\0", 8 <unfinished ...>
[pid     3] <... ppoll resumed>)        = 1 ([{fd=3, revents=POLLIN}])
[pid     4] <... write resumed>)        = 8
[pid     3] read(3, "\1\0\0\0\0\0\0\0", 8) = 8
[pid     4] +++ exited with 0 +++
write(1, "after Waker::new: eventfd -\n", 28) = 28
write(1, "after 3 x wake(): eventfd -\n", 28) = 28
write(1, "poll #1 (100ms): 1 event(s) [10]"..., 48) = 48
write(1, "after poll #1: eventfd -\n", 25) = 25
write(1, "poll #2 (0ms, no wake): 0 event("..., 35) = 35
write(1, "wake() once more, poll #3: 1 eve"..., 38) = 38
write(1, "after poll #3: eventfd -\n", 25) = 25
write(1, "poll #4 (None), woken from anoth"..., 69) = 69
+++ exited with 0 +++

这里的 eventfd 不是 Waker 自己的：poll(2) selector 在创建时就建了一个内部唤醒通道（notify_waker，没有 epoll 实例，所以也没有 epoll_ctl），放在 poll_fds 的第 0 项。用户的 Waker::new 只复制一份 selector 句柄，wake 调 SelectorState::wake：

    pub fn wake(&self, token: Token) -> io::Result<()> {
        self.pending_wake_token.lock().unwrap().replace(token);
        self.notify_waker.wake()
    }

select 发现第 0 项有 revents，就调 ack_and_reset，也就是那次 read(3, "\3\0\0\0\0\0\0\0", 8)：一次读出 3、计数归零；然后取出 pending_wake_token，合成一个事件。水平触发的 poll(2) 必须读空，否则下一次 poll 会立刻返回。pending_wake_token 是一个 Option<Token>，只装得下一个 token；它为什么这样设计、内部通道还用来做什么（跨线程注册时叫醒正在 poll 的线程），在 poll(2) 后备实现。

❗IMPORTANT
动脑筋：release 下给同一个 Poll 建两个 Waker，先 w1.wake() 再 w2.wake()，poll 得到哪些 token？

用上面的 two_wakers.rs，release 构建（debug 下第二个 Waker 会 panic，见前文）。waker1 是 Token(10)，waker2 是 Token(11)。程序的最后两行分别是「先 w1 后 w2」和「先 w2 后 w1」之后 poll 拿到的 token 列表。在 Linux（epoll）、macOS（kqueue）、Linux 强制 poll(2) 三种情况下，这两行各是什么？先读下一节再猜也可以。

macOS 与 BSD：EVFILT_USER

kqueue 平台不需要额外的 fd。waker/selector.rs 的 new 复制一份 kqueue 句柄（dup 出来的 fd 指向同一个 kqueue），在 FreeBSD 和三个 BSD 上再做一次 setup_waker；wake 调 kqueue selector 的 wake：

    pub fn wake(&self, token: Token) -> io::Result<()> {
        let mut kevent = kevent!(
            0,
            libc::EVFILT_USER,
            libc::EV_ADD | libc::EV_ONESHOT,
            token.0
        );
        kevent.fflags = libc::NOTE_TRIGGER;

        let kq = self.kq.as_raw_fd();
        syscall!(kevent(kq, &kevent, 1, ptr::null_mut(), 0, ptr::null()))?;
        Ok(())
    }

一次 kevent 调用，changelist 里一项：标识符（ident）固定是 0，过滤器 EVFILT_USER（用户触发的事件），EV_ADD | EV_ONESHOT（没有就添加，送达一次后自动删除），NOTE_TRIGGER（立即触发），token 放在 udata。EVFILT_USER 事件在 is_readable 里被当作可读（注释：「在用 eventfd 或 unix pipe 的平台上它会产生一个可读事件，所以这里也伪造一个」），所以用户看到的和 Linux 一样。

这几个参数直接决定了两个行为：

  • 多次 wake 一个事件。 同一个 kqueue 里 (ident, filter) 只能有一个 knote。几次 wake 都落在 (0, EVFILT_USER) 这同一个 knote 上，只是反复触发它，kevent 等待时只取到一个事件。上面 macOS 的 multi_wake 实测也是 1 个。
  • 不需要清零。 EV_ONESHOT 让 knote 在送达后自己消失，下次 wake 再 EV_ADD 一个新的。

这一节没有在 macOS 上跟踪系统调用，kevent 的参数只来自源码。

1.2.3 的 BSD 改动

CHANGELOG 的 1.2.3 一节有一条：「NetBSD、OpenBSD 和 DragonflyBSD 改用基于 EVFILT_USER 的 Waker，FreeBSD 和 macOS 早已在用」（#1993）。这次改动其实由三个 PR 接力完成，都在 1.2.3 发布：

  1. #1987「Avoid Waker setup for kqueue selector」：以前 Waker::new 先用 EV_ADD | EV_CLEAR | EV_RECEIPT 注册一个 EVFILT_USER 监听，每次 wake 再用 EV_ADD | EV_RECEIPT 触发并检查回执。说明里写「如果用 EV_ONESHOT，其实不需要这一步」，于是删掉了 setup，wake 改成现在的 EV_ADD | EV_ONESHOT，也不再要回执。
  2. #1988：「既然 kqueue 的实现不再需要任何 setup，两个实现就一样了」，把 kqueue 和 event ports 的 waker 合并成现在的 waker/selector.rs。
  3. #1993「Fix Waker for FreeBSD and switch DragonflyBSD, NetBSD and OpenBSD to EVFILT_USER based waker」，三个提交。第一个把 setup 加回给 FreeBSD：「在 Darwin（macOS、iOS 等）上不需要 setup……但在 FreeBSD 上不行。FreeBSD 上必须先建立这个机制，事件才会被送达；没有的话，提交的事件会被静默忽略。」对 DragonflyBSD、NetBSD、OpenBSD「不确定，所以也一并恢复了」。后两个把 NetBSD、OpenBSD 和 DragonflyBSD 从 pipe 切到 EVFILT_USER，理由都是一句「它比基于 pipe 的版本快」。

所以现在 setup_waker 的 cfg 是 dragonfly、freebsd、netbsd、openbsd，源码注释写着「For Darwin (macOS, iOS, etc.) we don't need any setup」。

源码里还有一处没跟上：is_readable 把 EVFILT_USER 当可读的那段 cfg，至今只列了 freebsd 和 Apple 各系统，没有加上 1.2.3 新切过来的 dragonfly、netbsd、openbsd。按源码读，这三个系统上 waker 事件的 is_readable() 会是 false（token() 不受影响，按 token 判断的代码不受影响）。mio 的 CI 在这几个目标上只做 cargo hack check，不跑测试，tests/waker.rs 里 assert!(event.is_readable()) 在那里没有执行过。我没有这几个系统可以验证，这里只是源码上的观察。

动脑筋答案

Linux（epoll），release，3 次相同：

debug_assertions = false
waker1 (token 10) created
waker2 (token 11) created
w1.wake()            -> tokens [10]
w2.wake()            -> tokens [11]
w1.wake(); w2.wake() -> tokens [10, 11]
w2.wake(); w1.wake() -> tokens [11, 10]

macOS（kqueue），release，3 次相同：

debug_assertions = false
waker1 (token 10) created
waker2 (token 11) created
w1.wake()            -> tokens [10]
w2.wake()            -> tokens [11]
w1.wake(); w2.wake() -> tokens [11]
w2.wake(); w1.wake() -> tokens [10]

Linux 强制 poll(2)，release：输出和 macOS 逐字相同（3 次）。

  • epoll：两个都到。 两个 Waker 是两个 eventfd，各自注册进 epoll，互不相干。strace 看得很清楚：

    eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)   = 4
epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0xa}) = 0
write(1, "waker1 (token 10) created\n", 26) = 26
eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK)   = 5
epoll_ctl(3, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0xb}) = 0

    后面 write(4, ...)、write(5, ...) 之后的 epoll_pwait 一次返回两项（data=0xa 和 data=0xb）。

  • kqueue：只剩最后一个。 两个 Waker 的 wake 都提交 (ident 0, EVFILT_USER)，是同一个 knote；第二次 EV_ADD 修改了已有的 knote，udata 被换成第二个 token。于是只有一个事件，token 是后 wake 的那个。

  • poll(2)：同样只剩最后一个。 pending_wake_token 是一个 Option<Token>，replace 覆盖掉前一个。

这就是文档说「多个 Waker 的行为未指定」的具体含义：在 Linux 上碰巧都能用，在 macOS 和 poll(2) 后备上会丢唤醒的 token。debug 下的 panic 把这个平台差异挡在了开发阶段。需要多个唤醒来源的程序，应该共享一个 Arc<Waker>，再用自己的队列或标志区分是谁叫醒的；tokio 就只有一个 TOKEN_WAKEUP。

小结

  • Waker 是 sys::Waker 的薄壳，wake(&self) 可以跨线程调用；它不走 event::Source，各平台在 new 里直接操作 selector。
  • 一个 Poll 一个 Waker：debug 下 Registry 里共享的 has_waker（Arc<AtomicBool>，#1706 起在 Registry 上）让第二个 Waker::new panic，消息是 Only a single \Waker` can be active per `Poll` instance`；标志从不复位，drop 第一个再建也 panic。release 不检查。
  • Linux：eventfd2(0, EFD_CLOEXEC|EFD_NONBLOCK) 加一次 EPOLL_CTL_ADD；wake 是 write 8 字节的 1。计数器在 epoll 路径上从不读回（只在溢出和 illumos 上 reset），实测 3 次 wake 后计数 3、poll 后仍是 3；边沿触发保证每次新的 write 都产生一个新事件。
  • macOS / BSD：kevent 提交 (0, EVFILT_USER, EV_ADD|EV_ONESHOT, NOTE_TRIGGER)，token 在 udata；BSD 先要 setup_waker（#1993 为 FreeBSD 加回）。其他 Unix 用 pipe，poll(2) 后备用 pending_wake_token 加内部通道，并在每次 poll 后读空它。
  • 几次 wake 再 poll，三种实现都只给一个事件。两个 Waker 同时 wake：epoll 给两个 token，kqueue 和 poll(2) 只给后一个。

💬 关联讨论

发起讨论
加载中...