1. 先说清楚协程到底解决什么问题
C++20 这几个字加上“协程”,很多人第一反应是套一层新的异步框架去写网络服务,把回调都改成co_await,代码看起来像同步的,性能又像异步的。这个印象不算错,但它把一个更重要的东西盖住了:协程本身不是异步框架,它是一种“让编译器帮你把函数拆成状态机”的语言机制。真正决定性能、决定代码写起来难不难的,是你自己写的那几个promise_type、awaiter和背后的调度器,标准库只给了语言层面的三个关键字(co_await、co_yield、co_return),一个函数对象都不给你。
所以我写这篇东西的定位很明确:面向已经会写 C++、懂 RAII、听说过 epoll 但自己没从头搭过一个协程网络服务的开发者。前半段拆机制,讲清编译器在背后插入了什么、Task<T>为什么必须这么写;后半段落到网络服务,把 socket 读写、事件循环、超时取消、连接关闭这些真会踩坑的地方一段一段写出来。中间会给出可以直接抄的代码骨架,但我不打算糊一个能上生产的框架——那玩意儿动辄几万行,任何人贴出来你都该怀疑。我给的是“能跑起来、能压测、能让你在此基础上长出自己的东西”的最小闭环。
有一个前提要摆在最前面:这篇里的协程指无栈协程(stackless coroutine),也就是编译器把局部变量搬进堆上的帧、把挂起点编译成 switch-case 的那种实现,不是用户态线程那种有栈协程。这个区别决定了后面所有的坑:局部变量生命周期变了、栈上内存变成了堆上内存、递归式的resume会直接吃掉线程栈。
1.1 协程不是线程,也不是回调的语法糖
先把最容易混的三件事切开。线程是操作系统调度的实体,栈由内核分配,代价在上下文切换和 8MB 默认栈;回调是纯手写的控制反转,代价在状态必须手动挂在某处,写多了就是嵌套和错误处理地狱;协程站在中间——它没有自己的栈,局部变量被编译器搬进一个堆上的协程帧(coroutine frame),挂起时保留帧、恢复时接着跑,控制权交给谁由你写的await_suspend决定。
这个“没有自己的栈”是理解一切的关键。你写:
Task<int> foo() { int a = 1; // 在协程帧里,不在线程栈上 int b = co_await bar(); // 挂起点,此时线程栈上什么都不剩 co_return a + b; }a和b都在堆上的帧里,co_await挂起后,调用foo()的那个线程可以立刻去跑别的协程,线程栈被完全释放出来。这就是为什么单线程能挂住十万个连接——每个连接只占一个几百字节的帧,而不是 8MB 栈。
反过来说,代价也很明确:
- 每一帧都要分配和释放。编译器插入
operator new/operator delete,虽然标准允许编译器把分配优化掉(HALO),但它不保证,你必须自己验证。 - 你不能在协程里做栈上的事。任何“地址必须稳定”的东西,比如把局部变量的地址传给异步操作,一旦挂起就悬垂了。
- 恢复的代价是一次函数调用。协程切换本质是
resume(),比线程切换便宜一到两个数量级,但也别指望零成本,尤其当你写的是基于链表的任务队列时。
我见过最常见的误解是“用协程就能自动并行”。不能。协程是**并发(concurrency)**的语法结构,单线程跑一万个协程照样只有一个核在动。要并行还是得多线程 + 每线程一个事件循环,协程只负责让“一个线程内跑很多任务”写起来不恶心。
1.2 手写状态机与协程的账本对比
为了让你对“编译器替你做了什么”有具体感觉,我拿一个真实场景做对比:先读 4 字节长度头,再读 N 字节包体,拼起来返回。用回调或异步接口写,你得把状态挂在 context 里:
struct Conn { enum { HEAD, BODY } state = HEAD; char hdr[4]; size_t need = 0; std::vector<char> body; }; void on_readable(Conn* c) { if (c->state == Conn::HEAD) { ssize_t n = ::read(c->fd, c->hdr + c->got, 4 - c->got); // ... 更新状态、判断是否切到 BODY、可能立刻再读一次 } else { // ... 另一份几乎一样的代码 } }这段代码的问题不是难看,是状态和逻辑分家。每加一个阶段(比如压缩、校验、解包),state枚举和on_readable里的分支就多一层,而且错误处理要在每个分支里重复一遍。
换成协程,同样的逻辑是:
Task<Packet> read_packet(Socket& s) { std::array<char,4> hdr; co_await s.read_exactly(hdr); // 挂起直到读满 4 字节 uint32_t len = parse_len(hdr); std::vector<char> body(len); co_await s.read_exactly(body); // 再挂起 co_return Packet{std::move(body)}; }编译器把这两次挂起编译成两个独立的恢复入口,中间的hdr、len、body全在帧里。状态机还是那台状态机,只是不用你手写了。
| 维度 | 手写状态机 | 回调嵌套 | 无栈协程 |
|---|---|---|---|
| 状态存放 | 自己定义的 struct | 闭包捕获 / 堆对象 | 编译器生成的帧 |
| 挂起代价 | 几乎为零 | 几乎为零 | 一次resume调用 + 可能的帧分配 |
| 局部变量生命周期 | 手动管理 | 手动管理 | 编译器搬进帧,自动 |
| 出错概率 | 中,状态漏写 | 高,资源释放易漏 | 中,生命周期语义变了 |
| 调试难度 | 低,状态可打印 | 高,栈是断的 | 高,栈是断的 |
最后一行值得单独说。协程挂起后调用栈是断的,你在gdb里看到的栈可能只有resume那一层,看不到“谁在等我”。这是所有异步模型的通病,协程不背这个锅,但你得提前准备好手段——后面第五章我会讲怎么用自维护的 trace 把链路接回去。
提醒一句:别指望“改成协程”能提升吞吐量。它能提升的是代码可维护性和单线程并发连接数,纯 CPU 密集场景用协程是负优化,因为多了帧分配。
2. 拆开 promise_type:一个可运行的 Task 是怎么拼出来的
这一章要把“编译器到底插了什么”摊开。你写一个含co_await的函数,编译器会把函数体改造成一个类似这样的结构:
{ coroutine_frame* f = new coroutine_frame(...); // 或栈上/优化掉 f->promise = promise_type{}; auto obj = f->promise.get_return_object(); // 返回给调用方 co_await f->promise.initial_suspend(); // 决定是否立刻开始 try { /* 你的函数体,每个 co_await 变成一次挂起+恢复点 */ } catch (...) { f->promise.unhandled_exception(); } co_await f->promise.final_suspend(); // 收尾 }记住这五行,promise_type里那堆看起来莫名其妙的成员函数,全都是被这五行调用的。get_return_object决定Task对象长什么样,initial_suspend决定协程是“调用即执行”还是“调用即挂起”,final_suspend决定结束时把控制权交给谁,unhandled_exception决定异常是吞掉、存起来还是重新抛出。
2.1 编译器在 co_await 背后插入的三个调用
co_await expr会被展开成对一个 awaiter 的三次调用:
auto&& aw = expr; // 拿到 awaiter if (!aw.await_ready()) { // 1. 是否已经就绪 auto h = aw.await_suspend(current); // 2. 挂起,可能返回 handle // 若返回 void:当前协程挂起,控制权回到 resume 的调用者 // 若返回 bool:false 表示不挂起,继续执行 // 若返回 handle:对称转移,立刻跳去执行那个 handle /* 恢复点:等到有人 resume(current) 才从这往下走 */ } T result = aw.await_resume(); // 3. 取结果,异常在这里抛三个返回值的语义差别极大,而网上大多数教程只讲bool。我当初就在这里绕了很久,直到自己实现调度器才明白:
void:无条件挂起,控制权交还给resume()的调用者。适合“把任务丢进队列就走人”的调度器。bool:true挂起,false不挂起继续跑。适合“可能立刻完成”的操作,比如缓冲里还有数据就直接返回,省一次调度。coroutine_handle<>:轻量级对称转移(symmetric transfer)。不是返回调用者,而是当前线程直接跳到那个 handle 去执行,不增长栈。
第三种是性能关键。假设 A 等 B,B 等 C,如果 A 的await_suspend返回void,那就是 A resume 了线程栈上的 scheduler,scheduler 再 resume B,B 再嵌套 resume C——每层等待多吃一帧线程栈,链一长就爆。用对称转移,A 直接把控制权转给 B,B 转给 C,栈深度恒定。
2.2 手写一个能跑的最小 Task
下面这份代码我删掉了void特化、取消支持这些扩展,只保留骨架。它能在单线程里地跑起来,编译要求 C++20 及以上。
#include <coroutine> #include <optional> #include <exception> #include <utility> template <typename T> class Task { public: struct promise_type { std::optional<T> value_; std::exception_ptr eptr_; std::coroutine_handle<> continuation_{}; // 谁在等我 struct FinalAwaiter { bool await_ready() const noexcept { return false; } std::coroutine_handle<> await_suspend( std::coroutine_handle<promise_type> h) noexcept { auto c = h.promise().continuation_; // 关键:空 handle 不能返回,用 noop_coroutine 垫底 return c ? c : std::noop_coroutine(); } void await_resume() const noexcept {} }; Task get_return_object() noexcept { return Task{std::coroutine_handle<promise_type>::from_promise(*this)}; } std::suspend_always initial_suspend() noexcept { return {}; } FinalAwaiter final_suspend() noexcept { return {}; } void return_value(T v) { value_.emplace(std::move(v)); } void unhandled_exception() noexcept { eptr_ = std::current_exception(); } }; using handle_t = std::coroutine_handle<promise_type>; explicit Task(handle_t h) noexcept : h_(h) {} Task(Task&& o) noexcept : h_(std::exchange(o.h_, {})) {} Task& operator=(Task&& o) noexcept { if (this != &o) { if (h_) h_.destroy(); h_ = std::exchange(o.h_, {}); } return *this; } Task(const Task&) = delete; Task& operator=(const Task&) = delete; ~Task() { if (h_) h_.destroy(); } // co_await Task 时用 bool await_ready() const noexcept { return !h_ || h_.done(); } std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaiting) noexcept { h_.promise().continuation_ = awaiting; return h_; // 对称转移:直接去跑被等待的协程 } T await_resume() { if (h_.promise().eptr_) std::rethrow_exception(h_.promise().eptr_); return std::move(*h_.promise().value_); } private: handle_t h_; };写完之后有个细节特别容易漏:await_suspend里return h_;之前,你没有先resume它,因为对称转移的动作由编译器帮你做。如果误写成h_.resume(); return true;,在深链场景下就退化成嵌套调用,栈会一层层涨——这是我从一次压测里 2000 层递归直接栈溢出才反应过来。
另一个坑在await_ready。我写了!h_ || h_.done(),看起来多余,其实是为了防止“await 一个已经被销毁或已完成的 Task”。如果漏掉h_.done(),一个已经跑完的协程再被 await 一次,await_resume里读value_就是空 optional,std::move(*value_)是未定义行为,在某些优化档位下读到的是垃圾值,而且不崩——比崩溃更难查。
2.3 值语义、移动与“谁负责销毁帧”
Task的拥有权必须唯一。我上面禁了拷贝、只留移动,析构里destroy()。但这里有一条硬约束:只能销毁已挂起或已完成的协程帧,不能销毁正在执行的。如果Task对象在协程运行中途被析构(比如外层作用域提前退出),destroy()释放帧的同时,正在跑的代码还在访问帧里的变量,这就是标准里明确的未定义行为。
我实际采用的做法是两条:
第一,Task只作为“生产的返回值”和“awaited 的临时对象”存在,生命周期不超过一个完整表达式,不往容器里塞裸Task。要存起来排队的,存std::coroutine_handle<>,配一个自定义的池子管理帧。
第二,final_suspend用了suspend_always(上面FinalAwaiter的await_ready恒返 false)。这意味着协程跑完后停在结尾不销毁,等Task析构或显式destroy(),这样调用方总能在await_resume里安全地取到结果。代价是帧必须有人回收——如果你让协程自持有、无人 await,那就永远漏一帧。
记住这条:
final_suspend返回suspend_always时,帧的销毁责任在Task;返回suspend_never时,协程结束即自毁,但你再也不能安全地取结果。二选一,没有中间地带。
3. 把协程接到事件循环上:网络服务的实操落地
协程本身不会做 IO,它只会挂起和恢复。要让co_await s.read(buf)真正等到数据,你需要一个反应器:把 fd 注册到内核事件通知机制上,事件到了之后resume对应的协程 handle。整个网络服务的骨架就是“事件循环 + 一堆挂起的 handle”。
3.1 事件循环骨架:注册、等待、恢复
我用的结构大致是这样,删掉了日志和统计,只留主干。这里用 Linux 的epoll,因为它足够成熟,语义也简单:每次epoll_wait返回一批就绪 fd,你逐个resume对应的协程。
#include <sys/epoll.h> #include <sys/eventfd.h> #include <unistd.h> #include <unordered_map> #include <vector> #include <deque> class Reactor { public: Reactor() : epfd_(::epoll_create1(EPOLL_CLOEXEC)), wakefd_(::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) { epoll_event ev{}; ev.events = EPOLLIN; ev.data.fd = wakefd_; ::epoll_ctl(epfd_, EPOLL_CTL_ADD, wakefd_, &ev); } ~Reactor() { ::close(wakefd_); ::close(epfd_); } // 把协程挂到 fd 上,等 EPOLLIN void await_readable(int fd, std::coroutine_handle<> h) { waiters_[fd] = h; epoll_event ev{}; ev.events = EPOLLIN | EPOLLONESHOT; // 一次性,触发后自动失效 ev.data.fd = fd; ::epoll_ctl(epfd_, EPOLL_CTL_MOD, fd, &ev); // 前提是 fd 已 ADD 过 } // 跨线程唤醒:别的线程 resume 不了本线程的协程,必须投递 void post(std::coroutine_handle<> h) { { std::lock_guard<std::mutex> lk(mu_); ready_.push_back(h); } uint64_t one = 1; ::write(wakefd_, &one, sizeof(one)); } void run() { epoll_event evs[256]; for (;;) { int n = ::epoll_wait(epfd_, evs, 256, -1); for (int i = 0; i < n; ++i) { int fd = evs[i].data.fd; if (fd == wakefd_) { drain_wakeup(); continue; } auto it = waiters_.find(fd); if (it != waiters_.end()) { auto h = it->second; waiters_.erase(it); h.resume(); // 恢复后协程自己决定下一步 } } flush_ready(); } } private: void drain_wakeup() { uint64_t v; while (::read(wakefd_, &v, 8) == 8) {} } void flush_ready() { std::vector<std::coroutine_handle<>> local; { std::lock_guard<std::mutex> lk(mu_); local.swap(ready_); } for (auto h : local) h.resume(); } int epfd_, wakefd_; std::unordered_map<int, std::coroutine_handle<>> waiters_; std::deque<std::coroutine_handle<>> ready_; std::mutex mu_; };三个设计选择值得说明,都是踩过坑之后的结论:
一,EPOLLONESHOT必须开。不开的话,一个 fd 上有新数据时epoll_wait会持续返回它,你 resume 一次、协程读走数据、再次挂起,注册的还是同一个 fd,很容易在“读空 → 挂起 → 立刻又被通知 → 读到 EAGAIN”之间空转烧 CPU。开了 oneshot,事件触发后该 fd 自动从关注集合里摘掉,等你下次重新EPOLL_CTL_MOD才会再通知,天然匹配协程“挂起-恢复”的模型。
二,waiters_用fd → handle一对一是简化版。真要做同时读写,一个 fd 上可能有两个挂起的协程,得用std::pair<handle, handle>或者把事件类型编码进data.u64。我一开始图省事写成一对一,结果在“一边收一边发”的连接上直接出现写协程占掉了读协程的槽,表现为对端数据卡住不处理。
三,跨线程恢复必须投递,不能直接resume。协程帧不属于任何线程,但resume会让它在调用者的线程上执行。如果 A 线程的协程被 B 线程resume,那段代码就跑在 B 线程上,而它以为自己在 A 线程的事件循环里,访问 A 线程的无锁结构——数据竞争。所以跨线程只允许post,让目标线程自己在flush_ready里恢复。
3.2 把 socket 变成可 co_await 的对象
有了注册接口,ReadAwaiter就很好写了。核心是把“发起读”和“读完之后取结果”拆到await_suspend和await_resume里:
struct ReadAwaiter { Reactor& loop; int fd; void* buf; size_t len; ssize_t n_ = 0; int err_ = 0; bool await_ready() const noexcept { return false; } // 一律走挂起路径 void await_suspend(std::coroutine_handle<> h) { loop.await_readable(fd, h); // 先注册再看,避免丢事件 } ssize_t await_resume() { // 真正读放在恢复之后:事件到了,数据肯定有,此时读不会阻塞 n_ = ::read(fd, buf, len); if (n_ < 0) err_ = errno; return n_; } };这里有个顺序问题,网上很多示例是反的:在await_ready里先试读一次,EAGAIN 了再在await_suspend里注册。这个写法性能更好(省一次挂起),但必须用EPOLLONESHOT或边沿触发,否则“试读失败 → 注册 → 数据在这两步之间到达 → 事件已丢失 → 永远等下去”。我建议新手先用“无条件挂起”的版本,逻辑清晰,等压测数据出来再考虑加快速路径。
更麻烦的是短读。TCP 是字节流,read返回 3 字节只代表现在有 3 字节,不代表你要的 100 字节都到齐了。read_exactly必须写成循环:
Task<void> read_exactly(int fd, char* p, size_t n, Reactor& loop) { size_t got = 0; while (got < n) { ssize_t r = co_await ReadAwaiter{loop, fd, p + got, n - got}; if (r == 0) throw std::runtime_error("peer closed"); if (r < 0) { if (errno == EINTR) continue; // 被信号打断,重试 if (errno == EAGAIN) continue; // 假唤醒,重新挂起 throw std::system_error(errno, std::generic_category()); } got += static_cast<size_t>(r); } }EINTR和EAGAIN这两个分支千万别省。前者在收到信号时必然出现,一跳出循环就当成错误,线上会莫名其妙断连;后者在epoll水平触发 + 多协程抢同一个 fd 时会出现,直接当错误处理同样会误杀连接。
3.3 连接生命周期:关闭顺序错了就是崩溃
协程网络服务里最容易出人命的不是读写,是关闭。考虑一个连接对象Conn,它的run()是个协程,里面无限co_await read_packet。你现在要关它,得同时处理三件事:
- 从
epoll摘掉 fd,防止后面还触发事件; - 取消所有挂在这个连接上的协程(读协程、写协程、定时器);
- 保证没有协程还在引用
Conn的成员。
顺序错了就是经典的 use-after-free。我吃过一次:先close(fd),再让读协程自然返回,读协程返回时访问Conn::buffer_,此时Conn已经被shared_ptr释放——线上表现为随机段错误,复现概率很低,因为要和内核事件到达的时序撞上。
后来我固定成这个顺序,实测稳定:
void close() { if (closed_.exchange(true)) return; // 1) 先摘事件,此时不会再有新的 resume 投递进来 loop.unregister_read(fd_); // 2) 把 fd 关掉,让所有阻塞中的系统调用立刻返回 EBADF/0 ::close(fd_); // 3) 主动恢复挂起的协程,它会看到 closed_ 标志然后退出 if (read_handle_) { read_handle_.resume(); read_handle_ = {}; } // 4) 此时已经没有协程引用 this,才能销毁对象 }第 3 步是关键。协程不会自己醒,你不resume它就永远挂着,帧永远不释放,连接对象也就永远不能释放——这就是协程版的内存泄漏。很多“连接数一直涨、内存一直涨”的线上问题,根源都是漏了这一步的主动唤醒。
至于Conn本身,我用std::shared_ptr让协程持有一份,await_resume里访问成员前用weak_ptr升级检查——但这属于防御性写法,性能有损耗。更干净的做法是让所有成员访问都只在协程体内、且在closed_检查之后进行,约定胜于运行时检查。
4. 内存、异常与取消:三个最容易翻车的地方
4.1 HALO 靠不住,得自己管分配
标准里协程帧的分配走promise_type::operator new,没定义就用全局的。标准鼓励编译器做分配消除(HALO),但没有任何保证。我在 1 万连接、每连接每秒 2 次co_await的压测里对比过:默认operator new的情况下,perf里 malloc/free 占了 11% 的采样;把operator new换成线程本地定长池之后降到 1.4%。
做法是给promise_type加一对静态函数:
struct promise_type { static void* operator new(size_t sz) { return FramePool::instance().alloc(sz); } static void operator delete(void* p, size_t sz) { FramePool::instance().free(p, sz); } // ... 其余成员 };池子按 size class 分桶,比如 64/128/256/512 字节四档,超出的回落到::operator new。注意operator delete的size_t参数是个技巧:如果你只定义operator delete(void*),编译器拿不到大小,池子就还不了。C++14 之后支持带 size 的版本,协程帧这块正好用得上。
有个限制要提醒:协程帧的大小由编译器决定,你没法精确控制。加一个成员变量可能让它从 128 跳到 129,直接换桶。所以上线前一定要打日志统计实际帧大小分布,别照抄别人的桶配置。
4.2 异常必须接住,不然整个循环挂掉
unhandled_exception()里如果不处理,异常会一路传播到resume()的调用点,也就是事件循环里。事件循环里抛异常 = 整个服务停摆。所以我的规范是:unhandled_exception里只存exception_ptr,绝不重新抛出,然后两处消费它——await_resume里rethrow,以及一个全局的“兜底协程”在resume外层try/catch。
void run() { // ... try { h.resume(); } catch (const std::exception& e) { log_error("coroutine escaped: {}", e.what()); // 不做任何恢复动作,让这个连接自然断开 } }resume外层这个 catch 是最后一道防线。它不该被触发,一旦被触发就说明某个协程的Task链条中间有地方漏了await_resume——典型症状是“日志里出现 escape,但业务逻辑看起来没问题”,因为异常被吞了,返回值是坏的。
还有一种情况要单独处理:协程里co_await一个已经挂掉的对象。比如Task被析构了才有人 await 它,await_suspend里访问h_.promise()就是访问已释放内存。所以我加了await_ready里那个!h_判断——虽然这只能挡住空 handle,不能挡住悬垂 handle。
4.3 取消:协程没有内置取消,得自己造
C++20 协程没有取消机制。这是个设计缺口,所有人都在自己造。最轻的方案是共享一个std::atomic<bool> cancelled,在每个可能的挂起点检查:
Task<void> handle_conn(std::shared_ptr<Conn> c) { while (!c->cancelled.load(std::memory_order_relaxed)) { auto pkt = co_await read_packet(*c); co_await process(c, pkt); } }但这只解决“轮询检查”,解决不了“卡在co_await里出不来”。一个挂在epoll_wait上的读协程,你置了标志位它也不知道,因为没人resume它。真正的取消需要三件事配合:
| 环节 | 要做的事 | 不做会怎样 |
|---|---|---|
| 标志位 | 每个连接一个 atomic flag | 上层无法通知下层 |
| 唤醒 | 取消时主动resume挂起协程 | 协程永远挂着,帧泄漏 |
| 清理 | 从 epoll 摘除 + close fd | 事件持续触发,空转 |
超时是取消的特例。我用一个最小堆管定时器,每轮循环取堆顶和epoll_wait的 timeout 取小值:
int timeout = -1; if (!timers_.empty()) { auto now = steady_now_ms(); timeout = std::max<int64_t>(0, timers_.top().deadline - now); } int n = ::epoll_wait(epfd_, evs, 256, static_cast<int>(timeout)); // 时间到了的定时器,取出对应 handle,resume 并让它看到 timeout 标志这里有个我自己写错过的地方:定时器到期时不能直接resume就完事,因为那个协程可能已经因为别的原因被恢复了。同一个 handleresume两次是灾难性的(第二次会从上次的挂起点继续跑,逻辑全乱)。我在定时器节点里存 handle + 一个 version 号,恢复前比对 version,不匹配就丢弃。
如果做的事比较复杂,别自己造
Task,去看看已有的协程库,把精力放在业务上。自己写一套能用和写一套不出事,中间差的是十几万行边界处理。
5. 常见问题排查实录
5.1 悬垂引用三连:局部变量、this、引用参数
协程里第一大类崩溃都叫“悬垂”,本质是“挂起期间有东西被销毁了”。我把遇到过的归成三类:
第一类,把局部变量地址传出去。co_await s.write(buf)里buf是栈上临时数组,如果write的实现只存了指针、真正发送在恢复之后,那挂起期间数组已经跟着作用域没了。修法是:任何跨越挂起点的缓冲区,都必须是协程帧里的成员(值捕获)或者调用方持有的对象。
第二类,this悬垂。成员协程挂起时,对象被另一个线程释放,恢复后访问this->member_就是野指针。这个必须靠所有权设计解决,shared_from_this是最直接的方案,代价是一次原子操作。
第三类,引用参数。Task<void> foo(const std::string& s)看着没问题,但如果这个Task没有立刻被 await(比如存进队列),s的引用在挂起前就失效了。协程参数永远按值传,这是我现在的硬规矩。
5.2 问题速查表
| 现象 | 最可能的原因 | 定位方法 | 处理方式 |
|---|---|---|---|
| 连接不涨但内存持续增长 | 挂起的协程帧没人回收 | 统计operator new/delete配对 | 检查关闭路径是否主动 resume |
| 随机段错误,复现率低 | 悬垂引用 / 关闭时序 | 打开 ASan,重点看destroy附近 | 固定关闭顺序,先摘事件再 close |
| 单核 CPU 100%,吞吐不涨 | fd 空转(漏 EPOLLONESHOT) | strace -c看 read/epoll 次数 | 加 oneshot 或边沿触发 |
| 压测时栈溢出 | 递归 resume 链 | gdb看栈深度 | await_suspend改返回 handle |
| 偶发“读到了脏数据” | await_ready漏判done() | 在await_resume打点 | 补done()检查 |
| 跨线程后偶发数据错乱 | 跨线程直接resume | 用线程 ID 打点验证 | 改成post投递 |
| 定时器误杀正常连接 | handle 被重复 resume | 记录 resume 次数 | 加 version 号校验 |
这张表里我最想强调第一行和最后一行,因为它们都属于“不崩溃、只是慢慢烂掉”的问题,往往上线一周才被发现,排查成本极高。
6. 性能实测与几个不那么显然的取舍
6.1 压测怎么设计才有意义
我搭的测试环境很简单:一台 8 核机器,服务端单线程事件循环 + 协程,客户端用多进程短连接打 echo。变量控制在三个:
- 协议解析复杂度:从“读到什么回什么”到“4 字节头 + 变长体 + 校验”,看协程嵌套层数对吞吐的影响;
- 帧分配策略:默认
operator newvs 线程本地池; await_suspend返回类型:voidvshandle(对称转移)。
数据我就不贴具体数字了,机器差异太大,贴了反而误导。但趋势是稳定的:
单线程 1 万长连接、每连接每秒 50 次小包 echo,默认分配下能跑到的 QPS,换成帧池之后提升约 15% 到 25%,连接数越多、每连接协程切换越频繁,差距越大。而对称转移 vsvoid的差距在小规模下看不出来,链深度超过 5 层之后开始明显——因为void版本每层多压一帧调度器调用,缓存局部性变差。
协议解析这个变量最反直觉:解析逻辑从 1 层加到 3 层嵌套协程,吞吐只掉了 4% 左右。这说明帧分配和调度的成本远大于“多几个挂起点”。所以优化顺序应该是:先治帧分配,再看调度频率,最后才去扣协程嵌套的层数。
6.2 什么时候不该用协程
这部分是我最想写给“准备上手”的人看的,因为宣传里很少提。
短连接、超高 QPS 的场景,协程收益有限。每连接一个帧,连接存活 2 毫秒,帧分配和销毁的开销占比就上来了。这种场景用固定大小的连接池 + 手写状态机,往往比协程更快、内存更可控。
纯计算密集任务,协程是负债。挂起点在数学计算里没有意义,加上去只是多了帧分配。
团队里没人愿意读生成代码的,慎用。协程出问题的时候,gdb里看到的是resume之后的一层帧,看不出调用关系。我现在的做法是给每个Task起名字,在get_return_object里注册到一张线程本地的映射表,出问题的时候打印“当前挂起的协程链”,这比任何断点都好使。
要跑在资源受限的嵌入式环境,先量帧大小。一个协程帧动辄一两百字节,几百个并发就把 RAM 吃光了,而且operator new在那些平台上可能根本没有。
最后分享一个我一直在用的小技巧:给promise_type加一个int line_;字段,在get_return_object里塞入__builtin_LINE()之类的信息(或者干脆用一个宏包一层),崩溃时打印“最后挂起在哪个源文件第几行”。这一行代码的成本,比我后面买咖啡查 bug 的时间便宜太多了。