Rust in Everything
基于 mio 1.2.4

Source 与 IoSource

主线位置：注册主线的中段。Registry::register(&mut source, token, interest) 什么也不做，直接调 source.register(self, token, interest)；从这里到平台 selector 之间的那一层就是本页：IoSource 在 debug 下查 SelectorId，再交给 IoSourceState，最后才到 epoll_ctl / kevent。事件主线上也有本页的一段：用户读写 mio 的 socket 时都经过 IoSource::do_io，它在 poll(2) 后备实现上负责在 WouldBlock 之后重新武装。

flowchart TB
  reg["Registry::register"] --> src["S::register"]
  src --> boxed["Box T：转给内部"]
  src --> iosrc["IoSource T"]
  src --> srcfd["SourceFd"]
  iosrc --> selid{"debug：SelectorId"}
  selid -->|已关联| errnode["Err AlreadyExists"]
  selid -->|未关联| state{IoSourceState}
  state -->|"epoll / kqueue"| stateless["无状态：直接转发"]
  state -->|"poll(2) / event ports"| stateful["记下 fd、token、interest"]
  srcfd --> sel["selector.register"]
  stateless --> sel
  stateful --> sel

Poll、Registry 的结构和 register / reregister / deregister 的约定在 Poll 与 Registry，Token 和 Interest 怎么进内核在 Token、Interest 与 Event。这一页讲 Registry 和平台 selector 之间的那一层：event/source.rs（139 行，一大半是文档）、io_source.rs（311 行）、sys/unix/sourcefd.rs，以及 sys/unix/selector/ 下的两个 *_io_source.rs。

event::Source：三个方法

Source 是所有能注册的东西的共同接口：

pub trait Source {
    fn register(
        &mut self,
        registry: &Registry,
        token: Token,
        interests: Interest,
    ) -> io::Result<()>;

    fn reregister(
        &mut self,
        registry: &Registry,
        token: Token,
        interests: Interest,
    ) -> io::Result<()>;

    fn deregister(&mut self, registry: &Registry) -> io::Result<()>;
}

文档要求用户不要直接调这三个方法，而是调 Registry 上的同名方法；实现者「几乎总是」转给一个更底层的句柄。Registry::register 确实只是一层转发，多打一行 trace!：

    pub fn register<S>(&self, source: &mut S, token: Token, interests: Interest) -> io::Result<()>
    where
        S: event::Source + ?Sized,
    {
        trace!(
            "registering event source with poller: token={:?}, interests={:?}",
            token,
            interests
        );
        source.register(self, token, interests)
    }

方向值得注意：不是 Registry 去问 source 要 fd，而是 Registry 把自己交给 source，由 source 决定怎么注册。这样 Windows 上的 socket（要建 AFD 状态）、Unix 上的裸 fd（直接 epoll_ctl）、一个包着两个 socket 的用户类型，都能用同一个入口。

S: ?Sized 让 &mut dyn Source 也能传进来，所以 mio 不需要给 &mut T 实现 Source。源码里唯一的转发实现是给 Box<T> 的，T: Source + ?Sized，方便把 Box<dyn Source> 放进集合：

impl<T> Source for Box<T>
where
    T: Source + ?Sized,
{
    fn register(
        &mut self,
        registry: &Registry,
        token: Token,
        interests: Interest,
    ) -> io::Result<()> {
        (**self).register(registry, token, interests)
    }
    // ... reregister / deregister 同样转发
}
📝NOTE
没有蠢问题

问： SourceFd 的 register 什么都不改，epoll 也允许多个线程同时 epoll_ctl，为什么三个方法都要 &mut self？

答： 因为有的平台要在用户态记账。改成 &mut 的 #1180（0.7.0）提交说明写的是：「一般来说事件源不需要被并发地（重新/取消）注册。虽然在多数平台上这样做是可以的，但在需要用户态记账的平台上（比如 Windows）会让事情更复杂。」本页后面的 registered_io_source.rs 就是这种记账：register 要往 self 里写一个 Box<InternalState>。&mut self 让编译器保证同一个 source 不会被两个线程同时注册，实现里就不用加锁。

Box<T> 的这个实现原来更宽。

📝NOTE
冷知识：Source 曾经对所有 DerefMut 类型都有实现

0.7.0-alpha.1 里，event/source.rs 有一个覆盖实现：impl<T, S> Source for T where T: DerefMut<Target = S>, S: Source + ?Sized，Box、&mut、任何智能指针都自动是 Source。随后它被换成了现在只针对 Box<T> 的版本，提交说明是：「这会阻止包装类型同时实现 DerefMut 和 event::Source，而内部类型也实现 event::Source 时会导致冲突的 trait 实现。也许等特化稳定之后可以恢复。」以及「这是为 IoSource 实现 DerefMut 所必需的。」也就是说，下一节的 IoSource 既要 DerefMut 到内部的 std socket，又要自己实现 Source，两者在覆盖实现下冲突。这次改动随 #1236 合入，首次发布于 0.7.0。

IoSource：mio 的 socket 都包在里面

IoSource<T> 有三个字段：

pub struct IoSource<T> {
    state: IoSourceState,
    inner: T,
    #[cfg(debug_assertions)]
    selector_id: SelectorId,
}

inner 是一个 std 类型（std::net::TcpStream、std::os::unix::net::UnixListener……），state 是平台相关的注册状态（后面讲），selector_id 只在 debug 下存在。mio::net 里的每个类型都是它的薄包装，例如 TcpStream 的字段就是 inner: IoSource<net::TcpStream>，Read 的实现是 self.inner.do_io(|mut inner| inner.read(buf))。

结构体本身是 pub 的，文档也写着「IoSource 让任何 FD 或 socket 的包装都能注册到 Poll」，但 lib.rs 里只有 mod io_source;，没有 pub use，它在 1.x 里是 crate 私有的，用户代码写不出 mio::IoSource。用户自己的 fd 只能走 SourceFd（下文）。

Unix 上的 Source 实现要求 T: AsRawFd，每个方法先做 debug 检查，再交给 state：

impl<T> event::Source for IoSource<T>
where
    T: AsRawFd,
{
    fn register(
        &mut self,
        registry: &Registry,
        token: Token,
        interests: Interest,
    ) -> io::Result<()> {
        #[cfg(debug_assertions)]
        self.selector_id.associate(registry)?;
        self.state
            .register(registry, token, interests, self.inner.as_raw_fd())
    }

    fn reregister(
        &mut self,
        registry: &Registry,
        token: Token,
        interests: Interest,
    ) -> io::Result<()> {
        #[cfg(debug_assertions)]
        self.selector_id.check_association(registry)?;
        self.state
            .reregister(registry, token, interests, self.inner.as_raw_fd())
    }

    fn deregister(&mut self, registry: &Registry) -> io::Result<()> {
        #[cfg(debug_assertions)]
        self.selector_id.remove_association(registry)?;
        self.state.deregister(registry, self.inner.as_raw_fd())
    }
}

Windows 版本要求 T: AsRawSocket，WASI preview 1 版本绕过 state 直接调 selector，结构相同。

SelectorId：debug 下的一道检查

每个 selector 在 debug 下有一个编号。epoll 版本的 Selector::new 从一个全局计数器 NEXT_ID（从 1 开始）取号，try_clone 沿用同一个号（注释：「It's the same selector, so we use the same id.」）。kqueue、poll(2)、event ports 的 selector 都有同样的 id。所以 Registry::try_clone 出来的注册器和原来的 Poll 算同一个 selector，两个 Poll::new() 算两个。

SelectorId 是 IoSource 里记「我注册在哪个 selector 上」的一个 AtomicUsize，0 表示未关联。associate 在 register 时调用：

    fn associate(&self, registry: &Registry) -> io::Result<()> {
        let registry_id = registry.selector().id();
        let previous_id = self.id.swap(registry_id, Ordering::AcqRel);

        if previous_id == Self::UNASSOCIATED {
            Ok(())
        } else {
            Err(io::Error::new(
                io::ErrorKind::AlreadyExists,
                "I/O source already registered with a `Registry`",
            ))
        }
    }

它不区分「已经注册在这个 Poll 上」和「已经注册在另一个 Poll 上」，两种都返回同一个 AlreadyExists，而且错误在系统调用之前返回。check_association（reregister 用）区分未注册（NotFound）和注册在别处（AlreadyExists，「already registered with a different Registry」）；remove_association（deregister 用）在编号对不上时返回 NotFound。整个类型和三处调用都在 #[cfg(debug_assertions)] 下，release 里什么都不查，交给内核。

mio 自己的测试 tcp_register_multiple_event_loops 就是测这个，并且标了 #[cfg(debug_assertions)] // Check is only present when debug assertions are enabled.。

实验：一个 listener 注册到两个 Poll

依赖只有一行（实验 crate 的 Cargo.toml，后面各实验相同）：

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

src/bin/two_polls.rs：同一个 TcpListener 先注册到 poll1，再注册一次到 poll1，再注册到 poll2；另用 SourceFd 把它的 fd 注册到 poll3。然后连一次，三个 Poll 各等 200ms。

use std::net::TcpStream as StdStream;
use std::os::fd::AsRawFd;
use std::time::Duration;

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

fn show(label: &str, r: std::io::Result<()>) {
  match r {
    Ok(()) => println!("{label}: Ok"),
    Err(e) => println!("{label}: Err(kind={:?}) {e}", e.kind()),
  }
}

fn drain(name: &str, poll: &mut Poll, events: &mut Events) {
  poll.poll(events, Some(Duration::from_millis(200))).unwrap();
  let tokens: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  println!("{name}: {} event(s), tokens {tokens:?}", tokens.len());
}

fn main() {
  println!("debug_assertions = {}", cfg!(debug_assertions));
  let mut poll1 = Poll::new().unwrap();
  let mut poll2 = Poll::new().unwrap();
  let mut poll3 = Poll::new().unwrap();
  let mut events = Events::with_capacity(8);

  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap()).unwrap();
  let addr = listener.local_addr().unwrap();
  let fd = listener.as_raw_fd();

  show("register   -> poll1", poll1.registry().register(&mut listener, Token(1), Interest::READABLE));
  show("register   -> poll1 again", poll1.registry().register(&mut listener, Token(1), Interest::READABLE));
  show("register   -> poll2", poll2.registry().register(&mut listener, Token(2), Interest::READABLE));
  show("SourceFd   -> poll3", poll3.registry().register(&mut SourceFd(&fd), Token(3), Interest::READABLE));

  let _client = StdStream::connect(addr).unwrap();
  drain("poll1", &mut poll1, &mut events);
  drain("poll2", &mut poll2, &mut events);
  drain("poll3", &mut poll3, &mut events);
}

Linux 6.12.28（Apple container 里的 Alpine，aarch64 musl 静态链接）、rustc 1.99.0，debug 构建（cargo build）：

debug_assertions = true
register   -> poll1: Ok
register   -> poll1 again: Err(kind=AlreadyExists) I/O source already registered with a `Registry`
register   -> poll2: Err(kind=AlreadyExists) I/O source already registered with a `Registry`
SourceFd   -> poll3: Ok
poll1: 1 event(s), tokens [1]
poll2: 0 event(s), tokens []
poll3: 1 event(s), tokens [3]

同一平台，release 构建：

debug_assertions = false
register   -> poll1: Ok
register   -> poll1 again: Err(kind=AlreadyExists) File exists (os error 17)
register   -> poll2: Ok
SourceFd   -> poll3: Ok
poll1: 1 event(s), tokens [1]
poll2: 1 event(s), tokens [2]
poll3: 1 event(s), tokens [3]

M2 Max、macOS、rustc 1.99.0：debug 构建的输出和 Linux debug 逐字相同；release 构建只有一行不同：

register   -> poll1 again: Ok

用 strace 看两次 Linux 运行的系统调用（strace -e trace=epoll_create1,epoll_ctl,epoll_pwait,socket,bind,listen,connect，程序的标准输出重定向掉了）。debug：

epoll_create1(EPOLL_CLOEXEC)            = 3
epoll_create1(EPOLL_CLOEXEC)            = 4
epoll_create1(EPOLL_CLOEXEC)            = 5
socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, IPPROTO_IP) = 6
bind(6, {sa_family=AF_INET, sin_port=htons(0), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
listen(6, 128)                          = 0
epoll_ctl(3, EPOLL_CTL_ADD, 6, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
epoll_ctl(5, EPOLL_CTL_ADD, 6, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x3}) = 0
socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 7
connect(7, {sa_family=AF_INET, sin_port=htons(39275), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
epoll_pwait(3, [{events=EPOLLIN, data=0x1}], 8, 200, NULL, 8) = 1
epoll_pwait(4, [], 8, 200, NULL, 8)     = 0
epoll_pwait(5, [{events=EPOLLIN, data=0x3}], 8, 200, NULL, 8) = 1
+++ exited with 0 +++

release：

epoll_create1(EPOLL_CLOEXEC)            = 3
epoll_create1(EPOLL_CLOEXEC)            = 4
epoll_create1(EPOLL_CLOEXEC)            = 5
socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, IPPROTO_IP) = 6
bind(6, {sa_family=AF_INET, sin_port=htons(0), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
listen(6, 128)                          = 0
epoll_ctl(3, EPOLL_CTL_ADD, 6, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
epoll_ctl(3, EPOLL_CTL_ADD, 6, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = -1 EEXIST (File exists)
epoll_ctl(4, EPOLL_CTL_ADD, 6, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x2}) = 0
epoll_ctl(5, EPOLL_CTL_ADD, 6, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x3}) = 0
socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 7
connect(7, {sa_family=AF_INET, sin_port=htons(45471), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
epoll_pwait(3, [{events=EPOLLIN, data=0x1}], 8, 200, NULL, 8) = 1
epoll_pwait(4, [{events=EPOLLIN, data=0x2}], 8, 200, NULL, 8) = 1
epoll_pwait(5, [{events=EPOLLIN, data=0x3}], 8, 200, NULL, 8) = 1
+++ exited with 0 +++

每种构建各跑 3 次，除端口号外相同。逐条看：

  • debug 下两次失败都没有系统调用。 两条 register 失败的消息完全一样，epoll_ctl 只有两次：fd 3（poll1）和 fd 5（poll3）。poll2 的 epoll 实例（fd 4）里什么都没有，所以它等满 200ms、0 个事件。
  • release 下由内核决定。 对同一个 epoll 实例重复 EPOLL_CTL_ADD 得到 EEXIST，io::Error 把它映射成 AlreadyExists，消息是 File exists (os error 17)；把同一个 fd 加进另一个 epoll 实例是合法的，poll2 和 poll1 各收到一个事件。macOS 上 kqueue 对已存在的过滤器再做一次 EV_ADD 只是修改，所以连重复注册都返回 Ok。
  • SourceFd 不受检查。 它没有 SelectorId，debug 下照样把同一个 fd 加进了第三个 epoll 实例。

也就是说，这道检查抓的是「同一个 mio 对象注册了两次」这类用法错误，而且只在 debug 下抓。在 Linux release 下，同一个 fd 进两个 epoll 是合法的，两个 Poll 都会收到事件；但边沿触发加上两个事件循环各自读同一个 socket，几乎一定是 bug。

associate 失败也会改编号

associate 先 swap 再判断，失败时也已经把编号换成了新的 selector。remove_association 也一样，先换成 0 再判断。debug 下一次失败的 register 会留下后遗症（src/bin/quiz_reregister.rs：注册到 poll1，再试着注册到 poll2，然后对 poll1 做 reregister、deregister，最后连一次、看 poll1）：

  show("register   poll1", poll1.registry().register(&mut listener, Token(1), Interest::READABLE));
  show("register   poll2", poll2.registry().register(&mut listener, Token(2), Interest::READABLE));
  show("reregister poll1", poll1.registry().reregister(&mut listener, Token(1), Interest::READABLE));
  show("deregister poll1", poll1.registry().deregister(&mut listener));

  let _client = StdStream::connect(addr).unwrap();
  poll1.poll(&mut events, Some(Duration::from_millis(200))).unwrap();
  let tokens: Vec<usize> = events.iter().map(|e| e.token().0).collect();
  println!("poll1 after deregister: {} event(s), tokens {tokens:?}", tokens.len());

debug 构建，macOS 和 Linux 输出相同（各 3 次）：

debug_assertions = true
register   poll1: Ok
register   poll2: Err(kind=AlreadyExists) I/O source already registered with a `Registry`
reregister poll1: Err(kind=AlreadyExists) I/O source already registered with a different `Registry`
deregister poll1: Err(kind=NotFound) I/O source not registered with `Registry`
poll1 after deregister: 1 event(s), tokens [1]

release 构建：

debug_assertions = false
register   poll1: Ok
register   poll2: Ok
reregister poll1: Ok
deregister poll1: Ok
poll1 after deregister: 0 event(s), tokens []

debug 下失败的 register 把编号改成了 poll2 的，于是对真正注册所在的 poll1 做 reregister 被拒（「a different Registry」）；deregister 也被拒，并且因为 ? 提前返回，epoll_ctl(EPOLL_CTL_DEL) 根本没发出去，内核里的注册还在，poll1 照样收到事件。release 下 poll2 的注册本来就合法，后面一切正常。这只影响「已经犯了一次错的程序」在 debug 下的后续表现，不是 release 的问题；但它说明这道检查的目的是尽早报错，而不是维护一个精确的状态。

SourceFd：任意 fd

SourceFd 是一个借用 fd 的元组结构（需要 os-ext feature，在 mio::unix 下）：

#[derive(Debug)]
pub struct SourceFd<'a>(pub &'a RawFd);

impl<'a> event::Source for SourceFd<'a> {
    fn register(
        &mut self,
        registry: &Registry,
        token: Token,
        interests: Interest,
    ) -> io::Result<()> {
        registry.selector().register(*self.0, token, interests)
    }
    // ... reregister / deregister 同样直接调 selector
}

它和 IoSource 有三处不同：

  • 不拥有 fd。 文档说它「不管理任何生命周期相关的操作，例如在 drop 时关闭 FD」，预期的用法是每次注册前临时构造：registry.register(&mut SourceFd(&fd), ...)。
  • 没有 SelectorId。 上面的实验里它在 debug 下也能把同一个 fd 注册进第二个 Poll。
  • 不经过 IoSourceState。 直接调 selector().register。在 epoll / kqueue 上没区别；在 poll(2) 后备实现上，它得不到下一节的重新武装（见文末的动脑筋）。

tokio 的 AsyncFd 就是用它注册的：AsyncFd::register_with_interest 里是 &mut SourceFd(&fd)，注册之后的事见 tokio 卷的反应器。

实验：一个 pipe 和一个普通文件

src/bin/sourcefd.rs：用 std::io::pipe() 建一个 pipe，mio 只看到它的读端 fd；再试一个普通文件 /etc/hosts。

use std::fs::File;
use std::io::{Read, Write};
use std::os::fd::AsRawFd;
use std::time::Duration;

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

fn show_events(label: &str, events: &Events) {
  println!("{label}: {} event(s)", events.iter().count());
  for e in events {
    println!(
      "  token={} readable={} writable={} read_closed={}",
      e.token().0,
      e.is_readable(),
      e.is_writable(),
      e.is_read_closed()
    );
  }
}

fn main() {
  let mut poll = Poll::new().unwrap();
  let mut events = Events::with_capacity(8);

  // A pipe from std: mio never sees the PipeReader type, only its fd.
  let (mut reader, mut writer) = std::io::pipe().unwrap();
  let rfd = reader.as_raw_fd();
  let r = poll.registry().register(&mut SourceFd(&rfd), Token(7), Interest::READABLE);
  println!("register pipe fd {rfd}: {r:?}");
  poll.poll(&mut events, Some(Duration::ZERO)).unwrap();
  show_events("before write", &events);
  writer.write_all(b"hi").unwrap();
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  show_events("after write", &events);
  let mut buf = [0u8; 2];
  reader.read_exact(&mut buf).unwrap();
  println!("read {:?}", std::str::from_utf8(&buf).unwrap());
  drop(writer);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  show_events("after writer dropped", &events);
  let r = poll.registry().deregister(&mut SourceFd(&rfd));
  println!("deregister pipe fd: {r:?}");

  // A regular file on the local filesystem.
  let file = File::open("/etc/hosts").unwrap();
  let ffd = file.as_raw_fd();
  let r = poll.registry().register(&mut SourceFd(&ffd), Token(8), Interest::READABLE);
  println!("register regular file fd {ffd}: {r:?}");
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  show_events("regular file", &events);
}

Linux（release，3 次相同）：

register pipe fd 4: Ok(())
before write: 0 event(s)
after write: 1 event(s)
  token=7 readable=true writable=false read_closed=false
read "hi"
after writer dropped: 1 event(s)
  token=7 readable=false writable=false read_closed=true
deregister pipe fd: Ok(())
register regular file fd 5: Err(Os { code: 1, kind: PermissionDenied, message: "Operation not permitted" })
regular file: 0 event(s)

对应的系统调用（strace -e trace=epoll_ctl,pipe2,openat）：

pipe2([4, 5], O_CLOEXEC)                = 0
epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x7}) = 0
epoll_ctl(3, EPOLL_CTL_DEL, 4, NULL)    = 0
openat(AT_FDCWD, "/etc/hosts", O_RDONLY|O_LARGEFILE|O_CLOEXEC) = 5
epoll_ctl(3, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x8}) = -1 EPERM (Operation not permitted)
+++ exited with 0 +++

macOS（release，3 次相同）：

register pipe fd 4: Ok(())
before write: 0 event(s)
after write: 1 event(s)
  token=7 readable=true writable=false read_closed=false
read "hi"
after writer dropped: 1 event(s)
  token=7 readable=true writable=false read_closed=true
deregister pipe fd: Ok(())
register regular file fd 5: Ok(())
regular file: 1 event(s)
  token=8 readable=true writable=false read_closed=false

pipe 在两个平台上都和 socket 一样工作：std 的 PipeReader 是阻塞的，mio 不关心，它只把 fd 交给内核（这里只读一次、读的是已经到达的 2 字节，所以不会阻塞；真实程序要先把 fd 设成非阻塞）。写端关闭后两边都报 read_closed，但 Linux 上 readable 是 false、macOS 上是 true，这是 EPOLLHUP 和 kqueue EV_EOF 的解释差异，见 Token、Interest 与 Event。

普通文件的结果完全取决于内核：epoll 拒绝普通文件（EPERM），kqueue 接受并且立刻报可读。SourceFd 文档里的「任何能注册到底层 OS selector 的 FD」说的就是这个：mio 不做任何过滤，能不能注册、事件长什么样，都是平台的事。

do_io 和两种 IoSourceState

IoSource 的另一半是 do_io：

    pub fn do_io<F, R>(&self, f: F) -> io::Result<R>
    where
        F: FnOnce(&T) -> io::Result<R>,
    {
        self.state.do_io(f, &self.inner)
    }

文档要求「所有 I/O 操作都必须经过 do_io」，否则在 WouldBlock 之后可能再也收不到事件。mio::net 的类型自己的 read / write / peek 都走它；想直接对 socket 做别的系统调用（比如 libc::recvmsg），要用 TcpStream::try_io 这类方法，它就是 self.inner.do_io(|_| f())。

IoSourceState 有两种 Unix 实现，由各 selector 文件在末尾选择：

selector文件IoSourceState适用平台（sys/unix/mod.rs 的 cfg）
epollselector/epoll.rsstateless_io_source.rsandroid、illumos、linux、redox
kqueueselector/kqueue.rsstateless_io_source.rsdragonfly、freebsd、netbsd、openbsd 和 Apple 各系统
event portsselector/event_ports.rsregistered_io_source.rssolaris
poll(2)selector/poll.rsregistered_io_source.rsaix、espidf、nuttx、fuchsia、haiku、hermit、hurd、nto、vita、cygwin、horizon、wasi（preview 2 起；preview 1 走 sys/wasip1）；或设了 --cfg mio_unsupported_force_poll_poll

Windows 不用这两个文件，它在 sys/windows/mod.rs 里有自己的 IoSourceState，形状和 registered_io_source.rs 一样：碰到 WouldBlock 就对 sock_state 重新注册，详见 Windows：IOCP 与 AFD。selector 怎么按平台选出来，见源码地图。

无状态：epoll 和 kqueue

文件开头一句话说明了为什么：「kqueue(2) 和 epoll(2) 都不需要持有任何用户态状态。」do_io 就是调用闭包：

pub(crate) struct IoSourceState;

impl IoSourceState {
    // ...
    pub(crate) fn do_io<T, F, R>(&self, f: F, io: &T) -> io::Result<R>
    where
        F: FnOnce(&T) -> io::Result<R>,
    {
        // We don't hold state, so we can just call the function and
        // return.
        f(io)
    }
    // register / reregister / deregister: "Pass through, we don't have any state."
}

IoSourceState 是零大小类型，do_io 内联之后什么也不剩。IoSource 是私有的，量不到，但可以量包着它的 mio::net::TcpStream（src/bin/sizes.rs，打印 size_of::<std::net::TcpStream>() 和 size_of::<mio::net::TcpStream>()）：

构建std::net::TcpStreammio::net::TcpStream
macOS release44
macOS debug416
Linux release44
Linux release，--cfg mio_unsupported_force_poll_poll416

release 下的 epoll / kqueue，mio 的 socket 就是一个 fd；debug 多出来的 12 字节是 8 字节的 SelectorId 加对齐；poll(2) 后备下多出来的是 Option<Box<InternalState>> 的一个指针加对齐。

边沿触发的内核自己会在下一次状态变化时再报一次，用户只要读到 WouldBlock（事件主线的要求），不需要 mio 做任何事。

有状态：poll(2) 和 event ports

registered_io_source.rs 开头一句是「poll(2) 和 event_port(2) 都需要持有每个 fd 的状态」：

pub(crate) struct IoSourceState {
    inner: Option<Box<InternalState>>,
}

struct InternalState {
    selector: Selector,
    token: Token,
    interests: Interest,
    fd: RawFd,
    shared_record: Arc<RegistrationRecord>,
}

register 先看 inner 是否已有值（有就 AlreadyExists，这个检查 release 下也在），然后 try_clone 一份 selector，调 register_internal，把 selector、token、interest、fd 都存进 Box。do_io 用它们在 WouldBlock 时重新注册：

    pub(crate) fn do_io<T, F, R>(&self, f: F, io: &T) -> io::Result<R>
    where
        F: FnOnce(&T) -> io::Result<R>,
    {
        let result = f(io);

        if let Err(err) = &result {
            if err.kind() == io::ErrorKind::WouldBlock {
                self.inner.as_ref().map_or(Ok(()), |state| {
                    state
                        .selector
                        .reregister(state.fd, state.token, state.interests)
                })?;
            }
        }

        result
    }

原因是这两种机制都不是边沿触发：

  • poll(2) 是水平触发，只要还可读就每次都报。mio 为了给出和 epoll 一样的「边沿」语义，在 select 里报出一个事件后把这一位从关注集合里摘掉（poll_fd.events &= !poll_fd.revents，旁边的注释写着「IoSourceState 的 do_io 包装会用 reregister 把关注加回去」）。用户读到 WouldBlock，说明数据读干净了，do_io 把关注加回去，下一次数据到达才会再报。
  • event ports 是一次性的：event_ports.rs 的注释写着「event ports 在投递之后会解除 fd 的关联」，同样要重新关联。

InternalState 还实现了 Drop：没 deregister 就 drop 时，它自己调一次 selector.deregister(fd)，因为 poll(2) 的关注集合在用户态，fd 关闭不会让它自动消失。RegistrationRecord 是和 selector 共享的一个标志，selector 遇到 POLLHUP / POLLERR 时会在内部先注销这个 fd，标志防止 drop 时再注销一次。

摘掉、加回的完整过程、跨线程注册时怎么唤醒正在 poll 的线程、1.2.3 修的 UDS listener 问题，都在 poll(2) 后备实现。这里只需要记住：do_io 是为这两个平台存在的，在 epoll 和 kqueue 上它是空的；而 SourceFd 不经过它。

❗IMPORTANT
动脑筋：用 SourceFd 注册一个 pipe，只读不读到 WouldBlock，第二轮还能收到事件吗？

src/bin/sourcefd_rearm.rs：pipe 读端用 SourceFd 注册 READABLE。每一轮写 1 字节、poll 100ms、数事件、读 1 字节（读完正好读空，但没有再读一次去碰 WouldBlock）；第 3 轮开始前手动 reregister 一次。

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

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

fn main() {
  let mut poll = Poll::new().unwrap();
  let mut events = Events::with_capacity(8);
  let (mut reader, mut writer) = std::io::pipe().unwrap();
  let fd = reader.as_raw_fd();
  poll.registry().register(&mut SourceFd(&fd), Token(7), Interest::READABLE).unwrap();

  let mut buf = [0u8; 1];
  for round in 1..=3 {
    if round == 3 {
      poll.registry().reregister(&mut SourceFd(&fd), Token(7), Interest::READABLE).unwrap();
      println!("(reregister)");
    }
    writer.write_all(b"x").unwrap();
    poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
    println!("round {round}: write 1 byte, poll -> {} event(s)", events.iter().count());
    reader.read_exact(&mut buf).unwrap(); // the byte is there either way
  }
}

跑三种：macOS（kqueue）、Linux（epoll）、Linux 上加 RUSTFLAGS="--cfg mio_unsupported_force_poll_poll" 强制用 poll(2)。都是 release 构建。三种情况下「round 2」那一行各是几个事件？

动脑筋答案

macOS 和 Linux（epoll）输出相同：

round 1: write 1 byte, poll -> 1 event(s)
round 2: write 1 byte, poll -> 1 event(s)
(reregister)
round 3: write 1 byte, poll -> 1 event(s)

Linux，--cfg mio_unsupported_force_poll_poll：

round 1: write 1 byte, poll -> 1 event(s)
round 2: write 1 byte, poll -> 0 event(s)
(reregister)
round 3: write 1 byte, poll -> 1 event(s)

三种各跑 3 次，结果不变。

  • epoll / kqueue 是 1。 边沿触发的「边」是新数据到达这个动作，不是「读到过 WouldBlock」。第二轮写入是一次新的到达，内核再报一次。读不到 WouldBlock 的真正风险在另一种情况：一次事件之后数据没读完，之后又没有新数据来，就再也不会有通知。
  • poll(2) 是 0。 第一轮报出事件后，selector 已经把 POLLIN 从这个 fd 的关注里摘掉了。SourceFd 的读写不经过 do_io，没人把它加回去，第二轮的数据到了也不报（字节其实在 pipe 里，read_exact 照样读到）。第 3 轮前手动 reregister，poll(2) 的关注恢复，又能收到事件。

所以在 poll(2) 后备实现上，SourceFd 的用户要自己承担 do_io 的职责：每次读写碰到 WouldBlock（这个例子里是每次处理完事件）就 reregister。用 mio 自己的 TcpStream 等类型就不用管，它们的读写都经过 do_io。

小结

  • event::Source 只有 register / reregister / deregister 三个方法，都要 &mut self（#1180，为了需要用户态记账的平台）。Registry 的同名方法只是转发；S: ?Sized 让 &mut dyn Source 直接可用，转发实现只有 Box<T> 一个（#1236 把对所有 DerefMut 的覆盖实现收窄成了它）。
  • IoSource<T> 包着 std 类型，mio::net 的类型都建在它上面；它在 1.x 里是 crate 私有的。
  • debug 下 SelectorId 让同一个 IoSource 的第二次 register 在系统调用之前就失败，消息固定是 I/O source already registered with a \Registry`，不分同一个还是另一个 Poll。release 下不查：Linux 上同一 epoll 重复添加得到 File exists (os error 17)，加进另一个 Poll 成功、两边都收到事件；macOS 上两种都成功。associate失败时也改了编号，之后对原Poll的reregister/deregister` 在 debug 下都被拒。
  • SourceFd(&fd) 借用任意 fd，直接调 selector，没有 SelectorId 也没有 do_io。能不能注册由内核决定：普通文件在 epoll 上是 EPERM，在 kqueue 上成功且立刻可读。
  • IoSourceState 在 epoll / kqueue 上是零大小的直通；在 poll(2) 和 event ports 上记下 selector、token、interest、fd，do_io 碰到 WouldBlock 就 reregister，因为这两种机制报一次之后关注就没了。实测 poll(2) 后备下 SourceFd 第二轮收不到事件，epoll / kqueue 上能收到。

💬 关联讨论

发起讨论
加载中...