Rust in Everything
基于 mio 1.2.4

TCP

主线位置：两条主线的起点和终点。注册主线从一个已经是非阻塞的 fd 开始，这个 fd 由本页的 bind、accept、connect 造出来；事件主线的最后一步「读写到 WouldBlock」由本页的 Read / Write 实现完成。非阻塞 connect 是两条线交叉的地方：先注册 WRITABLE，等一个可写事件，再用 take_error 和 peer_addr 问内核连接的结果。

sequenceDiagram
  participant U as 用户代码
  participant M as mio
  participant K as 内核
  U->>M: TcpStream::connect
  M->>K: socket 加 NONBLOCK
  M->>K: connect
  K-->>M: EINPROGRESS
  M-->>U: Ok 未连上的流
  U->>K: register WRITABLE
  U->>K: poll
  K-->>U: 可写事件
  U->>K: take_error 即 SO_ERROR
  alt 连上
    K-->>U: None，peer_addr 有地址
  else 被拒
    K-->>U: ConnectionRefused
  end

代码在 net/tcp/listener.rs（310 行）、net/tcp/stream.rs（481 行），以及 Unix 上的 sys/unix/tcp.rs（139 行）和 sys/unix/net.rs（221 行）。两个公开文件里文档占了一大半。

结构：一个 IoSource 字段

TcpListener 和 TcpStream 都只有一个字段：

pub struct TcpStream {
    inner: IoSource<net::TcpStream>,
}

IoSource<T> 怎么实现 event::Source、debug 下怎么查 SelectorId，在 Source 与 IoSource。这一页只关心另一面：IoSource<T> 实现了 Deref<Target = T>，所以 mio 的方法大多是一行转发到 std：

mio 方法实现落到哪里
peer_addr、local_addr、shutdownself.inner.xxx()std，getpeername / getsockname / shutdown
set_nodelay、nodelay、set_ttl、ttlself.inner.xxx()std，setsockopt / getsockopt
take_errorself.inner.take_error()std，getsockopt(SO_ERROR)
read、write、peek、try_ioself.inner.do_io(...)std 的读写，外面包一层 do_io
TcpStream::connect、TcpListener::bind、acceptmio 自己写sys::tcp 里的 libc 调用

mio 自己写的只有最后一行：std 的 bind、connect、accept 会建出阻塞的 socket，或者干脆在 connect 里阻塞等待，这三处 mio 必须自己调 libc。选项类的方法全部借 std，Cargo.toml 的 [dependencies] 里只有 log、libc（Unix）、windows-sys（Windows）和 wasi，没有 socket2。std 没提供的选项（SO_REUSEPORT、keepalive 参数、收发缓冲区大小）mio 也不提供：0.8.0 删掉了可以设这些选项的 TcpSocket 类型（#1514），PR 的说明是「The socket2 crate provide all the functionality and more. Furthermore supporting all socket options is beyond the scope of Mio.」需要这些选项时，用 socket2 建好 socket，再用 from_std、from_raw_fd 或 From<OwnedFd> 交给 mio。tokio 的 TcpSocket 就是这样：内部是一个 socket2::Socket，连上或监听之后用 mio::net::TcpStream::from_raw_fd / TcpListener::from_raw_fd 转过去。

bind：四个系统调用

TcpListener::bind 的文档列了四步，代码一一对应：

    pub fn bind(addr: SocketAddr) -> io::Result<TcpListener> {
        let socket = new_for_addr(addr)?;
        #[cfg(any(unix, target_os = "hermit", target_os = "wasi"))]
        let listener = unsafe { TcpListener::from_raw_fd(socket) };
        // ...

        // On platforms with Berkeley-derived sockets, this allows to quickly
        // rebind a socket, without needing to wait for the OS to clean up the
        // previous one.
        //
        // On Windows, this allows rebinding sockets which are actively in use,
        // which allows “socket hijacking”, so we explicitly don't set it here.
        // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
        #[cfg(not(windows))]
        set_reuseaddr(&listener.inner, true)?;

        bind(&listener.inner, addr)?;
        // Use the same backlog value as the standard library.
        // <https://github.com/rust-lang/rust/blob/0028f344ce9f64766259577c998a1959ca1f6a0b/library/std/src/sys/net/connection/socket/mod.rs#L559-L571>
        let backlog = if cfg!(target_os = "horizon") {
            20
        } else if cfg!(target_os = "haiku") {
            32
        } else {
            128
        };
        listen(&listener.inner, backlog)?;
        Ok(listener)
    }

几处安排：

  • fd 先交给 TcpListener，再设选项。 from_raw_fd 之后 fd 归 TcpListener 所有，后面任何一步用 ? 返回错误时，listener 被 drop，fd 随之关闭，不会泄漏。
  • SO_REUSEADDR 只在非 Windows 上设。 Unix 上它让监听端口在上一个进程退出后马上能重新 bind，不用等 TIME_WAIT 过去；Windows 上同名选项的语义是「允许抢占正在使用的端口」，所以不设。
  • backlog 和 std 一致，大多数平台是 128，下面的冷知识讲它改过两次。

第一步 new_for_addr 按地址族选 AF_INET / AF_INET6，调 new_socket。这个函数是 mio 所有 Unix socket（TCP、UDP、Unix 域）共用的出生地，按平台分三段：

pub(crate) fn new_socket(domain: libc::c_int, socket_type: libc::c_int) -> io::Result<libc::c_int> {
    #[cfg(any(
        target_os = "android",
        // ...
        target_os = "linux",
        // ...
    ))]
    let socket_type = socket_type | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC;
    // ...

    let socket = syscall!(socket(domain, socket_type, 0))?;

    // Mimic `libstd` and set `SO_NOSIGPIPE` on apple systems.
    #[cfg(any(
        target_os = "ios",
        target_os = "macos",
        // ...
    ))]
    if let Err(err) = syscall!(setsockopt(
        socket,
        libc::SOL_SOCKET,
        libc::SO_NOSIGPIPE,
        &1 as *const libc::c_int as *const libc::c_void,
        size_of::<libc::c_int>() as libc::socklen_t
    )) {
        let _ = syscall!(close(socket));
        return Err(err);
    }

    // Darwin (and others) doesn't have SOCK_NONBLOCK or SOCK_CLOEXEC.
    #[cfg(any(
        target_os = "aix",
        target_os = "ios",
        target_os = "macos",
        // ...
    ))]
    {
        if let Err(err) = syscall!(fcntl(socket, libc::F_SETFL, libc::O_NONBLOCK)) {
            let _ = syscall!(close(socket));
            return Err(err);
        }
        #[cfg(not(any(target_os = "espidf", target_os = "vita", target_os = "nto")))]
        if let Err(err) = syscall!(fcntl(socket, libc::F_SETFD, libc::FD_CLOEXEC)) {
            let _ = syscall!(close(socket));
            return Err(err);
        }
    }

    Ok(socket)
}

Linux 上非阻塞和 CLOEXEC 在 socket(2) 这一次调用里就设好了；macOS 没有这两个类型标志，要多三次调用：SO_NOSIGPIPE（对端关闭后写入时返回 EPIPE 而不是发 SIGPIPE 信号，和 std 一样）、F_SETFL O_NONBLOCK、F_SETFD FD_CLOEXEC。这里 fd 还是裸的 c_int，所以每一步失败都要手动 close。

步骤LinuxmacOS（按源码）
建 socketsocket(AF_INET, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0)socket(AF_INET, SOCK_STREAM, 0)
补标志—setsockopt(SO_NOSIGPIPE)、fcntl(F_SETFL, O_NONBLOCK)、fcntl(F_SETFD, FD_CLOEXEC)
复用地址setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)同左
绑定、监听bind、listen(fd, 128)同左

实验：bind、accept、set_nodelay

src/bin/tcp_bind.rs 建一个监听 socket，用 fcntl 和 getsockopt 读回它的标志；一个阻塞的 std 客户端连上来，poll 之后 accept 到 WouldBlock，读回新连接的标志，再 set_nodelay(true)。fd_flags 和 sockopt 是本实验 crate 里的两个小函数，分别用 F_GETFL / F_GETFD 和 getsockopt 读值：

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

use std::io::ErrorKind;
use std::os::fd::AsRawFd;

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

fn main() -> std::io::Result<()> {
  println!("os={}", std::env::consts::OS);
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);

  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
  let lfd = listener.as_raw_fd();
  println!("listener: {} reuseaddr={}", fd_flags(lfd), sockopt(lfd, libc::SOL_SOCKET, libc::SO_REUSEADDR) != 0);
  #[cfg(target_os = "macos")]
  println!("listener: nosigpipe={}", sockopt(lfd, libc::SOL_SOCKET, libc::SO_NOSIGPIPE) != 0);
  poll.registry().register(&mut listener, Token(0), Interest::READABLE)?;

  // a blocking std client; connect completes in the kernel backlog
  let _client = std::net::TcpStream::connect(listener.local_addr()?)?;

  poll.poll(&mut events, None)?;
  println!("poll: {} event(s)", events.iter().count());
  loop {
    match listener.accept() {
      Ok((stream, peer)) => {
        let fd = stream.as_raw_fd();
        println!("accepted {peer}: {}", fd_flags(fd));
        #[cfg(target_os = "macos")]
        println!("accepted: nosigpipe={}", sockopt(fd, libc::SOL_SOCKET, libc::SO_NOSIGPIPE) != 0);
        println!("nodelay before={}", stream.nodelay()?);
        stream.set_nodelay(true)?;
        println!("nodelay after={}", stream.nodelay()?);
      }
      Err(e) if e.kind() == ErrorKind::WouldBlock => {
        println!("accept: {e}");
        break;
      }
      Err(e) => return Err(e),
    }
  }
  Ok(())
}

M2 Max、macOS、rustc 1.99.0：

os=macos
listener: nonblock=true cloexec=true reuseaddr=true
listener: nosigpipe=true
poll: 1 event(s)
accepted 127.0.0.1:61825: nonblock=true cloexec=true
accepted: nosigpipe=true
nodelay before=false
nodelay after=true
accept: Resource temporarily unavailable (os error 35)

Linux 6.12.28（Apple container 里的 Alpine，aarch64 musl 静态链接）、rustc 1.99.0，用 strace -f -o ... -e trace=socket,setsockopt,bind,listen,epoll_ctl,epoll_pwait,connect,accept4,write 跑，程序自己的标准输出丢掉，只看 strace 文件（write(1, ...) 就是程序的 println!，当路标用；行首的 4 是线程号）：

4     write(1, "os=linux\n", 9)         = 9
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, IPPROTO_IP) = 4
4     setsockopt(4, SOL_SOCKET, SO_REUSEADDR, [1], 4) = 0
4     bind(4, {sa_family=AF_INET, sin_port=htons(0), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
4     listen(4, 128)                    = 0
4     write(1, "listener: nonblock=true cloexec="..., 52) = 52
4     epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 5
4     connect(5, {sa_family=AF_INET, sin_port=htons(41159), sin_addr=inet_addr("127.0.0.1")}, 16) = 0
4     epoll_pwait(3, [{events=EPOLLIN, data=0}], 8, -1, NULL, 8) = 1
4     write(1, "poll: 1 event(s)\n", 17) = 17
4     accept4(4, {sa_family=AF_INET, sin_port=htons(33720), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC|SOCK_NONBLOCK) = 6
4     write(1, "accepted 127.0.0.1:33720: nonblo"..., 53) = 53
4     write(1, "nodelay before=false\n", 21) = 21
4     setsockopt(6, SOL_TCP, TCP_NODELAY, [1], 4) = 0
4     write(1, "nodelay after=true\n", 19) = 19
4     accept4(4, 0xffffdcd68bc8, [128], SOCK_CLOEXEC|SOCK_NONBLOCK) = -1 EAGAIN (Resource temporarily unavailable)
4     write(1, "accept: Resource temporarily una"..., 55) = 55
4     +++ exited with 0 +++

Linux 上不加 strace 的输出和 macOS 相比，少了两行 nosigpipe，最后一行的错误码是 11（EAGAIN）而不是 35，其余各行（listener: nonblock=true cloexec=true reuseaddr=true、accepted ...: nonblock=true cloexec=true、两行 nodelay）相同。两个平台各跑 3 次，除端口外输出相同；strace 也跑了 3 次，除端口和地址外相同。

  • 第 2 到 5 行就是 bind 的四步。socket 的类型里直接带 SOCK_CLOEXEC|SOCK_NONBLOCK，之后没有 fcntl。
  • 第 8、9 行是 std 客户端：socket 只带 SOCK_CLOEXEC，阻塞的 connect 返回 0。
  • 第 15 行：set_nodelay(true) 是一次 setsockopt(SOL_TCP, TCP_NODELAY)，由 std 发出。nodelay() 用的 getsockopt 没在跟踪列表里。
📝NOTE
冷知识：backlog 从 1024 改到 -1，又改回 128

1.0.x 及以前，bind 里写的是 listen(&listener.inner, 1024)。1.1.0 的 #1896（PR）把它换成按平台取值的常量 LISTEN_BACKLOG_SIZE，想和 std 保持一致；PR 说明里提到「It is sometimes provided as a "magic" value, for example, -1」，于是 Linux、FreeBSD、OpenBSD 和 Apple 平台上这个常量是 -1，源码注释写着 Linux 上会被「Silently capped to /proc/sys/net/core/somaxconn」。可是这组值抄的是 std 给 UnixListener 用的。1.2.0 的 #1934（PR）改了回来：「For TCP we were using the values std lib used for UnixListener, now we're matching.」TCP 从此用 std TcpListener 的值，大多数平台 128，就是上面 bind 里那段 if cfg! 和 strace 第 5 行的 listen(4, 128)。

所以同一个 TcpListener::bind，在 Linux 上 mio 1.0 是 1024，1.1 是 -1（实验用的容器里 somaxconn 是 4096），1.2 是 128。tokio 的 TcpListener::bind 直接调 mio 的 bind（bind_addr），本站锁的 mio 1.2.1 也就是 128。Unix 域套接字仍然用 -1，见 UDP 与 Unix 域套接字。

accept：新连接一出生就是非阻塞的

TcpListener::accept 经过 do_io，里面调 sys::tcp::accept，再把 std 的流包成 TcpStream::from_std：

    pub fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
        self.inner.do_io(|inner| {
            sys::tcp::accept(inner).map(|(stream, addr)| (TcpStream::from_std(stream), addr))
        })
    }

sys::tcp::accept 分两条路，Linux 用 accept4(..., SOCK_CLOEXEC | SOCK_NONBLOCK)，macOS 用普通的 accept 加一次 fcntl(F_SETFD, FD_CLOEXEC)，非阻塞靠新 fd 从监听 socket 继承。两条路的选择、x86 Android 为什么是例外（#1445、#1450），见 一个事件循环 里的冷知识。上面的实验把两条路都验证了：Linux 第 12 行的 accept4 带着 SOCK_NONBLOCK；macOS 上新连接 nonblock=true，而 macOS 这条路上 mio 只设了 FD_CLOEXEC，非阻塞确实是继承来的。macOS 的输出还多一行 accepted: nosigpipe=true：accept 路径上没有任何 SO_NOSIGPIPE 的代码，这个选项也是从监听 socket 继承的。

accept 返回的地址由 to_socket_addr 从 sockaddr_storage 转成 std::net::SocketAddr，ss_family 既不是 AF_INET 也不是 AF_INET6 时返回 InvalidInput。

connect：立刻返回的 connect

std 的 TcpStream::connect 会阻塞到三次握手结束。mio 的 TcpStream::connect 先建一个非阻塞 socket，再调 sys::tcp::connect：

pub(crate) fn connect(socket: &net::TcpStream, addr: SocketAddr) -> io::Result<()> {
    let (raw_addr, raw_addr_length) = socket_addr(&addr);

    match syscall!(connect(
        socket.as_raw_fd(),
        raw_addr.as_ptr(),
        raw_addr_length
    )) {
        Err(err) if err.raw_os_error() != Some(libc::EINPROGRESS) => Err(err),
        _ => Ok(()),
    }
}

非阻塞 socket 上的 connect 发出 SYN 就返回，几乎总是 EINPROGRESS。mio 把 EINPROGRESS 和成功一样当成 Ok(())，其他错误原样返回。所以 TcpStream::connect 返回 Ok 只说明「连接已经开始」，不说明连上了。文档里写了判断连接结果的六步：

    ///  1. Call `TcpStream::connect`
    ///  2. Register the returned stream with at least [write interest].
    ///  3. Wait for a (writable) event.
    ///  4. Check `TcpStream::take_error`. If it returns an error, then
    ///     something went wrong. If it returns `Ok(None)`, then proceed to
    ///     step 5.
    ///  5. Check `TcpStream::peer_addr`. If it returns `libc::EINPROGRESS` or
    ///     `ErrorKind::NotConnected` it means the stream is not yet connected,
    ///     go back to step 3. If it returns an address it means the stream is
    ///     connected, go to step 6. If another error is returned something
    ///     went wrong.
    ///  6. Now the stream can be used.

take_error 读的是 SO_ERROR：内核把异步 connect 的失败原因存在这里，读一次就清零。peer_addr 是 getpeername，只有连上的 socket 才有对端地址。文档还提到 connect 本身也可能返回 WouldBlock，「it usually means there are insufficient entries in the routing cache」。

实验：连上一次，被拒一次

src/bin/tcp_connect.rs 对两个地址各走一遍六步：一个是正在监听的 std listener；另一个是先 bind 一个 listener 拿到端口、再立刻 drop 掉，这个端口上没人监听。raw 和 bools 与 Token、Interest 与 Event 的实验里相同，分别打印平台原始标志位和为真的 is_* 方法：

use std::net::SocketAddr;
use std::time::Duration;

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

fn try_connect(title: &str, addr: SocketAddr) -> std::io::Result<()> {
  println!("== {title}");
  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);

  let mut stream = TcpStream::connect(addr)?;
  println!("connect returned Ok");
  println!("before event: peer_addr={:?}", stream.peer_addr().map_err(|e| e.kind()));

  poll.registry().register(&mut stream, Token(0), Interest::WRITABLE)?;
  poll.poll(&mut events, Some(Duration::from_secs(1)))?;
  for e in &events {
    println!("event: {}  => {}", raw(e), bools(e));
  }
  println!("take_error #1 = {:?}", stream.take_error()?);
  println!("take_error #2 = {:?}", stream.take_error()?);
  println!("peer_addr = {:?}", stream.peer_addr());
  Ok(())
}

fn main() -> std::io::Result<()> {
  println!("os={}", std::env::consts::OS);
  let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
  try_connect("listening port", listener.local_addr()?)?;

  // bind then drop: the port is (almost certainly) closed now
  let closed = std::net::TcpListener::bind("127.0.0.1:0")?.local_addr()?;
  try_connect("closed port", closed)?;
  Ok(())
}

M2 Max、macOS、rustc 1.99.0：

os=macos
== listening port
connect returned Ok
before event: peer_addr=Err(NotConnected)
event: filter=WRITE flags=CLEAR fflags=0 data=146988 udata=0x0  => writable
take_error #1 = None
take_error #2 = None
peer_addr = Ok(127.0.0.1:61925)
== closed port
connect returned Ok
before event: peer_addr=Err(NotConnected)
event: filter=WRITE flags=EOF|CLEAR fflags=61 data=2048 udata=0x0  => writable error write_closed
take_error #1 = Some(Os { code: 61, kind: ConnectionRefused, message: "Connection refused" })
take_error #2 = None
peer_addr = Err(Os { code: 22, kind: InvalidInput, message: "Invalid argument" })

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

os=linux
== listening port
connect returned Ok
before event: peer_addr=Ok(127.0.0.1:35333)
event: events=OUT data=0x0  => writable
take_error #1 = None
take_error #2 = None
peer_addr = Ok(127.0.0.1:35333)
== closed port
connect returned Ok
before event: peer_addr=Err(NotConnected)
event: events=OUT|ERR|HUP data=0x0  => writable error read_closed write_closed
take_error #1 = Some(Os { code: 111, kind: ConnectionRefused, message: "Connection refused" })
take_error #2 = None
peer_addr = Err(Os { code: 107, kind: NotConnected, message: "Socket not connected" })

macOS 跑了 8 次、Linux 跑了 3 次，除端口外输出逐行相同。Linux 上的系统调用（strace -e trace=socket,connect,epoll_ctl,epoll_pwait,getsockopt,getpeername,write，3 次除端口和地址外相同）：

4     write(1, "os=linux\n", 9)         = 9
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 3
4     write(1, "== listening port\n", 18) = 18
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, IPPROTO_IP) = 5
4     connect(5, {sa_family=AF_INET, sin_port=htons(43595), sin_addr=inet_addr("127.0.0.1")}, 16) = -1 EINPROGRESS (Operation in progress)
4     write(1, "connect returned Ok\n", 20) = 20
4     getpeername(5, {sa_family=AF_INET, sin_port=htons(43595), sin_addr=inet_addr("127.0.0.1")}, [128 => 16]) = 0
4     write(1, "before event: peer_addr=Ok(127.0"..., 44) = 44
4     epoll_ctl(4, EPOLL_CTL_ADD, 5, {events=EPOLLOUT|EPOLLET, data=0}) = 0
4     epoll_pwait(4, [{events=EPOLLOUT, data=0}], 8, 1000, NULL, 8) = 1
4     write(1, "event: events=OUT data=0x0  => w"..., 40) = 40
4     getsockopt(5, SOL_SOCKET, SO_ERROR, [0], [4]) = 0
4     write(1, "take_error #1 = None\n", 21) = 21
4     getsockopt(5, SOL_SOCKET, SO_ERROR, [0], [4]) = 0
4     write(1, "take_error #2 = None\n", 21) = 21
4     getpeername(5, {sa_family=AF_INET, sin_port=htons(43595), sin_addr=inet_addr("127.0.0.1")}, [128 => 16]) = 0
4     write(1, "peer_addr = Ok(127.0.0.1:43595)\n", 32) = 32
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 4
4     write(1, "== closed port\n", 15)  = 15
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, IPPROTO_IP) = 5
4     connect(5, {sa_family=AF_INET, sin_port=htons(36965), sin_addr=inet_addr("127.0.0.1")}, 16) = -1 EINPROGRESS (Operation in progress)
4     write(1, "connect returned Ok\n", 20) = 20
4     getpeername(5, 0xfffffa95eb40, [128]) = -1 ENOTCONN (Socket not connected)
4     write(1, "before event: peer_addr=Err(NotC"..., 42) = 42
4     epoll_ctl(4, EPOLL_CTL_ADD, 5, {events=EPOLLOUT|EPOLLET, data=0}) = 0
4     epoll_pwait(4, [{events=EPOLLOUT|EPOLLERR|EPOLLHUP, data=0}], 8, 1000, NULL, 8) = 1
4     write(1, "event: events=OUT|ERR|HUP data=0"..., 79) = 79
4     getsockopt(5, SOL_SOCKET, SO_ERROR, [ECONNREFUSED], [4]) = 0
4     write(1, "take_error #1 = Some(Os { code: "..., 95) = 95
4     getsockopt(5, SOL_SOCKET, SO_ERROR, [0], [4]) = 0
4     write(1, "take_error #2 = None\n", 21) = 21
4     getpeername(5, 0xfffffa95eb40, [128]) = -1 ENOTCONN (Socket not connected)
4     write(1, "peer_addr = Err(Os { code: 107, "..., 87) = 87
4     +++ exited with 0 +++

对照六步读：

  • connect 两次都是 EINPROGRESS（strace 第 5、21 行），mio 都返回 Ok。连得上和连不上，在这一步看不出区别。
  • 注册 WRITABLE 时没有 EPOLLRDHUP（第 9、25 行）：EPOLLRDHUP 只跟着 READABLE 走，见 epoll。
  • 连上时，事件只有一个可写位：Linux EPOLLOUT，macOS 一个 EVFILT_WRITE。take_error 是 None，peer_addr 给出地址，到第 6 步。
  • 被拒时，可写事件照样会来，但带着错误位：Linux 是 EPOLLOUT|EPOLLERR|EPOLLHUP，macOS 是带 EV_EOF、fflags=61 的 EVFILT_WRITE。is_writable() 两边都为真，所以只看 is_writable 会以为连上了；take_error 第一次返回 ConnectionRefused（Linux 111、macOS 61，strace 第 28 行能看到内核交出的 ECONNREFUSED），第二次就是 None（第 30 行，SO_ERROR 读一次就清零）。各个 is_* 位的含义见 Token、Interest 与 Event。
  • 被拒之后的 peer_addr，两个平台的错误不同：Linux 仍是 NotConnected，macOS 是 InvalidInput（EINVAL）。
  • 事件来之前的 peer_addr：连本机正在监听的端口时，Linux 在 connect 返回后立刻 getpeername 已经成功了（第 7 行），macOS 是 NotConnected。

最后一条说明：在环回地址上，Linux 的握手在第一次 getpeername 之前就已经完成了，虽然 connect 返回的是 EINPROGRESS。这是时序上的巧合，不是保证；六步流程不依赖它，照样等可写事件。

📝NOTE
没有蠢问题

问： 第 5 步说 peer_addr 返回 NotConnected 表示「还没连上，回到第 3 步」。可上面 Linux 上被拒的连接，peer_addr 也是 NotConnected。按第 5 步，程序不就会永远等下去吗？

答： 所以第 4 步必须在第 5 步之前。被拒的连接，SO_ERROR 里存着 ECONNREFUSED，第 4 步的 take_error 先把它取出来，流程在那里就以错误结束，到不了第 5 步。顺序反过来，或者第 4 步的结果被丢掉（比如先调了一次 take_error 但没看返回值），SO_ERROR 已经清零，这时 Linux 上只剩一个和「正在连」一模一样的 NotConnected。macOS 上这时是 EINVAL，按第 5 步属于「another error」，倒是能当成失败。tokio 的 connect_mio 只做到第 4 步：等到可写，take_error() 有错误就返回，否则直接当作连上。

只注册 READABLE 又会怎样？下面这个问题的答案在页末。

❗IMPORTANT
动脑筋：只注册 READABLE，等得到连接结果吗？

把上面的程序改成 src/bin/connect_readable.rs：TcpStream::connect 之后只注册 Interest::READABLE，poll 一次，超时 1 秒，打印事件个数。对端连上之后不发任何数据。在 Linux 和 macOS 上，「listening port」和「closed port」两种情况各收到几个事件？

from_std：非阻塞由调用方负责

TcpStream::from_std 和 TcpListener::from_std 只是把 std 类型放进 IoSource：

    pub fn from_std(stream: net::TcpStream) -> TcpStream {
        TcpStream {
            inner: IoSource::new(stream),
        }
    }

文档说得很清楚：「The conversion assumes nothing about the underlying stream; it is left up to the user to set it in non-blocking mode.」mio 不检查、也不替你设。from_raw_fd、From<OwnedFd> 也是转到 from_std，同样不设。mio 内部的 accept 和 bind 也调 from_std，但它们交进去的 fd 已经是非阻塞的。

不设会怎样：注册、poll 一切正常，事件也会来；但事件主线最后一步「读到 WouldBlock」永远等不到，读干净之后的那次 read 在内核里阻塞，整个事件循环跟着停住。

实验：设与不设

src/bin/from_std.rs 用 std 的 listener accept 出一个阻塞的流，第一次先 set_nonblocking(true) 再 from_std，第二次直接 from_std。对端写 5 字节，poll 之后照边沿触发的规矩读到 WouldBlock。第二次永远读不到 WouldBlock，所以另起一个线程，1 秒后打印一行并退出进程：

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

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

fn run(set_nonblocking: bool) -> std::io::Result<()> {
  println!("== set_nonblocking({set_nonblocking})");
  let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
  let mut client = std::net::TcpStream::connect(listener.local_addr()?)?;
  let (std_stream, _) = listener.accept()?;
  if set_nonblocking {
    std_stream.set_nonblocking(true)?;
  }
  let mut stream = TcpStream::from_std(std_stream);

  let mut poll = Poll::new()?;
  let mut events = Events::with_capacity(8);
  poll.registry().register(&mut stream, Token(0), Interest::READABLE)?;
  client.write_all(b"hello")?;
  poll.poll(&mut events, None)?;
  println!("poll: {} event(s)", events.iter().count());

  let mut buf = [0u8; 64];
  loop {
    match stream.read(&mut buf) {
      Ok(n) => println!("read {n} byte(s)"),
      Err(e) if e.kind() == ErrorKind::WouldBlock => {
        println!("read: WouldBlock, back to poll");
        return Ok(());
      }
      Err(e) => return Err(e),
    }
  }
}

fn main() -> std::io::Result<()> {
  run(true)?;
  // guard: the blocking variant never returns on its own
  thread::spawn(|| {
    thread::sleep(Duration::from_secs(1));
    println!("guard: read still blocked after 1s, exiting");
    std::process::exit(0);
  });
  run(false)?;
  println!("unreachable");
  Ok(())
}

两个平台输出相同（各跑 3 次，逐字节相同）：

== set_nonblocking(true)
poll: 1 event(s)
read 5 byte(s)
read: WouldBlock, back to poll
== set_nonblocking(false)
poll: 1 event(s)
read 5 byte(s)
guard: read still blocked after 1s, exiting

Linux 上的系统调用（strace -f -o ... -e trace=socket,accept4,ioctl,epoll_ctl,epoll_pwait,recvfrom,write,exit_group；行首是线程号，4 是主线程，5 是看门狗线程）：

4     write(1, "== set_nonblocking(true)\n", 25) = 25
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 3
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 4
4     accept4(3, {sa_family=AF_INET, sin_port=htons(53384), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC) = 5
4     ioctl(5, FIONBIO, [1])            = 0
4     epoll_ctl(6, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
4     epoll_pwait(6, [{events=EPOLLIN, data=0}], 8, -1, NULL, 8) = 1
4     write(1, "poll: 1 event(s)\n", 17) = 17
4     recvfrom(5, "hello", 64, 0, NULL, NULL) = 5
4     write(1, "read 5 byte(s)\n", 15)  = 15
4     recvfrom(5, 0xffffc96fe260, 64, 0, NULL, NULL) = -1 EAGAIN (Resource temporarily unavailable)
4     write(1, "read: WouldBlock, back to poll\n", 31) = 31
4     write(1, "== set_nonblocking(false)\n", 26) = 26
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 3
4     socket(AF_INET, SOCK_STREAM|SOCK_CLOEXEC, IPPROTO_IP) = 4
4     accept4(3, {sa_family=AF_INET, sin_port=htons(56984), sin_addr=inet_addr("127.0.0.1")}, [128 => 16], SOCK_CLOEXEC) = 5
4     epoll_ctl(6, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
4     epoll_pwait(6, [{events=EPOLLIN, data=0}], 8, -1, NULL, 8) = 1
4     write(1, "poll: 1 event(s)\n", 17) = 17
4     recvfrom(5, "hello", 64, 0, NULL, NULL) = 5
4     write(1, "read 5 byte(s)\n", 15)  = 15
4     recvfrom(5 <unfinished ...>
5     write(1, "guard: read still blocked after "..., 44) = 44
5     exit_group(0)                     = ?
4     <... recvfrom resumed> <unfinished ...>) = ?
5     +++ exited with 0 +++
4     +++ exited with 0 +++
  • std 的 accept 在 Linux 上也是 accept4，但只带 SOCK_CLOEXEC（第 4、16 行），新连接是阻塞的。和上一个实验里 mio 自己的 accept4(..., SOCK_CLOEXEC|SOCK_NONBLOCK) 对比，差的就是这一个标志。
  • std 的 set_nonblocking(true) 在 Linux 上是 ioctl(FIONBIO)（第 5 行）。设过之后，第二次 recvfrom 返回 EAGAIN（第 11 行），循环回到 poll。
  • 没设的那次，注册和事件和前面一模一样（第 17、18 行），区别只在第二次 recvfrom：它没有返回，直到看门狗线程 exit_group 把进程结束（第 22 到 25 行）。事件循环里任何别的 socket 在这期间都得不到处理。

poll 拿到事件、第一次读成功，这两步都看不出 fd 是阻塞的，问题只在「读干净之后再读一次」时出现。这个坑在常见坑一章里还会单独出现。

读写：经过 do_io 转给 std

Read 和 Write 的每个方法都是一行：

impl Read for TcpStream {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.inner.do_io(|mut inner| inner.read(buf))
    }

    fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> io::Result<usize> {
        self.inner.do_io(|mut inner| inner.read_vectored(bufs))
    }
}

&TcpStream 也实现了 Read 和 Write（闭包里的 mut inner 是 &net::TcpStream，std 给 &TcpStream 实现了读写），所以一个共享引用就能同时读写。Linux 上 std 的读写落到 recvfrom / sendto（sendto 带 MSG_NOSIGNAL），见 一个事件循环 的 strace。

do_io 在 epoll 和 kqueue 上什么也不加，在 poll(2) 后备和 event ports 上负责 WouldBlock 之后重新武装，见 Source 与 IoSource。这就是 IoSource 的 Deref 上那句警告的由来：「Be careful when using this method. All I/O operations that may block must go through the [do_io] method.」Deref 让 self.inner.read(...) 也能编译，在 epoll 和 kqueue 上也能正常工作，只有在需要重新武装的平台上才会丢事件。想对 fd 做 mio 没包装的 I/O（比如 libc::recvmsg），要放进 try_io 的闭包里，它就是 self.inner.do_io(|_| f())。

📝NOTE
冷知识：peek 曾经绕过了 do_io

1.0.x 的 peek 是 self.inner.peek(buf)，经 Deref 直接调 std。1.1.0 的 #1895（PR，「fix: peek reregister after would block」）改成现在的样子，加了一行注释：

    pub fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
        // Need to re-register if `peek` returns `WouldBlock`
        // to ensure the socket will receive more events once it is ready again.
        self.inner.do_io(|inner| inner.peek(buf))
    }

问题先在 Windows 上被发现：Windows 的 selector 也要模拟边沿触发，报出可读事件后清掉可读关注。PR 的说明是「since peek is also a kind of special read, it also cancelled interest to readable event, hence cause the blocking peek bug on windows because interest is lost so blocking is forever」，随后又发现「under linux with cfg(mio_unsupported_force_poll_poll) peek would also have similiar problem so just write do_io on all platform instead」。CHANGELOG 1.1.0 的条目是「Fixed an issue where TcpStream::peek returned WouldBlock the events would stop working correctly」。1.2.3 的 UnixListener::accept（#1998）是同一类问题，见 怎么读这一卷 和 poll(2) 后备实现。

动脑筋答案

src/bin/connect_readable.rs 和 tcp_connect.rs 只差注册那一行（Interest::READABLE），每种情况 poll 一次后打印事件个数和事件。M2 Max、macOS、rustc 1.99.0：

os=macos
listening port: 0 event(s)
closed port: 1 event(s)
    filter=READ flags=EOF|CLEAR fflags=61 data=0 udata=0x0  => readable error read_closed

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

os=linux
listening port: 0 event(s)
closed port: 1 event(s)
    events=IN|ERR|HUP|RDHUP data=0x0  => readable error read_closed write_closed

两个平台各跑 3 次，输出相同。

  • 连上时 0 个事件。 连接建立只让 socket 变得可写，不会让它可读；对端不发数据，READABLE 就一直没有变化，poll 等满 1 秒超时返回。只注册 READABLE 的程序要等到对端第一次发数据才知道连上了，如果协议是客户端先说话，就会一直等下去。这就是文档第 2 步要求「at least write interest」的原因。
  • 被拒时 1 个事件。 失败会同时影响两个方向：Linux 给出 IN|ERR|HUP|RDHUP（只关注可读时也会报 EPOLLERR 和 EPOLLHUP，内核总是报告这两位），macOS 给出带 EV_EOF、fflags=61 的 EVFILT_READ。所以失败用 READABLE 也能发现，成功却不能。

小结

  • TcpListener / TcpStream 只有一个 IoSource<std 类型> 字段。选项、地址、take_error 经 Deref 直接用 std，mio 不依赖 socket2；0.8.0 删掉 TcpSocket（#1514），更多的选项交给 socket2。
  • bind 是 socket（Linux 带 SOCK_NONBLOCK|SOCK_CLOEXEC，macOS 事后补 SO_NOSIGPIPE 和两次 fcntl）、SO_REUSEADDR（Windows 不设）、bind、listen(128)。backlog 在 1.0 是 1024，1.1 是 -1，1.2 起和 std 一样是 128（#1896、#1934）。
  • accept 经过 do_io。Linux 用 accept4(SOCK_CLOEXEC|SOCK_NONBLOCK)；macOS 用 accept 加 FD_CLOEXEC，实测非阻塞和 SO_NOSIGPIPE 都从监听 socket 继承。
  • connect 把 EINPROGRESS 当成 Ok。连上还是被拒，要注册 WRITABLE、等可写事件、先 take_error（读一次就清零）再 peer_addr。被拒时可写事件照样来，带着错误位；被拒后的 peer_addr 在 Linux 上是 NotConnected、macOS 上是 EINVAL，所以 take_error 必须先做。只注册 READABLE 等不到连接成功。
  • from_std 不设非阻塞。忘了设时事件照常来，读干净后的那次 recvfrom 阻塞住整个事件循环。
  • 读写、peek、try_io 都经过 do_io；绕过它（经 Deref 直接调 std）只在需要重新武装的平台上出错，peek 在 1.1.0 之前就是这样（#1895）。

💬 关联讨论

发起讨论
加载中...