通过迭代WebServer逐步深入Rust异步编程

原文链接

我发现，理解一个新概念的最好方法之一是从头开始：从零起步，一点点地创建它，不仅要学习它是如何工作的，还要学习它为什么会这样设计。

这不是一篇实用的异步指南，希望它涵盖的背景知识能帮助您思考异步问题，或至少满足您的好奇心，而且不会因为太多细节让您觉得厌烦。

但它依然很长……

目录

• 最简单的 Web 服务
• 多线程的 Web 服务
• 非阻塞的 Web 服务
• 多路复用的 Web 服务
• Futures
  • 执行器
  • 任务的调度
  • 异步的 Web 服务
• 全功能的 Web 服务
• 优雅的 Web 服务
• 回到现实
  • Pinning
  • async/await
• [Tokio 实现的 Web 服务](#tokio 实现的-Web-服务)
• 后记

最简单的 Web 服务

我们从最简单的Web服务开始，进入异步编程之旅。我们的Web服务只使用 std::net crate 中的标准网络类型，只接受 HTTP 请求并用基本响应进行回复。接下来的内容中，我们将尽量忽略 HTTP 规范相关要求、忽略代码实用性，将焦点集中在 Web 服务的基本流程上。

HTTP 是构建于 TCP 之上，基于文本的传输协议。我们的 Web 服务通过建立一个TcpListener 来接受客户端的 TCP 连接请求：

use std::net::TcpListener;

fn main() {
    let listener = TcpListener::bind("localhost:3000").unwrap();
}

Web 服务相应来自客户端传入的连接请求，按顺序挨个执行它们。

use std::net::{TcpListener, TcpStream};
use std::io;

fn main() {
    // ...

    loop {
        let (connection, _) = listener.accept().unwrap();

        if let Err(e) = handle_connection(connection) {
            println!("failed to handle connection: {e}")
        }
    }
}

fn handle_connection(connection: TcpStream) -> io::Result<()> {
    // ...
}

TCP 连接是 Web 服务与客户端之间的双向数据通道，用 TcpStream 类型表示。它实现了 Read 和 Write 特性( trait )，抽象了 Tcp 内部细节，使我们能够读取或写入普通的字节数据。

Web 服务接受 HTTP 请求，为此，需要创建缓存数组来保存传入的请求。

fn handle_connection(connection: TcpStream) -> io::Result<()> {
    let mut request = [0u8; 1024];
    // ...

    Ok(())
}

接下来就是从连接读取数据，写入缓存。读取动作每次读到的字节数不定，所以需要跟踪已读取字节数。读取动作在循环中不断重复，依次将读到的内容写入缓存。

use std::io::Read;

fn handle_connection(mut connection: TcpStream) -> io::Result<()> {
    let mut read = 0;
    let mut request = [0u8; 1024];

    loop {
        // 尝试从流中读取数据
        let num_bytes = connection.read(&mut request[read..])?;

        // 跟踪已读取的字节数
        read += num_bytes;
    }

    Ok(())
}

直到读到一个特殊的字符序列 \r\n\r\n，我们约定用它来作为读取的结束标志。

fn handle_connection(mut connection: TcpStream) -> io::Result<()> {
    let mut read = 0;
    let mut request = [0u8; 1024];

    loop {
        // 尝试从流中读取数据
        let num_bytes = connection.read(&mut request[read..])?;

        // 跟踪已读取的字节数
        read += num_bytes;

        // 判断是否读到结束标记
        if request.get(read - 4..read) == Some(b"\r\n\r\n") {
            break;
        }
    }

    Ok(())
}

还有一种情况：如果连接已断开，读到的字节数将是0，此时如果客户端的请求未发送完毕，直接返回转入处理下一个连接。

再次声明，不用担心 HTTP 规范限制，我们的目的是让 Web 服务能工作就行。

fn handle_connection(mut connection: TcpStream) -> io::Result<()> {
    let mut read = 0;
    let mut request = [0u8; 1024];

    loop {
        // 尝试从流中读取数据
        let num_bytes = connection.read(&mut request[read..])?;

        // 客户端已断开
        if num_bytes == 0 { // 👈
            println!("client disconnected unexpectedly");
            return Ok(());
        }

        // 跟踪已读取的字节数
        read += num_bytes;

        // 判断是否读到结束标记
        if request.get(read - 4..read) == Some(b"\r\n\r\n") {
            break;
        }
    }

    Ok(())
}

一旦读取完成，就可以将读到的结果转换为字符串，并以日志形式输出到控制台。

fn handle_connection(stream: TcpStream) -> io::Result<()> {
    let mut read = 0;
    let mut request = [0u8; 1024];

    loop {
        // ...
    }

    let request = String::from_utf8_lossy(&request[..read]);
    println!("{request}");

    Ok(())
}

接下来实现响应部分。

跟读操作类似，写操作也可能不会一次完成。所以同样需要采用循环的方式，不断写入响应数据到流中，每次写入从上一次结束的位置开始，直到写入全部完成。

use std::io::Write;

fn handle_connection(mut connection: TcpStream) -> io::Result<()> {
    // ...

    // 生成HTTP的响应数据
    let response = concat!(
        "HTTP/1.1 200 OK\r\n",
        "Content-Length: 12\n",
        "Connection: close\r\n\r\n",
        "Hello world!"
    );

    let mut written = 0;

    loop {
        // 持续写入响应到流
        let num_bytes = connection.write(response[written..].as_bytes())?;

        // 客户端已断开
        if num_bytes == 0 {
            println!("client disconnected unexpectedly");
            return Ok(());
        }

        written += num_bytes;

        // 判断响应内容是否已经完整写入流
        if written == response.len() {
            break;
        }
    }
}

最后，我们执行刷新( flush )操作，确保写入操作已执行完毕¹。

fn handle_connection(mut connection: TcpStream) -> io::Result<()> {
    let mut written = 0;
    
    loop {
        // ...
    }

    // 刷新响应
    connection.flush()
}

好了，现在我们有了一个能工作的 Web 服务！

$ curl localhost:3000
# => Hello world!

多线程的 Web 服务

好吧，它虽然可以工作，但是有一点问题。

看看我们接受请求的循环：

let listener = TcpListener::bind("localhost:3000").unwrap();

loop {
    let (connection, _) = listener.accept().unwrap();

    if let Err(e) = handle_connection(connection) {
        // ...
    }
}

看到问题了吗？

我们一次只能响应一个请求。

从一个网络连接读写数据不是即时完成的，中间要经过大量的基础设备，比如网关、路由器等等，如果两个用户同时向我们发起请求会发生什么？十个、十万个呢？随用户规模增长，我们的服务会延迟、卡顿，直至不可用，那么如何改进呢？

可能的选择有好几种，但到目前为止，最简单的是线程的方式。为每个请求创建一个线程，就能我们的服务能响应无限的用户增长，对吧？

fn main() {
    // ...
    loop {
        let (connection, _) = listener.accept().unwrap();

        // 为每个传入请求创建一个线程
        std::thread::spawn(|| {
            if let Err(e) = handle_connection(connection) {
                // ...
            }
        });
    }
}

实际上，这很有可能！就算不是无限的，但随着每个请求都在单独的线程中处理，我们服务的吞吐量会显著增加。

这到底是怎么回事？

跟大多数现代操作系统一样，在 linux 中，程序都是在一个单独的进程中运行的。虽然看起来每个活动程序都是同时运行的，但在物理上，一个 CPU 内核一次只能执行一个任务，或者通过超线程技术同时执行两个。为了让所有的程序都能执行，操作系统内核会不断切换它们，暂停当前正在运行的程序，切换到另一个并运行它，如此往复。这些上下文切换以毫秒为单位发生，形成了感觉上的“并行”。

内核调度器通过在多个内核之间分配工作负载来利用多核。每个核心管理一部分进程²，这意味着某些程序可以真正意义上得到并行运行。

 cpu1 cpu2 cpu3 cpu4
|----|----|----|----|
| p1 | p3 | p5 | p7 |
|    |____|    |____|
|    |    |____|    |
|____| p4 |    | p8 |
|    |    | p6 |____|
| p2 |____|    |    |
|    | p3 |    | p7 |
|    |    |    |    |

这种调度类型被称为抢占式多任务调度：内核决定进程运行多长时间被抢占，切换到其他进程。

该模式下，内核确保各独立进程不会访问到其他进程的内存，从而保证各种类型的程序都能得到良好地运行。但是，这使得上下文切换更加昂贵，因为内核在执行上下文切换之前必须刷新内存的某些部分，以确保内存被正确隔离³。

线程跟进程类似⁴，区别是线程可以与同一父进程下的线程共享内存，从而实现在同一程序的线程之间共享状态。除此之外线程的调度和进程没有任何区别。

对我们的服务而言，1 线程 / 1 请求 模式最关键的问题是我们的服务是 I/O 绑定的。handle_connection 执行过程中中绝大部分时间并不是用于计算，而是用于等待，等待从网络连接中收发数据，等待读、写、刷新等 I/O 阻塞的操作执行完毕。我们希望的是，发送一个 I/O 请求后，让出控制权给内核，等操作完成后内核再将控制权交回。在此期间，内核可以执行其他程序。

通常情况下，处理一个网络请求时绝大部分时间都在等待其他任务完成，比如数据库查询或接收 HTTP 请求。多个工作线程效率高的原因是我们可以利用等待的时间来处理其他请求。

非阻塞的 Web 服务

看起来多线程已经完全满足我们的需求，并且它使用也很简单，那么为什么我们还要继续呢？

您也许听说过线程很“重”、上下文切换非常“昂贵”等说法，但是现在，这并不准确，现代的服务器能毫不费力地处理上万的线程。

问题在于阻塞 I/O 将程序的控制权完全交给了操作系统内核，在程序执行完成之前，我们没有任何的干预手段可用，这让我们实现某些操作变得非常困难，比如取消操作和选择操作。

假设我们要实现优雅的服务关停操作。当我们按下 ctrl+c，程序不会马上退出，而是立刻停止接受新的连接请求，当前已建立连接的任务会继续执行，直到完成，或者是 30 秒后被强行终止，最后服务才退出。

在阻塞 I/O 模式下，这里的问题是：我们的 accept 循环会阻塞，直到下一个连接到来。我们可以在新连接请求被接受之前或之后检查 ctrl+c 信号，如果在处理 accept 时信号进来，我们必须等待下一次连接被接受，这期间只有内核拥有程序完全的控制权。

loop {
    // 调用accept之前检查ctrl+c信号
    if got_ctrl_c() {
        break;
    }

    // **如果ctrl+c在这里发生，会出现什么情况?**
    let (connection, _) = listener.accept().unwrap();

    // 在新的连接被接受之前，这不会被调用
    if got_ctrl_c() {
        break;
    }

    std::thread::spawn(|| /* ... */);
}

我们想要的是像 match 操作一样，针对 I/O，同时侦听连接请求和 ctrl+c 信号：

loop {
    // 类似这样...
    match {
        ctrl_c() => {
            break;
        },
        Ok((connection, _)) = listener.accept() => {
            std::thread::spawn(|| ...);
        }
    }
}

对于运行时间超过 30 秒的任务，又该怎么处理呢？我们可以设置一个标记让线程停止，那么又该多久检测一次标记呢？我们又回到了老问题：因为 I/O 阻塞导致我们丧失了程序的控制权，除了等它执行完毕，没有好的方式来强制取消一个线程的执行。

这正是线程和阻塞 I/O 令人头疼的地方，因为应用程序的控制权完全交给了内核，导致实现基于事件的逻辑变得非常困难。

某些平台下，可以使用平台特定接口来实现这一点，比如Unix信号处理机制。虽然信号处理机制简单，并且在某些场景下工作得很好，但在场景变得复杂的时候，信号处理机制会变得非常繁琐。在本文末尾，我们描述了另一种表达复杂控制流的方法。您可以根据实际情况来挑选合适的方式。

那么，有没有既能执行 I/O，又不用出让控制权给内核的实现方法呢？

实际上，还有另一种实现 I/O 操作的方法，称为非阻塞 I/O( non-bloking I/O )。顾名思义，非阻塞操作永远不会阻塞调用线程，它会立即返回，如果给定的资源不可用，则返回一个错误。

通过将 TCP 侦听器和连接置于非阻塞模式，我们可以切换到非阻塞 I/O 的实现方式。

let listener = TcpListener::bind("localhost:3000").unwrap();
listener.set_nonblocking(true).unwrap();

loop {
    let (connection, _) = listener.accept().unwrap();
    connection.set_nonblocking(true).unwrap();

    // ...
}

非阻塞 I/O 的工作模式有一些不同：如果 I/O 请求不能立即完成，内核将返回一个 WouldBlock 错误代码。尽管被表示为错误代码，但 WouldBlock 并不是真正的错误，它只是意味着当前操作无法立即执行完毕，让我们可以自行决定接下来要做什么。

use std::io;

// ...
listener.set_nonblocking(true).unwrap();

loop {
    let connection = match listener.accept() {
        Ok((connection, _)) => connection,
        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
            // 操作还不能执行
            // ...
        }
        Err(e) => panic!("{e}"),
    };

    connection.set_nonblocking(true).unwrap();
    // ...
}

假设在调用 accept() 之后没有连接请求进来，在阻塞 I/O 模式下，我们只能一直等待新的连接，但现在，WouldBlock 不是将控制权交给内核，而是交回我们手里。

我们的 I/O 终于不阻塞了！但此时我们能做点什么呢？

WouldBlock 是一个临时的状态，意味着在未来某个时刻，当前套接字会准备好用于读或写。所以从技术上讲，我们应该一直等到(作者用了自旋这个单词-spin until)套接字状态变成可用( ready )。

loop {
    let connection = match listener.accept() {
        Ok((connection, _)) => connection,
        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
            continue; // 👈
        }
        Err(e) => panic!("{e}"),
    };
}

但是自旋还不如阻塞，至少阻塞 I/O 模式下，操作系统还可以给其他线程执行的机会。所以我们真正需要的，是为全部任务创建一个有序的调度器，来完成曾经由操作系统来为我们做的事情。

让我们从头回顾一遍：

首先我们创建了一个 TCP 侦听器：

let listener = TcpListener::bind("localhost:3000").unwrap();

然后设置它为非阻塞模式：

listener.set_nonblocking(true).unwrap();

接下来进入主循环，循环中第一件事情是接受一个新的 TCP 连接请求。

// ...

loop {
    match listener.accept() {
        Ok((connection, _)) => {
            connection.set_nonblocking(true).unwrap();

            // ...
        },
        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
        Err(e) => panic!("{e}"),
    }
}

现在，我们不能继续直接为已建立的连接服务，导致其他请求被忽略。我们必须能跟踪所有的活动连接。

// ...

let mut connections = Vec::new(); // 👈

loop {
    match listener.accept() {
        Ok((connection, _)) => {
            connection.set_nonblocking(true).unwrap();
            connections.push(connection); // 👈
        },
        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
        Err(e) => panic!("{e}"),
    }
}

但是我们不能无休止地接受连接请求。当没有操作系统调度的便利时，我们需要在主循环的每一次迭代中，将所有的事情都推进一点点。一旦新的连接请求被接受，我们需要处理所有的活跃连接。

对于每一个连接，我们必须执行任何需要的操作来推进请求的处理，无论是读取请求还是写入响应。

// ...
loop {
    // 尝试接受新的连接请求
    match listener.accept() {
        // ...
    }

    // 针对活跃的连接进行处理
    for connection in connections.iter_mut() {
        // 🤔
    }
}

还记得之前的 handle_connection 功能吗？

fn handle_connection(mut connection: TcpStream) -> io::Result<()> {
    let mut request = [0u8; 1024];
    let mut read = 0;

    loop {
        let num_bytes = connection.read(&mut request[read..])?;  // 👈
        // ...
    }

    let request = String::from_utf8_lossy(&request[..read]);
    println!("{request}");

    let response = /* ... */;
    let mut written = 0;

    loop {
        let num_bytes = connection.write(&response[written..])?; // 👈

        // ...
    }

    connection.flush().unwrap(); // 👈
}

我们需要执行不同的 I/O 操作，比如读、写和刷新。阻塞模式下，代码会按我们写的顺序执行。但现在我们必须面对这样一个事实，在执行 I/O 的任何时候都可能面临WouldBlock，导致当前执行无法取得进展。

同时，我们不能简单地丢掉这个结果去处理下一个活动连接，我们需要跟踪当前连接的状态，方便在下次回来时能从正确的地方继续。

我们设计了一个枚举来保存 handle_connetion 的状态，它有三种可能的状态：

enum ConnectionState {
    Read,
    Write,
    Flush
}

请记住，我们需要的不是记录事务单独的状态，例如将请求转换为字符串，我们需要的是在遇到 WouldBlock 时，能记住当时的状态。

读、写操作的状态还包含当前已读写的字节数和本地缓存。之前我们在函数中定义它，现在我们需要它在整个主循环的生命周期中存在。

enum ConnectionState {
    Read {
        request: [u8; 1024],
        read: usize
    },
    Write {
        response: &'static [u8],
        written: usize,
    },
    Flush,
}

我们在每一次 handle_connection 开始执行时初始化连接状态为 Read，request 为 0 值，read 为 0 字节。

// ...

let mut connections = Vec::new();

loop {
    match listener.accept() {
        Ok((connection, _)) => {
            connection.set_nonblocking(true).unwrap();


            let state = ConnectionState::Read { // 👈
                request: [0u8; 1024],
                read: 0,
            };

            connections.push((connection, state));
        },
        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
        Err(e) => panic!("{e}"),
    }
}

现在，我们可以尝试根据其当前状态，将每个连接向前推进了。

// ...
loop {
    match listener.accept() {
        // ...
    }

    for (connection, state) in connections.iter_mut() {
        if let ConnectionState::Read { request, read } = state {
            // ...
        }

        if let ConnectionState::Write { response, written } = state {
            // ...
        }

        if let ConnectionState::Flush = state {
            // ...
        }
    }
}

如果当前连接仍然处于 Read 状态，继续做读取操作，唯一不同的是，如果收到WouldBlock, 则继续处理下一个活动连接。

// ...

'next: for (connection, state) in connections.iter_mut() {
    if let ConnectionState::Read { request, read } = state {
        loop {
            // 尝试从流中读取数据
            match connection.read(&mut request[*read..]) {
                Ok(n) => {
                    // 跟踪已读取的字节数
                    *read += n
                }
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    // 当前连接的操作还未就绪，继续处理下一个连接
                    continue 'next; // 👈
                }
                Err(e) => panic!("{e}"),
            }

            // 判断是否读到结束标记
            if request.get(*read - 4..*read) == Some(b"\r\n\r\n") {
                break;
            }
        }

        // 操作完成，打印收到的请求数据
        let request = String::from_utf8_lossy(&request[..*read]);
        println!("{request}");
    }

    // ...
}

还有读到 0 字节的问题需要处理，之前我们只是从 handle_connection 中退出，state 变量会自动被清空。但是现在，我们必须自己处理当前连接。当前我们正在遍历connections 列表，所以需要一个单独的列表来收集已完成的活动连接，后续再来处理。

let mut completed = Vec::new(); // 👈

'next: for (i, (connection, state)) in connections.iter_mut().enumerate() {
    if let ConnectionState::Read { request, read } = state {
        loop {
            // 尝试从流中读取数据
            match connection.read(&mut request[*read..]) {
                Ok(0) => {
                    println!("client disconnected unexpectedly");
                    completed.push(i); // 👈
                    continue 'next;
                }
                Ok(n) => *read += n,
                // 当前连接未准备好，先处理下一个活动连接
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => continue 'next,
                Err(e) => panic!("{e}"),
            }

            // ...
        }

        // ...
    }
}

// 按相反顺序迭代以保留索引
for i in completed.into_iter().rev() {
    connections.remove(i); // 👈
}

读操作完成后，我们必须切换到 Write 状态并尝试写入回应。写操作的逻辑跟读操作非常相似，写操作完成后，需要切换到 Flush 状态。

if let ConnectionState::Read { request, read } = state {
    // ...

    // 切换到写状态
    let response = concat!(
        "HTTP/1.1 200 OK\r\n",
        "Content-Length: 12\n",
        "Connection: close\r\n\r\n",
        "Hello world!"
    );

    *state = ConnectionState::Write { // 👈
        response: response.as_bytes(),
        written: 0,
    };
}

if let ConnectionState::Write { response, written } = state {
    loop {
        match connection.write(&response[*written..]) {
            Ok(0) => {
                println!("client disconnected unexpectedly");
                completed.push(i);
                continue 'next;
            }
            Ok(n) => {
                *written += n;
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                // 当前连接的操作还未就绪，继续处理下一个连接
                continue 'next;
            }
            Err(e) => panic!("{e}"),
        }

        // 判断响应数据是否已全部写入完毕
        if *written == response.len() {
            break;
        }
    }

    // 写操作完成，进入 Flush 状态
    *state = ConnectionState::Flush;
}

成功完成刷新操作后，我们标记当前连接为完成，并从 completed 列表中移除。

if let ConnectionState::Flush = state {
    match connection.flush() {
        Ok(_) => {
            completed.push(i); // 👈
        },
        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
            // 当前连接的操作还未就绪，继续处理下一个连接
            continue 'next;
        }
        Err(e) => panic!("{e}"),
    }
}

就是这样！以下是新的更高水平的 web 服务流程：

fn main() {
    // 绑定侦听器
    let listener = TcpListener::bind("localhost:3000").unwrap();
    listener.set_nonblocking(true).unwrap();

    let mut connections = Vec::new();

    loop {
        // 尝试接受一个连接请求
        match listener.accept() {
            Ok((connection, _)) => {
                connection.set_nonblocking(true).unwrap();

                // 跟踪连接状态
                let state = ConnectionState::Read {
                    request: Vec::new(),
                    read: 0,
                };

                connections.push((connection, state));
            },
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
            Err(e) => panic!("{e}"),
        }

        let mut completed = Vec::new();

        // 尝试驱动活动连接向前进展
        'next: for (i, (connection, state)) in connections.iter_mut().enumerate() {
            if let ConnectionState::Read { request, read } = state {
                // ...
                *state = ConnectionState::Write { response, written };
            }

            if let ConnectionState::Write { response, written } = state {
                // ...
                *state = ConnectionState::Flush;
            }

            if let ConnectionState::Flush = state {
                // ...
            }
        }

        // 保持索引不变，反序遍历 completed 列表，删除已完成操作的连接
        for i in completed.into_iter().rev() {
            connections.remove(i);
        }
    }
}

现在，我们必须自己管理调度，事情变得越来越复杂了……

关键的时候来了……

$ curl localhost:3000
# => Hello world!

工作正常！

多路复用的 Web 服务

现在我们的 Web 服务能在单线程中执行多个请求，没有任何阻塞。如果某个操作被阻塞，它将记住自己的状态并切换，让其他操作执行，这跟内核调度器的行为一致。但是，新设计带来了两个问题.

首先是所有的工作都在主线程中运行，只利用了一个 CPU 核心。我们尽最大努力高效地利用这一核心，但一次仍然只执行一个任务。如果线程能分布在多个核心上，我们同一时间就可以做更多的工作。

不过有一个更大的问题。

我们的主循环效率并不高。

我们对每一个活跃的连接，每一次循环的迭代，都要向内核发出一个 I/O 请求，来检查它是否准备好了。即使调用 read 或 write 返回了 WouldBlock，实际没有执行任何 I/O，它仍然是一个系统调用。系统调用并不便宜。我们可能有 10k 个活跃的连接，但只有 500 个是准备好的。当只有 500 个连接会真正做些事情的时候，调用 read 或 write 10k 次是对 CPU 周期的巨大浪费。

随着连接数的增加，我们的循环变得越来越低效，浪费了更多的时间做无用的工作。

怎么解决这个问题呢？使用阻塞 I/O 时，内核能够有效地调度任务，因为它知道资源什么时候准备好了。使用非阻塞 I/O 时，我们不检查就不知道，但是检查是很昂贵的。

我们需要的是一种高效的方式来跟踪所有的活跃连接，并且在它们准备好的时候得到通知。

事实证明，我们并不是第一个遇到这个问题的人。每个操作系统都提供了针对这个问题的解决方案。在 Linux 上，它叫做 epoll。

epoll(7) - I/O 事件通知机制

epoll API 执行的任务与 poll(2) 类似：监视多个文件描述符，看看是否有任何一个可以进行 I/O 操作。epoll API 可以作为边缘触发（edge-triggered）或水平触发（level-triggered）的接口使用，并且能够很好地扩展到监视大量的文件描述符。

听上去很完美！我们试试看。

epoll 是一组 Linux 系统调用，让我们可以处理一组非阻塞的套接字。直接使用 epoll 并不是很方便，所以我们将使用 epoll crate，它是一个对 C 接口的轻度封装。

首先，我们使用 create 函数来初始化一个 epoll 实例。

// ```toml
// [dependencies]
// epoll = "4.3"
// ```

fn main() {
    let epoll = epoll::create(false).unwrap(); // 👈
}

epoll::create 返回一个文件描述符，它代表了新创建的 epoll 实例。你可以把它看作是一个文件描述符的集合，我们可以从中添加或删除文件描述符。

在 Linux/Unix 中，一切都被视为文件。文件系统上的实际文件、TCP 套接字、以及外部设备都是可以读写的文件。文件描述符是一个整数，它表示系统中打开的“文件”。本文接下来的部分，我们将频繁使用它。

我们要添加的第一个文件描述符是 TCP 监听器。可以用 epoll::ctl 命令来修改 epoll 集合，添加文件描述符使用EPOLL_CTL_ADD标志。

use epoll::{Event, Events, ControlOptions::*};
use std::os::fd::AsRawFd;

fn main() {
    let listener = TcpListener::bind("localhost:3000").unwrap();
    listener.set_nonblocking(true).unwrap();

    // 添加 listener 到 epoll
    let event = Event::new(Events::EPOLLIN, listener.as_raw_fd() as _);
    epoll::ctl(epoll, EPOLL_CTL_ADD, listener.as_raw_fd(), event).unwrap(); // 👈
}

我们传入要注册的资源的文件描述符，也就是 TCP 监听器，以及一个事件。一个事件有两个部分，interest flag 和 data field。interest flag 让我们可以告诉 epoll 我们感兴趣的 I/O 事件。在 TCP 监听器中，我们想要在有新连接进来时得到通知，所以传入 EPOLLIN 标志。

data field 让我们可以存储一个能够唯一标识每个资源的 ID。记住，文件描述符是一个给定文件的唯一整数，所以直接使用它。你会在下一步看到为什么这很重要。

现在轮到主循环。这次不用自旋，用 epoll::wait。

epoll_wait(2) - 等待 epoll 文件描述符上的 I/O 事件

对文件描述符 epfd 指向的 epoll(7) 实例而言，epoll_wait() 系统调用会等待其上的事件。interest list 中的文件描述符指向的 ready list 中，如果有一些可用事件的信息，那么这些信息通过 events 指向的缓冲区返回。

调用 epoll_wait() 将阻塞，直到以下任一情况发生：

• 文件描述符提交了一个事件；
• 调用被信号处理器中断；
• 超时；

epoll::wait 是 epoll 的神奇之处：它阻塞直到我们注册的任何事件变得就绪，并告诉我们哪些事件就绪了。此时这仅用于有新连接进来时，但是很快我们将使用它来阻塞读、写和刷新事件，这些事件我们之前是用自旋的方式处理的。

您可能不喜欢 epoll::wait 是“阻塞”的这一事实，但是，它只在没有任何事情可做的时候才阻塞，而之前我们是在自旋并且做无用的系统调用。这种同时阻塞多个操作的方法被称为 I/O 多路复用。

epoll::wait 接受一个事件的列表，当所关注的文件描述符就绪，它会将文件描述符的信息填充到列表，然后返回被添加的事件的数量。

// ...
loop {
    let mut events = [Event::new(Events::empty(), 0); 1024];
    let timeout = -1; // 阻塞，直到某些事情发生
    let num_events = epoll::wait(epoll, timeout, &mut events).unwrap(); // 👈

    for event in &events[..num_events] {
        // ...
    }
}

每个事件都包含数据字段，该字段与就绪的资源相关联。

for event in &events[..num_events] {
    let fd = event.data as i32;
    // ...
}

还记得我们用文件描述符来标记数据字段吗？我们可以用它来检查事件是否是针对TCP监听器的，如果是，那就意味着有一个传入连接已经准备好被接受了。

for event in &events[..num_events] {
    let fd = event.data as i32;

    // listener准备就绪了吗?
    if fd == listener.as_raw_fd() {
        // 尝试建立一个连接
        match listener.accept() {
            Ok((connection, _)) => {
                connection.set_nonblocking(true).unwrap();

                // ...
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
            Err(e) => panic!("{e}"),
        }
    }
}

如果返回 WouldBlock， 则移动到下一个连接，等待下一次事件发生。

现在需要在 epoll 中注册新的连接，跟注册侦听器一样。

for event in &events[..num_events] {
    let fd = event.data as i32;

    // listener准备就绪了吗?
    if fd == listener.as_raw_fd() {
        // 尝试建立一个连接
        match listener.accept() {
            Ok((connection, _)) => {
                connection.set_nonblocking(true).unwrap();
                 let fd = connection.as_raw_fd();

                 // 注册新连接到epoll
                 let event = Event::new(Events::EPOLLIN | Events::EPOLLOUT, fd as _);
                 epoll::ctl(epoll, EPOLL_CTL_ADD, fd, event).unwrap(); // 👈
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
            Err(e) => panic!("{e}"),
        }
    }
}

这次我们注册了EPOLLIN 和 EPOLLOUT 事件，因为根据连接状态，我们要关注读或写事件。

注册了连接之后，我们将得到 TCP 侦听器和某个连接的事件。我们需要用某种方式存储连接和它们的状态，并能通过查找文件描述符的方式来访问它们。

这次不用 List，用 HashMap。

let mut connections = HashMap::new();

loop {
    // ...
    'next: for event in &events[..num_events] {
        let fd = event.data as i32;

        // listener准备就绪了吗?
        if fd == listener.as_raw_fd() {
            match listener.accept() {
                Ok((connection, _)) => {
                    // ...

                    let state = ConnectionState::Read {
                        request: [0u8; 1024],
                        read: 0,
                    };

                    connections.insert(fd, (connection, state)); // 👈
                }
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
                Err(e) => panic!("{e}"),
            }

            continue 'next;
        }

        // listener未就绪的话，必须有连接是就绪的
        let (connection, state) = connections.get_mut(&fd).unwrap(); // 👈
    }
}

一旦连接和它的状态就绪，我们可以用和之前一样的方法来推进它。从流中读写数据的操作没有任何变化，区别是现在我们仅在接到 epoll 通知时才进行操作。

以前我们必须检查每一个连接，看看是否有什么变得就绪，但现在由 epoll 来处理，避免了任何无用的系统调用。

// ...

// epoll告诉我们连接是否就绪
let (connection, state) = connections.get_mut(&fd).unwrap();

if let ConnectionState::Read { request, read } = state {
    // connection.read...
    *state = ConnectionState::Write { response, written };
}

if let ConnectionState::Write { response, written } = state {
    // connection.write...
    *state = ConnectionState::Flush;
}

if let ConnectionState::Flush = state {
    // connection.flush...
}

所有操作都完成后，我们从 connections 中移除当前连接，它会自动从 epoll 中注销。

for fd in completed {
    let (connection, _state) = connections.remove(&fd).unwrap();
    // 会自动从epoll注销
    drop(connection);
}

就是现在，更高水平的 Web 服务完成了！

fn main() {
    // 创建 epoll
    let epoll = epoll::create(false).unwrap();

    // 绑定到listener
    let listener = /* ... */.

    // 添加 listener 到 epoll
    let event = Event::new(Events::EPOLLIN, listener.as_raw_fd() as _);
    epoll::ctl(epoll, EPOLL_CTL_ADD, listener.as_raw_fd(), event).unwrap();

    let mut connections = HashMap::new();

    loop {
        let mut events = [Event::new(Events::empty(), 0); 1024];

        // 阻塞，直到被 epoll 唤醒
        let num_events = epoll::wait(epoll, 0, &mut events).unwrap();
        let mut completed = Vec::new();

        'next: for event in &events[..num_events] {
            let fd = event.data as i32;

            // listener 是否就绪?
            if fd == listener.as_raw_fd() {
                match listener.accept() {
                    Ok((connection, _)) => {
                        // ...

                        // 添加 connection 到 epoll
                        let event = Event::new(Events::EPOLLIN | Events::EPOLLOUT, fd as _);
                        epoll::ctl(epoll, EPOLL_CTL_ADD, fd, event).unwrap();

                        // 跟踪 connection 的状态
                        let state = ConnectionState::Read {
                            request: [0u8; 1024],
                            read: 0,
                        };

                        connections.insert(fd, (connection, state));
                }

                continue 'next;
            }

            // connection 变得就绪了
            let (connection, state) = connections.get_mut(&fd).unwrap();

            // 尝试基于状态推动当前 connection 的执行
            if let ConnectionState::Read { request, read } = state {
                // connection.read...
                *state = ConnectionState::Write {
                    response: response.as_bytes(),
                    written: 0,
                };
            }

            if let ConnectionState::Write { response, written } = state {
                // connection.write...
                *state = ConnectionState::Flush;
            }

            if let ConnectionState::Flush = state {
                // connection.flush...
            }
        }

        for fd in completed {
            let (connection, _state) = connections.remove(&fd).unwrap();
            // connection 会自动从 epoll 注销
            drop(connection);
        }
    }
}

现在……

$ curl localhost:3000
# => Hello world!

工作正常！

Futures

好吧，我们的服务器现在能在单一线程中同时处理多个请求了。感谢 epoll，它在我们的工作场景中非常高效。但是，仍然还存在问题。

我们需要自己规划任务的执行，需要自己考虑如何高效地调度任务，这使得我们的代码复杂度急剧增加。

任务的执行，也从一个简单的顺序执行循环变成了庞大的事件循环，需要管理多个状态机。

总感觉差点意思。

使我们的原始服务器成多线程非常简单，只需在 thread::spawn 中添加一行代码即可。仔细想想，我们的服务器仍然是一组并发任务，只是我们在一个巨大的循环中混乱地管理它们。

这让扩展功能变得非常困难，在程序中添加的功能越多，循环就变得越复杂，因为所有东西都紧密地耦合在一起。如果可以编写一个类似 thread::spawn 的抽象，能让我们将任务写成独立的单元，能集中在一个地方处理所有任务的调度和事件处理，从而重新获得流程控制权，会怎么样呢？

这种思想被称为异步编程。

我们来看看 thread::spawn 的函数签名：

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
    F: FnOnce() -> T + Send + 'static,
    T: Send + 'static;

thread::spawn 接受一个闭包，但我们的版本其实并不能，因为我们不是操作系统，不能随意抢占代码。我们需要以某种方式来表达一项不受阻碍、可恢复的任务。

// fn spawn<T: Task>(task: T);

trait Task {}

处理一个请求是一个任务,从连接读取或写入数据也是。一个任务实质上是一段待执行的代码，代表着它将在未来某个时候需要得到执行。

Future (译注：本意是未来，在异步中我们保留原文代表这种 trait)，确实是个好名字，不是吗？

trait Future {
    type Output;

    fn run(self) -> Self::Output;
}

这个签名并不能工作。run 函数直接返回 Self::Output 意味着它会阻塞直到返回，而这正是我们极力在避免的。我们要寻找其他方法，实现不阻塞的同时，能推动我们的 future 前进，就像我们之前在事件循环的状态机中实现的那样。

实际上，在执行一个 future 的时候，我们要做的就是询问它是否已就绪，轮询( polling ) 它，然后给它执行的机会。

trait Future {
    type Output;

    fn poll(self) -> Option<Self::Output>;
}

看起来差不多了。

但是，如果我们多次调用 poll，除了等着，我们并不能获取 self， 所以它应该是一个引用，一个可变的引用，通过它，我们可以改变任务内部的状态，比如 ConnectState。

trait Future {
    type Output;

    fn poll(&mut self) -> Option<Self::Output>;
}

现在，来设想一下执行这些 future 的调度器：

impl Scheduler {
    fn run(&self) {
        loop {
            for future in &self.tasks {
                future.poll();
            }
        }
    }
}

这看起来不怎么样。

future 初始化完成后，当 epoll 返回的它的事件时，调度器调用它的 poll 方法来给它一个执行的机会。

如果 future 是 I/O 操作，在 接到 epoll 通知时我们就知道它可以执行了。问题是我们不知道 epoll 事件对应的是哪个 future, 因为 future 的执行过程都在内部的 poll 中。

调度器需要传递一个 ID 给 future，它可以用这个 ID 而不是文件描述符向 epoll 注册任何 I/O 资源。通过这种方式，调度器就能把 epoll 事件和 future 对应起来了。

impl Scheduler {
    fn spawn<T>(&self, mut future: T) {
        let id = rand();
        // 对 future 调用一次轮询让它运转起来，传入的参数是它的 ID
        future.poll(event.id);
        // 保存 future
        self.tasks.insert(id, future);
    }

    fn run(self) {
        // ...

        for event in epoll_events {
            // 根据事件 ID 轮询相应的 future
            let future = self.tasks.get(&event.id).unwrap();
            future.poll(event.id);
        }
    }
}

您知道的，如果有一种更通用的方式来告诉调度器 future 当前的进度，而不是把每个 future 都绑定到 epoll，那就太好了。future 有不同的类型，每种类型都可能有不同的执行方式，比如在后台线程中执行的定时器、或者是一个通道，它需要在消息已就绪的时候通知相应的任务。

如果我们给 future 更多的控制权呢？ 如果我们不是简单地用一个 ID, 而是给每个 future 一个能唤醒自己的方法，能通知调度器它已经准备好可以执行了呢？

一个简单的回调函数就可以做到。

#[derive(Clone)]
struct Waker(Arc<dyn Fn() + Send + Sync>);

impl Waker {
    fn wake(&self) {
        (self.0)()
    }
}

trait Future {
    type Output;

    fn poll(&mut self, waker: Waker) -> Option<Self::Output>;
}

调度器可以为每个 future 提供一个回调函数，它被调用时更新该 future 在调度器中的状态，标记 future 为就绪。这样调度器就完全和 epoll 或其他任何独立通知系统解耦了。

唤醒器 ( Waker ) 是线程安全的，允许我们使用后台线程唤醒 future。目前所有的任务都已连接到 epoll，这马上就会派上用场了。

执行器

考虑一个 future 从 TCP 连接读取数据的场景： 它收到一个唤醒器, 需要在 epoll 返回给它 EPOLLIN 事件的时候得到执行，但事件发生时，它只会呆在调度器的队列里面，并不会得到执行。显然，future 不能唤醒自己，我们还需要其他帮助。

所有 I/O 类型的 future 都需要将他们的唤醒器传递给 epoll, 实际上，它们需要的不止这些， 他们还需要一种驱动 epoll 的后台服务，以便我们可以在其中注册唤醒器。

这种服务通常被称为反应器( Reactor )。

像之前一样，反应器只是一个简单的对象，它保存了 epoll 描述符和以这个描述符为 键值的任务列表。不同之处是键值对应的值不是已建立的 TCP 连接，而是唤醒器。

thread_local! {
    static REACTOR: Reactor = Reactor::new();
}

struct Reactor {
    epoll: RawFd,
    tasks: RefCell<HashMap<RawFd, Waker>>,
}

impl Reactor {
    pub fn new() -> Reactor {
        Reactor {
            epoll: epoll::create(false).unwrap(),
            tasks: RefCell::new(HashMap::new()),
        }
    }
}