协程

系列导航：入门篇 | 进阶篇 | 实战篇 前置知识 阅读本篇前，请确保已理解 入门篇 中的以下概念： io_context 的作用和 run() 执行流程 异步操作的生命周期（发起 → 完成 → handler 执行） post/dispatch 的区别 1. TCP 编程：三步演进 我们通过构建一个 Echo Server（收到什么就回什么），从最简单的同步版本逐步演进到生产级协程版本。 1.1 第一步：同步阻塞版 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 // echo_server_sync.cpp // 编译: g++ -std=c++20 echo_server_sync.cpp -lboost_system -lpthread -o echo // 测试: 另开终端 nc localhost 9999，输入文字会回显 #include <boost/asio.hpp> #include <iostream> using boost::asio::ip::tcp; int main() { boost::asio::io_context ioCtx; // 创建 acceptor：监听 TCP 连接 // 参数：io_context, 绑定地址(IPv4, 端口9999) tcp::acceptor acceptor(ioCtx, tcp::endpoint(tcp::v4(), 9999)); std::cout << "同步 Echo Server 监听端口 9999\n"; while (true) { // accept() 阻塞，直到有客户端连接 tcp::socket socket(ioCtx); acceptor.accept(socket); std::cout << "客户端连接: " << socket.remote_endpoint().address().to_string() << ":" << socket.remote_endpoint().port() << "\n"; // 处理这个连接（阻塞：处理期间无法接受新连接！） boost::system::error_code ec; char buf[1024]; while (true) { // read_some：读取可用的数据（可能只有一部分） size_t n = socket.read_some(boost::asio::buffer(buf), ec); if (ec == boost::asio::error::eof) { std::cout << "客户端断开\n"; break; } if (ec) throw boost::system::system_error(ec); // 将收到的数据原样写回 boost::asio::write(socket, boost::asio::buffer(buf, n)); } } return 0; } 问题：同一时刻只能服务一个客户端。当客户端 A 连接后，客户端 B 必须等 A 断开才能被接受。 ...

IO 与无锁序列：cppcoro 的网络文件 I/O，以及 LMAX Disruptor 的协程化 本专栏文章：拆开 cppcoro 给你看 · 第 6 篇（完结篇） 这是系列的最后一篇，也是最"硬"的一篇。要搞懂两件事：① cppcoro 怎么把 Windows IOCP 封装成协程友好的文件/网络 I/O 接口；② cppcoro 借鉴 LMAX Disruptor 的无锁序列怎么协程化，让消费者不是忙等而是挂起等待。 注：IOCP 基础、win32_overlapped_operation CRTP 模式、四态取消状态机、cancellation_token 三层模型、io_service 和 async_scope 在上一篇 Layer 3 中已经讲过了。本篇聚焦文件 I/O 类型层次、socket 的协程封装，以及无锁序列原语的设计推导。 1. 文件 I/O 类型体系 1.1 类层次结构 1 2 3 4 5 6 file (基类: size(), 持有 Windows HANDLE) ├── readable_file (抽象: read(offset, buffer, size)) │ └── read_only_file (具体类: open() 工厂方法) ├── writable_file (抽象: write(offset, buffer, size), set_size()) │ └── write_only_file (具体类: open() 工厂方法) └── read_write_file (多继承: readable_file + writable_file) 1.2 为什么用静态工厂？ 1 2 3 4 5 static read_only_file open( io_service& ioService, const path& path, file_share_mode shareMode = file_share_mode::read, file_buffering_mode bufferingMode = file_buffering_mode::default_); Windows 上打开文件涉及多个系统调用（CreateFile + CreateIoCompletionPort），且可能失败。工厂方法把全部设置逻辑封装在一处，返回值语义对象，调用者负责生命周期。 ...

协程调度器到底在调度什么？从 inline_scheduler 到工作窃取线程池 本专栏文章：拆开 cppcoro 给你看 · 第 5 篇 前几篇我们一直在协程"内部"兜圈子——task 怎么设计、同步原语怎么写、并发怎么编排。但有一个问题一直没认真回答：协程在哪个线程上执行？ co_await 之后你可能在任何线程上恢复——这取决于谁调了 handle.resume()。大部分时候这不是问题，但有时你确实想控制恢复的线程。这就是调度器的概念。 本文分三层递进： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 Layer 1: 调度器概念（30 分钟能读完） ├─ inline_scheduler — 22 行，不调度也是调度 ├─ round_robin_scheduler — 125 行，对称转移调度的杰作 └─ schedule_on vs resume_on Layer 2: 工作窃取线程池（核心） ├─ 本地 LIFO 队列 ├─ 全局 MPSC 队列 ├─ 偷取 (work-stealing) └─ 睡眠/唤醒协议 Layer 3: IOCP + 取消 + 文件/网络 I/O ├─ OVERLAPPED 嵌入 awaiter (CRTP) ├─ 可取消操作的四态状态机 ├─ cancellation_token/source/registration ├─ io_service + io_work_scope └─ async_scope Layer 1：调度器概念 问题：协程在哪个线程上执行？ 1 2 3 4 5 6 task<> my_coro() { std::cout << "I'm on thread " << std::this_thread::get_id() << std::endl; co_await some_io(); std::cout << "Now I'm on thread " << std::this_thread::get_id() << std::endl; // ↑ 恢复后可能在完全不同的线程上！ } 1.1 inline_scheduler —— “不调度"也是调度 1 2 3 4 5 6 class inline_scheduler { public: std::suspend_never schedule() const noexcept { return {}; // 不挂起，原地继续 } }; 返回 suspend_never 意味着 co_await scheduler.schedule() 等价于什么都不做。但它在泛型代码中有实际意义——你的函数接受一个调度器参数，inline_scheduler 就是"不需要调度的调度器”。 ...

从 generator 到 async_generator：协程生成器的三层进化 本专栏文章：拆开 cppcoro 给你看 · 第 4 篇 前几篇一直在讲"等一个结果返回"的协程模式——task<T>、同步原语、when_all。但协程还有另一面：用协程产生一系列值，而不是一次返回一个结果。 cppcoro 为此提供了三种生成器，每一种都是为解决前一种的瓶颈而生的。本文从一个具体痛点出发：遍历一棵二叉树的所有节点，看着普通的 generator 怎么在递归场景下性能退化成 O(N²)，然后 recursive_generator 怎么用 O(1) 的 pull() 解决，最后 async_generator 怎么让生成器支持 co_await。 1 2 3 generator<T> → O(1) 遍历平面序列，但递归时 operator++() 退化成 O(depth) recursive_generator<T> → pull() 直接驱动叶子，递归场景 O(1) async_generator<T> → 支持 co_await，值可以异步产生 1. generator<T> —— 同步惰性序列 1.1 最简单的使用场景 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 generator<int> fibonacci() { int a = 0, b = 1; for (int i = 0; i < 10; ++i) { co_yield b; int t = a; a = b; b += t; } } int main() { for (int n : fibonacci()) { std::cout << n << " "; // 1 1 2 3 5 8 13 21 34 55 } } 1.2 和 task<T> 的设计对比 特性 task<T> generator<T> 用途 产生一个最终结果 产生一系列中间值 关键字 co_return co_yield 可以用 co_await？ ✅ ❌ final_suspend FinalAwaiter（转移控制权） suspend_always 谁决定何时结束 协程自己 (co_return) 调用者 (不再调用 ++it) 1.3 为什么 final_suspend 是 suspend_always？ 与 task<T> 的 FinalAwaiter（把控制权转回等待者）不同，generator 的 final_suspend 就是纯纯的 suspend_always。 ...

协程世界的并发编排：从需求反推 when_all 的实现 本专栏文章：拆开 cppcoro 给你看 · 第 3 篇 这篇文章我想换个写法。前两篇是"先给答案再解释"，这篇反过来——从需求出发，一步步推到最终实现。 需求很简单：三个查询并发执行，全部完成后取结果。但从这个需求到最终的 when_all，踩了三个坑：顺序等待太慢 → 手动管理太繁琐 → 注册竞态太难搞。我们逐个填。 1. 场景引入：三个并发查询 1 2 3 task<User> load_user(int id); task<Order> load_orders(int userId); task<Address> load_address(int userId); 尝试 1：顺序等待（慢） 1 2 3 4 5 6 task<void> handle(int userId) { auto user = co_await load_user(userId); // ~200ms auto orders = co_await load_orders(user.id); // ~300ms auto address = co_await load_address(user.id); // ~100ms // 总耗时: ~600ms——三个没有依赖的操作却串行执行了 } 尝试 2：手动管理 counter（繁琐） 每次需要并发时都要手动写计数器逻辑——大量重复代码，容易出错。 我们需要一个泛化的并发执行工具。这就是 when_all 的由来。 2. 核心问题：怎么知道"所有任务都完成了"？ 答案就是上一篇文章提到的 when_all_counter： 1 2 3 4 class when_all_counter { std::atomic<std::size_t> m_count; std::coroutine_handle<> m_awaitingCoroutine; }; 让我们从头推导它的设计。 Step 1：计数器初始化 1 when_all_counter counter(3); // 3 个任务 Step 2：每个任务完成时递减 1 2 3 4 5 6 void notify_awaitable_completed() noexcept { if (m_count.fetch_sub(1, std::memory_order_acq_rel) == 1) { // 我是最后一个完成的 → 唤醒等待者 m_awaitingCoroutine.resume(); } } Step 3：等待者注册自己 1 2 3 4 5 6 bool try_await(std::coroutine_handle<> awaitingCoroutine) noexcept { m_awaitingCoroutine = awaitingCoroutine; return m_count.fetch_sub(1, std::memory_order_acq_rel) > 1; // >1 → 还有任务没完成 → 挂起 // ==1 → 所有任务在注册前就完成了 → 不挂起 } 为什么 m_count 初始 = 任务数 + 1？ 💡 这是本文第一个关键洞察——额外的那 1 票代表"注册尚未完成"。 ...

用一个生产者-消费者场景，把 cppcoro 的 7 个协程同步原语全部串起来 本专栏文章：拆开 cppcoro 给你看 · 第 2 篇 上一篇文章我把 cppcoro 的 task<T> 拆成了 5 个版本迭代。今天要搞定协程世界里另一个核心问题：多个协程之间怎么同步？ 市面上的做法是一张 API 列表——“这是 async_mutex，这是 async_latch，啥时候用自己看着办”。我不这么讲。本文用同一个生产者-消费者场景，从简单到复杂逐步升级，每级只比上一级多一个特性，一共串起 7 个同步原语： 1 2 3 4 5 6 7 8 9 场景：生产者协程产生数据，消费者协程处理数据 Level 1: 单消费者，发信号通知 → single_consumer_event Level 2: 同上，但信号自动复位 → single_consumer_async_auto_reset_event Level 3: 多消费者，手动复位 → async_manual_reset_event Level 4: 多消费者，自动复位 → async_auto_reset_event Level 5: 共享数据的互斥访问 → async_mutex Level 6: 等待 N 个操作完成 → async_latch Bonus: when_all 的核心组件 → when_all_counter 每级都是可运行的代码。 ...

cppcoro 的 task<T> 到底做了什么？从 25 行到 90 行，5 版迭代拆开看 本专栏文章：拆开 cppcoro 给你看 · 第 1 篇 上一篇我们手写了 30 行的 Generator，理解了协程帧、co_await 的展开步骤和 promise_type 的 7 个接口。今天来真的——把 cppcoro 的 task<T> 拆开看。 cppcoro 的 task.hpp 一共 482 行，但你不需要直接从第 1 行读到第 482 行。其中差不多 300 行是关于对称转移/非对称转移的条件编译、MSVC 的 workaround、异常处理细节。真正核心的东西不到 100 行。 本文的学习策略是分 5 版迭代——每个版本只加一个功能点，每版都是可编译运行的完整代码。 版本 功能 行数 v1 只能 co_return，不能 co_await ~25 v2 加上 co_await 支持（用 sync_wait 驱动） ~40 v3 加上对称转移（final_awaitable） ~55 v4 加上异常处理 ~70 v5 完整版——对应 cppcoro 源码 ~90 v1：只能 co_return 的 task v1 的目标很简单：理解一个协程怎么把结果从协程帧传回调用者。 ...

从 0 到 1 理解 C++ 协程：30 行代码搞定协程帧、co_await 和 promise_type 本专栏文章：拆开 cppcoro 给你看 · 第 0 篇 如果你写过 C++ 异步代码，一定踩过回调地狱的坑——3 层嵌套是起步，5 层不稀奇。C++20 的协程本该解决这个问题，但市面上的入门材料不是太浅（“协程就是可以暂停的函数”——然后呢？），就是太深（上来就讲对称转移和 IOCP），中间缺了一环。 这篇文章补上这一环。读完之后你不需要看任何其他"协程入门"材料，可以直接开始拆 cppcoro 的 task<T> 源码。 1. 协程到底解决了什么问题？ 1.1 先看一段让你血压升高的代码 假设你要从数据库查用户、再查订单、最后发网络请求——三个操作都是异步的。传统回调写法长这样： 1 2 3 4 5 6 7 8 9 10 11 12 // 回调地狱（callback hell） void handle_request(int userId) { db.query_user(userId, [](User user) { // 回调1 db.query_orders(user.id, [](Orders orders) { // 回调2 http.send(orders, [](Response rsp) { // 回调3 // 三层嵌套...实际项目里可能是五六层 // 错误处理散落各处，上下文信息全部丢失 finalize(rsp); }); }); }); } 你发现没有——代码逻辑明明是线性的（查用户 → 查订单 → 发请求），但写出来是嵌套的。三层还算友好，加到五六层的时候，你根本不想看自己的代码。 ...