基于 epoll 的协程调度器——零基础深入浅出 C++20 协程

浅思暖意 發表於 2025-8-18 09:57:00

基于 epoll 的协程调度器——零基础深入浅出 C++20 协程

<h1>前言</h1>
上一篇《没有调度器的协程不是好协程》谈到协程如何自动运行，然而那个例子里的调度器还是不太自然，考查一下真实场景，挂起的协程一般是在等待异步事件的完成，如果异步事件没完成就轮到自己执行，它其实还是无法继续，相当于一次无效唤醒。所以这一篇准备引入异步事件，看看在真实的场景下，调度器是如何运作的。
文章仍然遵守之前的创作原则：
* 选取合适的 demo 是头等大事
* 以协程为目标，涉及到的新语法会简单说明，不涉及的不旁征博引
* 若语法的原理非常简单，也会简单展开讲讲，有利于透过现象看本质，用起来更得心应手
上一篇文章里不光引入了初级的调度器，还说明了 final_suspend 与协程自清理的关系、协程句柄通过类型擦除来屏蔽用户定义承诺对象的差异、以及 lambda 表达式的本质是仿函数等，如果没有这些内容铺垫，看本文时会有很多地方难以理解，还没看过的小伙伴，墙裂建议先看那篇。
工具还是之前介绍过的 C++ Insights ，这里不再用到 Compile Explorer，主要是它的运行环境不支持像文件、网络之类的异步 IO，为此需要用户自行搭建开发环境。
<h1>基于 epoll 的 IO 多路复用</h1>
本文演示的异步 IO 以文件操作为主，相比网络操作它具有代码量少、易于测试的优点。为了简化复杂度，这里没有接入任何三方库，而是直接调用操作系统 raw API，阅读本文需要具有 IO 多路复用 (multiplexing) 的知识基础，例如 Linux 的 epoll 或 Windows 的 IOCP。
在单线程时代，想要处理多个 IO 事件也不是不行，只要将异步 IO 句柄交给 select / poll / epoll / kqueue 等待即可，当任一 IO 事件到达时，控制权将从阻塞等待中返回，并告知用户哪个句柄上有何种事件发生，从而方便用户直接处理那个句柄上的 IO 事件，并且预期将不会被阻塞。这种模型因为检测完成后，还需要用户动作一下，也称为 Reactor 模型；相对的，还有 Proactor 模型，主要是基于 Windows IOCP，当事件完成时，相应的读、写动作已由系统完成，不再需要用户动作，故有此区别，关于这一点，后面在介绍基于 IOCP 的调度器时详述。
类 Unix 系统上的 IO 多路分离器比较多，早期的 select 就能监控 IO 句柄的读、写、异常三个事件集，并且带超时能力；后面发展的 poll 消除了 select 对句柄数量的限制；Linux 上诞生的 epoll 解决了 select & poll 在句柄数量增长时效能线性下降的问题，主要优化了句柄集合在用户态与内核态的来回复制、返回时遍历句柄集等性能开销；kqueue 则是 BSD 系统上的 epoll 平替，两者都支持水平触发与边缘触发两种模式。
水平触发意味着只要句柄上有事件，分离器就会一直通知，上述四个默认都是水平触发，适合少量离散数据的场景；边缘触发意味着一次通知中如果不将对应的事件处理完，下次不会再通知，除非有新的事件产生，epoll / kqueue 可选边缘触发，适合大数据量的场景，可以有效缓解高频通知导致的数据传输低效问题。
恶补了 IO 多路复用机制相关的知识后，考虑到我们是在 Linux 上进行测试，这里选取了 epoll 作为分离器。需要注意的是 epoll 不能直接处理普通文件读写，需要借助 fifo 文件，后面我们会看到这一点，话不多说直接上 demo：
<pre class="language-cpp highlighter-hljs"><code>#include <coroutine>
#include <unordered_map>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <vector>
#include <stdexcept>
#include <iostream>
#include <sstream>

#define MAX_EVENTS 10

struct Task {
struct promise_type {
 Task get_return_object() { return {}; }
 std::suspend_never initial_suspend() { return {}; }
 std::suspend_never final_suspend() noexcept { return {}; }
 void return_void() {}
 void unhandled_exception() { std::terminate(); }
};
};

class EpollScheduler {
private:
int epoll_fd;
std::unordered_map<int, std::coroutine_handle<>> io_handles;
public:
EpollScheduler() {
 epoll_fd = epoll_create(MAX_EVENTS);
 if (epoll_fd == -1) {
 std::stringstream ss;
 ss << "epoll_create failed, error " << errno;
 throw std::runtime_error(ss.str());
 }
}

~EpollScheduler() {
 close(epoll_fd);
}

void register_io(int fd, std::coroutine_handle<> handle) {
 if (io_handles.find(fd) == io_handles.end()) {
 io_handles = handle;

 epoll_event event{};
 event.events = EPOLLIN | EPOLLET;
 event.data.fd = fd;
 if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1) {
 std::stringstream ss;
 ss << "epoll_ctl failed, error " << errno;
 throw std::runtime_error(ss.str());
 }
 }
}

void run() {
 while (true) {
 epoll_event events = { 0 };
 int n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
 for (int i = 0; i < n; ++i) {
 int ready_fd = events.data.fd;
 if (auto it = io_handles.find(ready_fd); it != io_handles.end()) {
 it->second.resume();
 }
 }
 }
}
};

struct AsyncReadAwaiter {
EpollScheduler& sched;
int fd;
std::string buffer;

AsyncReadAwaiter(EpollScheduler& s, int file_fd, size_t buf_size)
 : sched(s), fd(file_fd), buffer(buf_size, '\0') {}

bool await_ready() const {
 return false;
}

void await_suspend(std::coroutine_handle<> h) {
 sched.register_io(fd, h);
}

std::string await_resume() {
 ssize_t n = read(fd, buffer.data(), buffer.size());
 if (n == -1) {
 std::stringstream ss;
 ss << "read failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 buffer.resize(n);
 return std::move(buffer);
}
};

Task async_read_file(EpollScheduler& sched, const char* path) {
int fd = open(path, O_RDONLY | O_NONBLOCK);
if (fd == -1) {
 std::stringstream ss;
 ss << "open failed, error " << errno;
 throw std::runtime_error(ss.str());
}

while (true) {
 auto data = co_await AsyncReadAwaiter(sched, fd, 4096);
 std::cout << "Read " << data.size() << " bytes\n";
 // if (data.size() == 0)
 // break;
}
close(fd);
}

int main(int argc, char* argv[]) {
if (argc < 2) {
 std::cout << "Usage: sample pipe" << std::endl;
 return 1;
}

EpollScheduler scheduler;
async_read_file(scheduler, argv);
scheduler.run();
return 0;
}</code></pre>
先来看编译，公司的开发环境中安装的 gcc 最高版本为 12.1：
<pre class="language-cpp highlighter-hljs"><code>$ /opt/compiler/gcc-12/bin/g++ --version
/opt/compiler/gcc-12/bin/g++ (GCC) 12.1.0
Copyright (C) 2022 Free Software Foundation, Inc.
This is free software; see the source for copying conditions.There is NO
warranty; not even for MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.</code></pre>
经 Compile Explorer 验证，可用：
<img src="https://img2024.cnblogs.com/blog/1707550/202507/1707550-20250716155338159-2029192553.png">
一点点降低版本尝试，发现能编译这段代码的最低 gcc 版本是 11.1，如果你需要在本地安装 gcc 的话，大于等于这个版本就行。
包装一个简单的 Makefile：
<pre class="language-cmake highlighter-hljs"><code>all: sample

sample : sample.cpp
/opt/compiler/gcc-12/bin/g++ -std=c++20 -o $@ $^
mkfifo communication.pipe

clean:
rm sample communication.pipe</code></pre>
mkfifo 用于管道文件 (communication.pipe) 的创建。启动 sample 程序后可以在管道另一侧用脚本写一些数据进去：
<pre class="language-bash highlighter-hljs"><code>for ((i=1;i<500;++i)); doecho hello > communication.pipe; done</code></pre>
写入 500 个 hello 字符串，接收端的 sample 输出如下：
<pre class="language-bash highlighter-hljs"><code>$ ./sample communication.pipe
Read 6 bytes
Read 60 bytes
Read 6 bytes
Read 54 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 12 bytes
Read 0 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
Read 6 bytes
...</code></pre>
demo 唯一的参数是 pipe 文件路径。如果使用普通文件做同样的测试：
<pre class="language-bash highlighter-hljs"><code>$ ./sample sample.cpp
terminate called after throwing an instance of 'std::runtime_error'
what():epoll_ctl failed, error 1
Aborted (core dumped)</code></pre>
果然报错了，这就是开头所说 epoll 不支持普通文件的特性：对于普通文件，Linux 认为永远可读可写，没必要通过 epoll 进行等待，所以 epoll_ctl 直接返回 EPERM 了。
这个顺便演示了 C++20 编译器会对协程体代码进行 try...catch 的逻辑，任何未捕获的异常终将调用用户承诺对象的 unhandled_exception 接口，这里调了 terminate 来终止进程，关于这一点，请参考《协程本质是函数加状态机》。
 
代码比较长，下面分段看下：
<pre class="language-cpp highlighter-hljs"><code>#include <coroutine>
#include <unordered_map>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <vector>
#include <stdexcept>
#include <iostream>
#include <sstream>

#define MAX_EVENTS 10</code></pre>
返回对象定义，相比之前经典的定义，承诺对象的 final_suspend 未中断协程、返回对象没有析构时销毁协程句柄的动作，意味着协程是个启动后“不管”的类型
<pre class="language-cpp highlighter-hljs"><code>struct Task {
struct promise_type {
 Task get_return_object() { return {}; }
 std::suspend_never initial_suspend() { return {}; }
 std::suspend_never final_suspend() noexcept { return {}; }
 void return_void() {}
 void unhandled_exception() { std::terminate(); }
};
};</code></pre>
跳到 main，果然没有接收协程体 async_read_file 的返回对象，它返回的临时对象将自动析构，不影响协程体正常运转
<pre class="language-cpp highlighter-hljs"><code>int main(int argc, char* argv[]) {
if (argc < 2) {
 std::cout << "Usage: sample pipe" << std::endl;
 return 1;
}

EpollScheduler scheduler;
async_read_file(scheduler, argv);
scheduler.run();
return 0;
}</code></pre>
回到调度器，构造与析构负责 epoll 句柄的生命周期管理，联系 main 中 scheduler 的定义，它会贯穿整个进程生命期
<pre class="language-cpp highlighter-hljs"><code>class EpollScheduler {
private:
int epoll_fd;
std::unordered_map<int, std::coroutine_handle<>> io_handles;
public:
EpollScheduler() {
 epoll_fd = epoll_create(MAX_EVENTS);
 if (epoll_fd == -1) {
 std::stringstream ss;
 ss << "epoll_create failed, error " << errno;
 throw std::runtime_error(ss.str());
 }
}

~EpollScheduler() {
 close(epoll_fd);
}</code></pre>
调度器提供协程注册接口。与之前相比这里不再使用简单的先进先出队列，而是将文件句柄与协程句柄通过 map 关联起来，方便后面根据事件句柄唤醒协程
<pre class="language-cpp highlighter-hljs"><code> void register_io(int fd, std::coroutine_handle<> handle) {
 if (io_handles.find(fd) == io_handles.end()) {</code></pre>
select 或 poll 需要每次检测前都准备句柄集，epoll 则不同，句柄只需注册一次，后续就能一直监听该句柄上的事件，重复注册还会导致 epoll_ctl 返回失败，因此这里有判重逻辑
<pre class="language-cpp highlighter-hljs"><code> io_handles = handle;</code></pre>
只注册读事件 (EPOLLIN)，并且使用边缘触发模式 (EPOLLET) 
<pre class="language-bash highlighter-hljs"><code> epoll_event event{};
 event.events = EPOLLIN | EPOLLET;
 event.data.fd = fd;
 if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1) {
 std::stringstream ss;
 ss << "epoll_ctl failed, error " << errno;
 throw std::runtime_error(ss.str());
 }
 }
}</code></pre>
调度器提供的运行接口，循环 wait IO 事件，有读事件才唤醒对应的协程
<pre class="language-cpp highlighter-hljs"><code> void run() {
 while (true) {
 epoll_event events = { 0 };
 int n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
 for (int i = 0; i < n; ++i) {
 int ready_fd = events.data.fd;
 if (auto it = io_handles.find(ready_fd); it != io_handles.end()) {
 it->second.resume();
 }
 }
 }
}
};</code></pre>
协程体内部打开文件句柄准备进行异步读取 (O_NONBLOCK)，每次通过等待对象读取数据并展示在控制台，与之前 co_await 纯粹用于挂起协程等待相比，这里可以通过它返回数据
<pre class="language-cpp highlighter-hljs"><code>Task async_read_file(EpollScheduler& sched, const char* path) {
int fd = open(path, O_RDONLY | O_NONBLOCK);
if (fd == -1) {
 std::stringstream ss;
 ss << "open failed, error " << errno;
 throw std::runtime_error(ss.str());
}

while (true) {
 auto data = co_await AsyncReadAwaiter(sched, fd, 4096);
 std::cout << "Read " << data.size() << " bytes\n";
 // if (data.size() == 0)
 // break;
}
close(fd);
}</code></pre>
等待对象是本次的核心：await_ready 返回 false 挂起协程；await_suspend 在协程挂起前注册协程句柄到调度器；await_resume 在协程恢复后读取数据，并返回给 co_await 调用者
<pre class="language-cpp highlighter-hljs"><code>struct AsyncReadAwaiter {
EpollScheduler& sched;
int fd;
std::string buffer;

AsyncReadAwaiter(EpollScheduler& s, int file_fd, size_t buf_size)
 : sched(s), fd(file_fd), buffer(buf_size, '\0') {}

bool await_ready() const {
 return false;
}

void await_suspend(std::coroutine_handle<> h) {
 sched.register_io(fd, h);
}

std::string await_resume() {
 ssize_t n = read(fd, buffer.data(), buffer.size());
 if (n == -1) {
 std::stringstream ss;
 ss << "read failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 buffer.resize(n);
 return std::move(buffer);
}
};</code></pre>
老规矩，下面有请 C++ Insights 上场，看看编译器底层做的工作与之前相比有何差异，内容比较多，只捡关键的看下：
<pre class="language-cpp highlighter-hljs"><code>struct __async_read_fileFrame
{
void (*resume_fn)(__async_read_fileFrame *);
void (*destroy_fn)(__async_read_fileFrame *);
std::__coroutine_traits_impl<Task>::promise_type __promise;
int __suspend_index;
bool __initial_await_suspend_called;</code></pre>
协程状态与之前别无二致，注意除了参数外，局部变量如 ss、data 也都放进来了，因此在编写协程体时需要格外注意，能放在内部调用的变量，不要直接放在协程体
<pre class="language-bash highlighter-hljs"><code>EpollScheduler & sched;
const char * path;
int fd;
std::basic_stringstream<char> ss;
std::basic_string<char, std::char_traits<char>, std::allocator<char> > data;
std::suspend_never __suspend_100_6;
AsyncReadAwaiter __suspend_109_30;
std::basic_string<char, std::char_traits<char>, std::allocator<char> > __suspend_109_30_res;
std::suspend_never __suspend_100_6_1;
};</code></pre>
真正的协程体逻辑被挪到协程 resume 中了
<pre class="language-cpp highlighter-hljs"><code>/* This function invoked by coroutine_handle<>::resume() */
void __async_read_fileResume(__async_read_fileFrame * __f)
{
try
{</code></pre>
开头就是 duff device，这个之前已经见识过了
<pre class="language-cpp highlighter-hljs"><code> /* Create a switch to get to the correct resume point */
switch(__f->__suspend_index) {
 case 0: break;
 case 1: goto __resume_async_read_file_1;
 case 2: goto __resume_async_read_file_2;
 case 3: goto __resume_async_read_file_3;
}</code></pre>
initial_suspend 返回 suspend_never 直接跳过不挂起
<pre class="language-cpp highlighter-hljs"><code> /* co_await insights.cpp:100 */
__f->__suspend_100_6 = __f->__promise.initial_suspend();
if(!__f->__suspend_100_6.await_ready()) {
 __f->__suspend_100_6.await_suspend(std::coroutine_handle<Task::promise_type>::from_address(static_cast<void *>(__f)).operator std::coroutine_handle<void>());
 __f->__suspend_index = 1;
 __f->__initial_await_suspend_called = true;
 return;
}

__resume_async_read_file_1:
__f->__suspend_100_6.await_resume();</code></pre>
打开文件，失败直接抛异常
<pre class="language-cpp highlighter-hljs"><code> __f->fd = open(__f->path, 0 | 2048);
if(__f->fd == -1) {
 __f->ss = std::basic_stringstream<char>();
 std::operator<<(__f->ss, "open failed, error ").operator<<((*__errno_location()));
 throw std::runtime_error(std::runtime_error(__f->ss.str()));
} </code></pre>
循环读文件，AsyncWaitReader 的 await_ready 返回 false 挂起协程，挂起前调用 await_suspend 注册协程到调度器
<pre class="language-cpp highlighter-hljs"><code> while(true) {

 /* co_await insights.cpp:109 */
 __f->__suspend_109_30 = AsyncReadAwaiter(__f->sched, __f->fd, 4096);
 if(!__f->__suspend_109_30.await_ready()) {
 __f->__suspend_109_30.await_suspend(std::coroutine_handle<Task::promise_type>::from_address(static_cast<void *>(__f)).operator std::coroutine_handle<void>());
 __f->__suspend_index = 2;
 return;
 } </code></pre>
文件句柄上有可读数据时，调度器恢复协程运行，AsyncWaitReader 的 await_resume 读取数据并记录在 data 中
<pre class="language-cpp highlighter-hljs"><code> __resume_async_read_file_2:
 __f->__suspend_109_30_res = __f->__suspend_109_30.await_resume();
 __f->data = __f->__suspend_109_30_res;
 std::operator<<(std::operator<<(std::cout, "Read ").operator<<(__f->data.size()), " bytes\n");
}</code></pre>
结束循环前关闭句柄 (目前是个死循环走不到这里)，协程终止前调用承诺对象的 return_void，有未捕获异常时调用承诺对象的 unhandled_exception
<pre class="language-cpp highlighter-hljs"><code> close(__f->fd);
/* co_return insights.cpp:100 */
__f->__promise.return_void()/* implicit */;
goto __final_suspend;
} catch(...) {
if(!__f->__initial_await_suspend_called) {
 throw ;
}

__f->__promise.unhandled_exception();
}</code></pre>
final_suspend 返回 suspend_never 直接跳过不挂起，调用 destroy 自动销毁协程状态释放内存
<pre class="language-cpp highlighter-hljs"><code>__final_suspend:

/* co_await insights.cpp:100 */
__f->__suspend_100_6_1 = __f->__promise.final_suspend();
if(!__f->__suspend_100_6_1.await_ready()) {
__f->__suspend_100_6_1.await_suspend(std::coroutine_handle<Task::promise_type>::from_address(static_cast<void *>(__f)).operator std::coroutine_handle<void>());
__f->__suspend_index = 3;
return;
}

__resume_async_read_file_3:
__f->destroy_fn(__f);
}</code></pre>
有上一篇的铺垫，看起来没什么尿点，甚至有点老三样。唯一有新意的地方是 co_await 也能通过 await_resume 获取返回数据，这与 co_yield & co_return 有异曲同工之妙，体现出 C++20 协程灵活的一面。
<h1>多文件并行</h1>
上面的例子虽然通过多次读取展示了协程多次唤醒的过程，但没有展示多个 IO 句柄并发的能力，下面稍加改造，同时读取多个 fifo：
<pre class="language-cpp highlighter-hljs"><code>Task async_read_file(EpollScheduler& sched, const char* path) {
...
while (true) {
 auto data = co_await AsyncReadAwaiter(sched, fd, 4096);
 std::cout << "Read [" << data.size() << "] " << data;
 if (data.size() == 0)
 std::cout << std::endl;
}
...
}

int main(int argc, char* argv[]) {
if (argc < 3) {
 std::cout << "Usage: sample pipe1 pipe2" << std::endl;
 return 1;
}

EpollScheduler scheduler;
async_read_file(scheduler, argv);
async_read_file(scheduler, argv);
scheduler.run();
return 0;
}</code></pre>
主要的改动是：
* 协程体展示数据内容，便于区分是从哪个 fifo 读到了数据
* demo 接收两个 pipe 路径，分别调用两个协程进行处理
对应的，修改写数据的脚本：
<pre class="language-bash highlighter-hljs"><code>$ for ((i=1;i<500;++i)); do if [ $((i%2)) -eq 0 ]; then echo hello > communication.pipe; else echo world > communication2.pipe; fi; done</code></pre>
交替在两个 pipe 上写入 hello 与 world，下面是程序输出：
<pre class="language-bash highlighter-hljs"><code>$ ./sample communication.pipe communication2.pipe
Read world
Read world
world
world
Read hello
hello
hello
hello
Read
Read world
Read hello
hello
Read world
world
Read hello
Read world
Read hello
Read world
Read hello
Read world
Read hello
Read world
Read hello
Read world
Read hello
Read world
Read
Read hello
Read world
Read hello
Read
...</code></pre>
读取看起来并不是严格的交替执行，这与 pipe 可读时积累的数据量有关，如果读取前发送端已经累计发送了多次，就会出现上面的情况。不论怎样，这里实现了用协程并行读取文件的能力，并且不需要对跨协程的公共变量做任何并发防护 (如调度器内部 map)，且每个文件的读取逻辑清晰易懂，这可能就是协程的魅力吧。读取 N 个文件的场景 (N>2)，都可以参考上面的进行拓展，此处就不再赘述了。
最后补充一张调用顺序图：
<img src="https://img2024.cnblogs.com/blog/1707550/202507/1707550-20250716173027229-1569251099.png" alt="" height="693" width="2022">
为了便于绘制，调度器的 register_io & run 分开画了。另外非首次读取时，没有 7-8 这条路径，取而代之的是 run 内部的事件循环。

<h1>await_suspend & 试读</h1>
众所周知读写异步 IO 句柄 (<code>O_NONBLOCK</code>) 时不会被阻塞，当系统能满足用户请求时，会读取尽可能多的数据返回；当没有可用数据时，系统立即返回一个错误，一般是 <code>EAGAIN</code> 或 <code>EWOULDBLOCK</code> (Windows)，此时再进入 epoll 等待也不迟，当数据比较频繁时能节约相当可观的 epoll 等待与唤醒，从而提高吞吐性能。
回到 demo，试读的结果决定是否挂起协程，因此最佳的判断位置是在 await_ready，下面是改造后的代码：
<pre class="language-cpp highlighter-hljs"><code>struct AsyncReadAwaiter {
EpollScheduler& sched;
int fd;
int len;
std::string buffer;

AsyncReadAwaiter(EpollScheduler& s, int file_fd, size_t buf_size)
 : sched(s), fd(file_fd), len(0), buffer(buf_size, '\0') { }

bool await_ready(){
 len = 0;
 ssize_t n = read(fd, buffer.data(), buffer.size());
 if (n > 0) {
 len = n;
 return true;
 } else if (n == -1 && errno != EAGAIN) {
 std::stringstream ss;
 ss << "pre read failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 return false;
}

void await_suspend(std::coroutine_handle<> h) {
 sched.register_io(fd, h);
}

std::string await_resume() {
 ssize_t n = read(fd, buffer.data() + len, buffer.size() - len);
 if (n == -1) {
 if (len > 0) {
 buffer.resize(len);
 return std::move(buffer);
 }

 if (errno != EAGAIN) {
 std::stringstream ss;
 ss << "read failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 n = 0;
 }

 buffer.resize(n + len);
 if (len > 0) {
 std::cout << "pre-read " << len << ", read " << n << std::endl;
 }
 return std::move(buffer);
}
};</code></pre>
内容不长，不过也分段解读下：
<pre class="language-cpp highlighter-hljs"><code>struct AsyncReadAwaiter {
EpollScheduler& sched;
int fd;</code></pre>
增加 len 字段记录试读的结果长度
<pre class="language-cpp highlighter-hljs"><code> int len;
std::string buffer;

AsyncReadAwaiter(EpollScheduler& s, int file_fd, size_t buf_size)
 : sched(s), fd(file_fd), len(0), buffer(buf_size, '\0') { }

bool await_ready(){
 len = 0; </code></pre>
增加一次读取，若成功记录读取的长度，返回 true 继续协程
<pre class="language-cpp highlighter-hljs"><code> ssize_t n = read(fd, buffer.data(), buffer.size());
 if (n > 0) {
 len = n;
 return true; </code></pre>
非 EAGAIN 错误直接抛异常
<pre class="language-cpp highlighter-hljs"><code> } else if (n == -1 && errno != EAGAIN) {
 std::stringstream ss;
 ss << "pre read failed, error " << errno;
 throw std::runtime_error(ss.str());
 }</code></pre>
EAGAIN 无数据，返回 false 挂起协程等待
<pre class="language-cpp highlighter-hljs"><code> return false;
}</code></pre>
挂起前调用 register_io 注册协程句柄
<pre class="language-cpp highlighter-hljs"><code> void await_suspend(std::coroutine_handle<> h) {
 sched.register_io(fd, h);
}</code></pre>
在正式读取时跳过试读的长度，避免数据覆盖
<pre class="language-cpp highlighter-hljs"><code> std::string await_resume() {
 ssize_t n = read(fd, buffer.data() + len, buffer.size() - len);
 if (n == -1) {</code></pre>
若读取失败且有试读数据，直接返回试读数据
<pre class="language-cpp highlighter-hljs"><code> if (len > 0) {
 buffer.resize(len);
 return std::move(buffer);
 }</code></pre>
若非 EAGAIN 错误直接抛出异常，否则重置 n 的长度为 0，防止将 -1 加和到最终长度
<pre class="language-cpp highlighter-hljs"><code> if (errno != EAGAIN) {
 std::stringstream ss;
 ss << "read failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 n = 0;
 }</code></pre>
若成功，将结果与试读结果合并后返回给用户
<pre class="language-cpp highlighter-hljs"><code> buffer.resize(n + len);
 if (len > 0) {
 std::cout << "pre-read " << len << ", read " << n << std::endl;
 }
 return std::move(buffer);
}
};</code></pre>
主要的改动已经在代码中解读了，下面是程序运行效果：
<pre class="language-bash highlighter-hljs"><code>$ ./sample communication.pipe communication2.pipe
Read world
pre-read 30, read 0
Read world
world
world
world
world
pre-read 6, read 0
Read world
Read
Read hello
hello
hello
hello
hello
hello
hello
Read world
Read hello
hello
hello
hello
pre-read 6, read 0
Read hello
Read world
world
world
world
Read
Read world
Read
Read hello
hello
Read world
world
Read
Read hello
Read world
Read hello
Read world
Read hello
Read world
Read hello
Read world
Read hello
Read world
Read hello
...</code></pre>
新增的 pre-read 日志就是试读成功的场景，看起来发生次数并不多，可能是数据量比较小的缘故。一般在试读成功后，正式读取时就没有数据了。
总流程变为两条路径：
* 返回 true：await_ready -> await_resume
* 返回 false：await_ready -> await_suspend -> 挂起等待 -> await_resume
注意为了能在 await_ready 中修改成员 len 的内容，将接口 const 修饰符去掉了，编译器似乎对这些细节没有要求，只要函数主体签名能对得上就 ok。
 
一些细心的读者可能注意到了，<code>std::string::resize()</code> 会在扩张字符串尺寸时，将当前 size 到新 size 之间的内容重置为 '\0'，一般不适用于搭配 read 读取数据使用，之前的例子可以这样做，是基于以下几个事实：
* <code>AsyncReadAwaiter</code> 构造函数中将其初始化为最大尺寸: <code>buffer(buf_size, '\0')</code>
* 读取成功后调用 resize 属于尺寸缩小，因此不存在数据重置的问题
* 第二次读取时会重新构造一个 <code>AsyncReadAwaiter</code> 临时对象，旧的会随着作用域的结束自动析构，从而保证了 buffer 每次都始化为最大长度
新例子中 1、3 点保持了延续，第 2 点也得到了妥善的处理：
* 试读时只记录读取长度 len，不进行 resize 操作
* 正式读取时
 * 若失败，有试读内容时，直接 <code>resize(len)</code> 并返回
 * 若成功，<code>resize(n+len)</code> 并返回
换句话说，最终总能保证从最大尺寸缩小到目标尺寸，而不是分别 <code>resize(len)</code> 和 <code>resize (len+n)</code>，从而避免 size 增长和内容重置。
经过多轮测试，终于复现了一次试读与正式读取都有内容的场景：
<pre class="language-bash highlighter-hljs"><code>$ ./sample communication.pipe communication2.pipe
Read world-war
pre-read 20, read 10
Read world-war
world-war
world-war
Read
Read hello
hello
hello
hello
Read world-war
Read hello
Read
Read world-war
Read hello
hello
hello
hello
hello
Read world-war
world-war
world-war
world-war
pre-read 10, read 0
Read world-war
Read
Read
Read hello
pre-read 6, read 0
Read hello
Read world-war
Read
...</code></pre>
为了避免 hello 与 world 同长度掩盖问题，这里修改了写入 communication2.pipe 的内容为 world-war，这样在读取 hello 后再读取 world-war，size 增长了而内容没有被截断，可以证明之前的结论 1、3；在第一次 pre-read 过程中，先读取 20，后读取 10，总长度 30，size 也增长了，最终输出的内容没截断，证明了结论 2。
可以看到，使用 read 搭配 <code>std::string::resize()</code> 处理数据是非常麻烦的，不建议在真实的环境中使用，这里主要是出于便于演示的目的。
 
行文至此，本节的主角还没有登场：其实 await_suspend 这个接口也可以返回 bool 值，true 表示挂起，false 表示继续，与 await_ready 刚好相反，下面改它试试：
<pre class="language-cpp highlighter-hljs"><code> bool await_ready() const {
 return false;
}

bool await_suspend(std::coroutine_handle<> h) {
 len = 0;
 ssize_t n = read(fd, buffer.data(), buffer.size());
 if (n > 0) {
 len = n;
 return false;
 } else if (n == 0 || (n == -1 && errno == EAGAIN)) {
 sched.register_io(fd, h);
 return true;
 } else {
 std::stringstream ss;
 ss << "pre read failed, error " << errno;
 throw std::runtime_error(ss.str());
 }
}</code></pre>
修改局限于上面两个接口中，主要是将试读从 await_ready 移到了 await_suspend 中，其它没有变化；新的组织形式，让 await_ready 显得不那么臃肿了，看起来更协调和具有可读性，更推荐这种形式。
通过 C++ Insights 看下新 await_suspend 的编译器中间结果：
<pre class="language-cpp highlighter-hljs"><code> while(true) {

 /* co_await insights.cpp:129 */
 __f->__suspend_129_30 = AsyncReadAwaiter(__f->sched, __f->fd, 4096);
 if(!__f->__suspend_129_30.await_ready()) {
 if(__f->__suspend_129_30.await_suspend(std::coroutine_handle<Task::promise_type>::from_address(static_cast<void *>(__f)).operator std::coroutine_handle<void>())) {
 __f->__suspend_index = 2;
 return;
 }

 }

 __resume_async_read_file_2:
 __f->__suspend_129_30_res = __f->__suspend_129_30.await_resume();
 __f->data = __f->__suspend_129_30_res;
 std::operator<<(std::operator<<(std::operator<<(std::cout, "Read [").operator<<(__f->data.size()), "] "), __f->data);
 if(__f->data.size() == 0) {
 std::cout.operator<<(std::endl);
 }

}</code></pre>
它被放置到了 if 条件中，总的流程变为：
* 返回 true：await_ready -> await_suspend -> 挂起等待 -> await_resume
* 返回 false：await_ready -> await_suspend -> await_resume
可以期望当 await_suspend 返回 false 时，后续的 await_resume 会被立即调用。
<h1>signalfd & 完美退出</h1>
上面的 demo 目前只能通过 Ctrl C 强制杀死，毕竟调度器的 run 是个死循环没法退出。用来做做演示没问题，但是要用来开发项目就不行了，本着做出工业级强度代码的使命感，下面对它进行一番改造，看看能否实现完美退出。
核心思路是检测用户按下 Ctrl C 让 epoll_wait 感知并退出 run 循环，按下 Ctrl C 简单，等价于处理 SIGINT 信号，但让 epoll 感知比较难，查了下 deepseek 给了三种方案：
* 通过 signalfd 将信号转化为 IO 事件，交给 epoll 统一处理
* 建立一个进程内的 pipe 通道，注册到 epoll，在检测到 SIGINT 事件时写入一字节以唤醒 epoll_wait 并退出
* 信号处理器设置一个标志位，使用 epoll_wait 的超时功能，定时检测该标志位
方案 III 有延迟，首先排除；方案 II 就是传说中的 self-pipe trick，比较通用但不够高效；方案 I 最直接，也比较适合 Linux，就它了：
<pre class="language-cpp highlighter-hljs"><code>#include <signal.h>
#include <sys/signalfd.h>

class EpollScheduler {
private:
int epoll_fd;
int signal_fd;
std::unordered_map<int, std::coroutine_handle<>> io_handles;
public:
EpollScheduler(int signum) {
 epoll_fd = epoll_create(MAX_EVENTS);
 if (epoll_fd == -1) {
 std::stringstream ss;
 ss << "epoll_create failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 sigset_t mask;
 sigemptyset(&mask);
 sigaddset(&mask, signum);
 sigprocmask(SIG_BLOCK, &mask, NULL);
 signal_fd = signalfd(-1, &mask, SFD_NONBLOCK);
 if (signal_fd == -1) {
 std::stringstream ss;
 ss << "signalfd failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 struct epoll_event ev;
 ev.events = EPOLLIN;
 ev.data.fd = signal_fd;
 if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, signal_fd, &ev) == -1) {
 std::stringstream ss;
 ss << "epoll_ctl failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 std::cout << "register signal " << signum << " as fd " << signal_fd << std::endl;
}

~EpollScheduler() {
 for(auto handle : io_handles) {
 std::cout << "coroutine destroy" << std::endl;
 handle.second.destroy();
 }
 close(signal_fd);
 close(epoll_fd);
}

...

void run() {
 while (true) {
 epoll_event events = { 0 };
 int n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
 for (int i = 0; i < n; ++i) {
 int ready_fd = events.data.fd;
 if (ready_fd == signal_fd) {
 struct signalfd_siginfo fdsi = { 0 };
 read(signal_fd, &fdsi, sizeof(fdsi));
 std::cout << "signal " << fdsi.ssi_signo << " detected, exit..." << std::endl;
 return;
 }

 if (auto it = io_handles.find(ready_fd); it != io_handles.end()) {
 it->second.resume();
 }
 }
 }
}
};</code></pre>
改动主要集中在 EpollScheduler 类的构造、析构与 run 方法。内容不长，分段解读一下：
<pre class="language-cpp highlighter-hljs"><code>class EpollScheduler {
private:
int epoll_fd;</code></pre>
增加成员记录信号对应的句柄，方便后续在 epoll_wait 返回时做对比
<pre class="language-cpp highlighter-hljs"><code> int signal_fd;
std::unordered_map<int, std::coroutine_handle<>> io_handles;
public:</code></pre>
构造函数接收一个信号作为监听对象，main 中会传递 SIGINT 或 SIGQUIT
<pre class="language-cpp highlighter-hljs"><code> EpollScheduler(int signum) {
 epoll_fd = epoll_create(MAX_EVENTS);
 if (epoll_fd == -1) {
 std::stringstream ss;
 ss << "epoll_create failed, error " << errno;
 throw std::runtime_error(ss.str());
 }</code></pre>
构建信号对应的异步文件句柄
<pre class="language-cpp highlighter-hljs"><code> sigset_t mask;
 sigemptyset(&mask);
 sigaddset(&mask, signum);</code></pre>
下面这句是关键，如果不屏蔽默认的信号处理方式，默认的信号处理器会让进程退出，epoll 就没机会啦
<pre class="language-cpp highlighter-hljs"><code> sigprocmask(SIG_BLOCK, &mask, NULL);
 signal_fd = signalfd(-1, &mask, SFD_NONBLOCK);
 if (signal_fd == -1) {
 std::stringstream ss;
 ss << "signalfd failed, error " << errno;
 throw std::runtime_error(ss.str());
 }</code></pre>
将信号句柄注册到 epoll，成功时打印一条日志，失败时抛异常
<pre class="language-cpp highlighter-hljs"><code> struct epoll_event ev;
 ev.events = EPOLLIN;
 ev.data.fd = signal_fd;
 if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, signal_fd, &ev) == -1) {
 std::stringstream ss;
 ss << "epoll_ctl failed, error " << errno;
 throw std::runtime_error(ss.str());
 }

 std::cout << "register signal " << signum << " as fd " << signal_fd << std::endl;
}</code></pre>
析构除了增加信号句柄的关闭，还增加了挂起协程的销毁，如果调度器的生命周期与进程不一致时 (多次初始化与销毁调度器)，这就比较关键了，可以防止协程泄漏
<pre class="language-cpp highlighter-hljs"><code> ~EpollScheduler() {
 for(auto handle : io_handles) {
 std::cout << "coroutine destroy" << std::endl;
 handle.second.destroy();
 }
 close(signal_fd);
 close(epoll_fd);
}

...

void run() {
 while (true) {
 epoll_event events = { 0 };
 int n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
 for (int i = 0; i < n; ++i) {
 int ready_fd = events.data.fd;</code></pre>
epoll_wait 返回时，优先处理信号句柄上的事件
<pre class="language-cpp highlighter-hljs"><code> if (ready_fd == signal_fd) {
 struct signalfd_siginfo fdsi = { 0 };
 read(signal_fd, &fdsi, sizeof(fdsi));
 std::cout << "signal " << fdsi.ssi_signo << " detected, exit..." << std::endl;
 return;
 }</code></pre>
之后才是普通 IO 事件及协程的恢复
<pre class="language-cpp highlighter-hljs"><code> if (auto it = io_handles.find(ready_fd); it != io_handles.end()) {
 it->second.resume();
 }
 }
 }
}
};</code></pre>
 下面是程序运行效果：
<pre class="language-cpp highlighter-hljs"><code>$ ./sample communication.pipe communication2.pipe
register signal 2 as fd 4
Read world-war
pre-read 30, read 0
Read world-war
world-war
world-war
pre-read 10, read 0
Read world-war
Read
...
Read hello
Read world-war
Read hello
Read world-war
Read
Read hello
Read world-war
^Csignal 2 detected, exit...
coroutine destroy
coroutine destroy</code></pre>
内容比较长，中间忽略了一部分；开始的 register 日志显示新增的信号句柄值为 4；最后的 ^C 是用户按下了 Ctrl C，demo 能正常检测到信号值为 2 并退出事件循环，析构中还销毁了两个挂起的协程，符合预期。
其实不光事件循环存在完美退出的问题，单个 IO 句柄也存在同样的问题：正常的管道不可能一直读下去。当 writer 关闭管道或连接断开时，应该检测此种场景并加以处理，例如将 fd 从 epoll 中移除，从而让 IO 句柄也能完美退出。但不幸的是，目前选取的 fifo 文件，在 <code>O_NONBLOCK</code> 模式下，似乎无法感知对端关闭这种操作，传统的 read 返回 0 并不能代表这种情况，像上面的输出中，在正常的传输过程中，就会出现多次 read 返回 0 的情况，显然并不是对端关闭管道所致 (也不是 read 返回 EAGAIN 的问题，这个我加日志确认过了)。不过对于 socket，还是可以通过 read 返回 0 来检测连接断开的场景，这个就当作课外题就交给感兴趣的读者吧 ~
<h1>结语</h1>
本文介绍了一种基于真实 IO 事件驱动的协程调度器，通过特定的等待对象，实现协程在没有异步事件时挂起等待、异步事件到达时恢复运行的逻辑，更加贴近实际应用场景。除此之外，还说明了 await_suspend 与试读写、signalfd 与进程完美退出的关系等，可用于构建工业级强度的代码。
最后，由于本文中 demo 经历多次迭代，想要复制最终版的代码进行验证的小伙伴，可以 follow 这个开源 git 库获取：cpp20coroutine。
本文的 demo 是基于 Linux epoll 的，下一篇来看看怎么用 Windows 的 IOCP 实现类似的能力。
<h1>参考</h1>
. epoll_ctl : Operation not permitted error - c program
. std::string::resize() 对缓冲区一些用处
. select/poll/epoll对比分析
. Netty - 五种 I/O 多路复用机制 select、poll、epoll、kqueue、iocp(windows) 对比
. 水平触发和边缘触发

</div>
<div id="MySignature" role="contentinfo">
本文来自博客园，作者：goodcitizen，转载请注明原文链接：https://www.cnblogs.com/goodcitizen/p/18965895/epoll_based_multiplexing_coroutine_scheduler 
来源：https://www.cnblogs.com/goodcitizen/p/18965895/epoll_based_multiplexing_coroutine_scheduler

頁: [1]

琼殿技术论坛's Archiver

基于 epoll 的协程调度器——零基础深入浅出 C++20 协程