前言
最近的新闻里 C++20 已经确认的内容里已经有了协程组件，之前都是粗略看过这个协程草案。最近抽时间更加系统性的看了下接入和实现细节。
我的测试代码都是在MSVC下开启 /await 选项后测试的，在我本地的Linux clang环境中，可以通过 $LLVM_CLANG_PREFIX/bin/clang++ -std=c++2a -O0 -g -ggdb -stdlib=libc++ -fcoroutines-ts -lc++ -lc++abi -Wl,-rpath=$LLVM_CLANG_PREFIX/lib/ test.cpp 编译和运行。
在gcc 10+中，可以使用 g++ -std=c++20 -O0 -g -ggdb -fcoroutines 并把所有的 std::experimental:: 都换成 std:: 之后编译运行。
LLVM+Clang+libc++/libc++abi的编译安装脚本可以参见:
C++20 的协程基本原理
C++20 整个协程体系是 "无栈协程" 的思路，整个功能是需要结合编译器功能和STL来配合实现的。主要就是三个关键字（co_yield 、 co_await 或 co_return）和围绕这三个关键字的接入。无栈协程对API的设计是有要求的，C++20 Coroutine也不例外， 编译器在检测到内部有使用 这三个关键字时会对函数的流程做patch，然后它的返回值类型必须符合你所使用的关键字的规范。这三个关键字的规范要求不太一样，下面会列举。
我原本以为的会放在协程的 awaiter 或者 handle 对象闭包里，然后由编译器分析和对闭包内的各级对象进行扩充的引用（类似Rust的那种实现）。但是在测试的MSVC和Clang的协程流程的过程中发现，实际上还是另外堆上分配空间来保存协程函数的栈上数据，并用这种方式实现Zero-Copy的。协程函数的执行栈和主函数执行栈并不在一个地址段内，这和之前猜想的不太一样。所以，C++20 的协程也不能完全说是 "无栈" ，只是在协程函数中需要能够评估出来它需要多少栈空间存数据，不像有栈协程那样会浪费比较大的地址空间且不利于内存页复用。
同时受限于这种设计，在C++20 的协程函数里，动态栈分配是不受支持的。在MSVC下，如果你使用了动态栈分配的函数 ( _alloca ) ，直接编译就不通过了。而在gcc/clang 下如果你使用了动态栈分配的函数 ( alloca ) ，分配出来的栈地址是不会受到协程的管理（即：多个协程分配出来的地址可能是重合的），在使用的时候用户得自己保证如果涉及协程且如何切出的话，运行结果不受这部分动态长度栈数据的影响，因为可能会被其他协程改掉（简单地说就是动态分配出来的栈只能在 co_yield 和 co_await 之前使用）。
不过我觉得类似GCC动态栈的那种方案可以让它支持动态栈空间，就是在栈溢出的signal里再mmap一段地址进去，按需增大栈空间。但是这玩意性能被诟病，信号和缺页中断都不稳定且栈空间地址分散不利于CPU Cache，估计最后也不会被采纳吧。
我目前看的提案以 为准（还有个使用文档是 ）。一旦一个函数被断定为协程函数，那么它会被扩充为如下形式:
2019-09-29: 更新文档 这个 比 完整得多 : 后面会整合进 : 目前还没有太大的变化。
COROUTINE_OBJECT func(args...) {
try {
P p(promise_constructor_arguments);
// 这个P是自己定义的 using P = COROUTINE_OBJECT::promise_type;
// 文档上说promise_constructor_arguments 是空或者函数的参数的左值传入 args... ，但是目前版本的MSVC还仅支持空参数列表
COROUTINE_OBJECT r = p.get_return_object(); // MSVC 1, Clang 2
co_await p.initial_suspend(); // 初始化接口 MSVC 2, Clang 1
// 上面这两行是MSVC的顺序，在clang里上面两行的顺序相反
try {
// 原函数体 ...
p.return_void() or p.return_value(RET) // 取决于函数体里有没有 co_return RET
} catch(...) {
p.unhandled_exception(); // 未捕获的异常接口
}
final_suspend:
co_await p.final_suspend(); // final suspend point
return r;
} catch(...) {
return COROUTINE_OBJECT::promise_type::get_return_object_on_allocation_failure(); // noexcept
}
}
关键字 co_await 先简单理解为判定是否需要切出，后面会有更加详细一点的解释。 而协程函数一切的核心都在于上面的 COROUTINE_OBJECT 类型。里面有一些规范， co_yield 和 co_return 涉及 COROUTINE_OBJECT 里的 COROUTINE_OBJECT::promise_type 类型。必须实现某些公共接口和函数功能接口。而返回到外层的 COROUTINE_OBJECT 对象里也需要保存协程的handle(目前是 std::experimental::coroutine_handle<PROMISE_TYPE> )，以用于后续控制协程的上下文切换使用。 同样， 下面的例子因为必须写一个协程入口对象，所以有一部分比较烦杂的必须接入的代码。
结构如下:
协程闭包类型(promise)
要支持协程函数，首先要准备一个包装的类型，里面包含 promise_type ，然后提供基本的创建、维护handle的函数。比如:
struct coroutine_task {
struct promise_type {
coroutine_task get_return_object() {
return coroutine_task{};
}
bool initial_suspend() const {
return false;
}
bool final_suspend() const {
return false;
}
};
};
然后，就可以声明使用协程函数 coroutine_task f(); 了。但是要让他转变为协程函数，还需要至少接入一样协程关键字才行。我们先从最基本的 co_await 开始。
关键字 - co_await
关键字 co_await 是一个操作符，所以我们只要实现这个操作符重载就可以实现协程等待任意类型。 当然，返回的类型要求实现 bool await_ready(T&) noexcept 、 void await_suspend(T&, std::experimental::coroutine_handle<T>) noexcept 、 void await_resume(T&) noexcept 这三个接口。比如我们实现一个 co_wait 后结束的结构，可以按下面这种接入方式:
struct wait_some_times {
int left_times;
std::experimental::coroutine_handle<> handle;
wait_some_times(int t) : left_times(t), handle(nullptr) {}
};
struct suspend_some_times {
wait_some_times& d;
suspend_some_times(wait_some_times& _d): d(_d) {}
bool await_ready() noexcept {
std::cout << "call await_ready: " << d.left_times << std::endl;
return d.left_times <= 0;
}
void await_suspend(std::experimental::coroutine_handle <> h) noexcept {
// 记下来handle以便后面resume用
d.handle = h;
std::cout << "call await_suspend: " << (--d.left_times) << std::endl;
}
void await_resume() noexcept {
std::cout << "call await_resume: " << d.left_times << std::endl;
d.handle = nullptr;
}
};
然后把整个连起来，完整的例子如下:
#include <iostream>
#include <iomanip>
#include <vector>
#include <memory>
#include <experimental/coroutine>
struct wait_some_times {
int left_times;
std::experimental::coroutine_handle<> handle;
wait_some_times(int t) : left_times(t), handle(nullptr) {}
};
struct suspend_some_times {
wait_some_times& d;
suspend_some_times(wait_some_times& _d) : d(_d) {}
bool await_ready() noexcept {
std::cout << "call await_ready: " << d.left_times << std::endl;
return d.left_times <= 0;
}
void await_suspend(std::experimental::coroutine_handle <> h) noexcept {
// 记下来handle以便后面resume用
d.handle = h;
std::cout << "call await_suspend: " << (--d.left_times) << std::endl;
}
void await_resume() noexcept {
std::cout << "call await_resume: " << d.left_times << std::endl;
d.handle = nullptr;
}
};
struct coroutine_task {
struct promise_type {
coroutine_task get_return_object() {
return coroutine_task{};
}
auto initial_suspend() {
return std::experimental::suspend_never{};
}
auto final_suspend() {
return std::experimental::suspend_never{};
}
void unhandled_exception() {}
void return_void() {}
};
};
auto operator co_await(wait_some_times& x) noexcept {
return suspend_some_times{ x };
}
coroutine_task f(wait_some_times& waiter) {
std::cout << "begin to co_await" << std::endl;
co_await waiter; // 只有前三次会协程切出
co_await waiter;
co_await waiter;
co_await waiter; // 这之后await_ready返回true了，不会再切出
co_await waiter;
std::cout << "end of corotine" << std::endl;
}
int main(int argc, char* argv[]) {
#ifdef __cpp_coroutines
std::cout << "__cpp_coroutines: " << __cpp_coroutines << std::endl;
#endif
wait_some_times waiter{ 3 };
f(waiter);
while (waiter.handle && !waiter.handle.done()) {
std::cout << "about to resume: " << waiter.left_times << std::endl;
// 这里用传出的handle来恢复切入协程
waiter.handle.resume();
}
return 0;
}
输出如下:
__cpp_coroutines: 201703
begin to co_await
call await_ready: 3
call await_suspend: 2
about to resume: 2
call await_resume: 2
call await_ready: 2
call await_suspend: 1
about to resume: 1
call await_resume: 1
call await_ready: 1
call await_suspend: 0
about to resume: 0
call await_resume: 0
call await_ready: 0
call await_resume: 0
call await_ready: 0
call await_resume: 0
end of corotine
关键字 - co_yield
关键字 co_yield 要求实现 COROUTINE_OBJECT::promise_type::yield_value(参数) 。比较贴近于单独一次异步调用的实现。
简单的描述 co_yield VALUE 就是相当于 co_await p.yield_value(VALUE) 。
#include <iostream>
#include <iomanip>
#include <vector>
#include <memory>
#include <experimental/coroutine>
struct test_rpc_generator {
test_rpc_generator(const test_rpc_generator&) = delete;
test_rpc_generator(test_rpc_generator&& other): coro(other.coro) {
other.coro = nullptr;
};
~test_rpc_generator() {
if (coro) {
coro.destroy();
}
}
struct promise_type;
using handle = std::experimental::coroutine_handle<promise_type>;
struct promise_type {
int* current_value;
static auto get_return_object_on_allocation_failure() {
return test_rpc_generator{nullptr};
}
auto get_return_object() {
return test_rpc_generator{handle::from_promise(*this)};
}
auto initial_suspend() {
current_value = nullptr;
return std::experimental::suspend_never{};
}
auto final_suspend() {
return std::experimental::suspend_always{};
}
void unhandled_exception() {
std::terminate();
}
void return_void() {
}
auto yield_value(int* value) {
current_value = value;
return std::experimental::suspend_always{};
}
};
int* value() const {
if (coro) {
return coro.promise().current_value;
}
return 0;
}
bool move_next(int rpc_result) {
if (coro && coro.promise().current_value) {
*coro.promise().current_value = rpc_result;
}
return coro ? (coro.resume(), !coro.done()) : false;
}
bool await_ready() const {
return !coro || coro.done();
}
private:
test_rpc_generator(handle h) : coro(h) {}
handle coro;
};
test_rpc_generator f() {
int rpc_res1, rpc_res2;
co_yield &rpc_res1;
// _alloca(rpc_res1);
std::cout<< "resumed got rpc_res1: "<< rpc_res1<< "(@"<< &rpc_res1<< ")" << std::endl;
co_yield &rpc_res2;
// _alloca(rpc_res2);
std::cout<< "resumed got rpc_res1: "<< rpc_res1<< "(@"<< &rpc_res1<< ")" << ", rpc_res2: "<< rpc_res2 <<"(@"<< &rpc_res2<< ")"<< std::endl;
}
int main(int argc, char * argv[]) {
#ifdef __cpp_coroutines
std::cout<< "__cpp_coroutines: "<< __cpp_coroutines<< std::endl;
#endif
int rpc_fake_data = 1;
auto g1 = f();
auto g2 = f();
void* detect_addr = malloc(4000);
std::cout << "detect_addr:" << detect_addr << std::endl;
free(detect_addr);
for (bool is_continue = true; is_continue; is_continue = (!g1.await_ready() || !g2.await_ready())) {
if (!g1.await_ready()) {
g1.move_next(++ rpc_fake_data);
std::cout << "g1 value:" << g1.value() << std::endl;
}
if (!g2.await_ready()) {
g2.move_next(++ rpc_fake_data);
std::cout << "g2 value:" << g2.value() << std::endl;
}
}
return 0;
}
我这里一次示例输出是:
__cpp_coroutines: 201703
detect_addr:00881BD0
resumed got rpc_res1: 2(@0087F96C)
g1 value:0087F980
resumed got rpc_res1: 3(@00881B1C)
g2 value:00881B30
resumed got rpc_res1: 2(@0087F96C), rpc_res2: 4(@0087F980)
g1 value:0087F980
resumed got rpc_res1: 3(@00881B1C), rpc_res2: 5(@00881B30)
g2 value:00881B30
关键字 - co_return
最后一个是 co_return 。 这个关键字主要是用于直接退出协程函数的, 因为协程函数的返回值是我们自己定义的这个 COROUTINE_OBJECT ， 所以函数逻辑附带的返回值就要用这个关键字来实现。 这个关键字要求实现 COROUTINE_OBJECT::promise_type::return_value(参数) 或者 COROUTINE_OBJECT::promise_type::return_void() 。如果协程函数中有 co_return VALUE 。则是最终调用了 COROUTINE_OBJECT::promise_type::return_value(VALUE) ， 否则是调用 COROUTINE_OBJECT::promise_type::return_void() 。
我们可以把协程函数的最终结果放在这里面来实现转储到某个地方。这部分的代码和 yield 的很像。就不另外贴了，下面贴一个全部整合到一起的吧。
全功能整合到一起
我们来一个功能完整，并且贴近单线程工程并且时候异步IO的实践的例子。
#include <iostream>
#include <iomanip>
#include <vector>
#include <memory>
#include <experimental/coroutine>
static std::vector<std::pair<int*, std::experimental::coroutine_handle<> > > g_test_rpc_manager;
static int g_test_rpc_fake_data = 0;
struct test_rpc_generator {
struct test_rpc_data {
int final_value;
int yield_times;
std::vector<std::experimental::coroutine_handle<> > follower;
};
struct promise_type;
using data_ptr = std::shared_ptr<test_rpc_data>;
struct promise_type {
data_ptr data;
static auto get_return_object_on_allocation_failure() {
return test_rpc_generator{ nullptr };
}
auto get_return_object() {
data = std::make_shared<test_rpc_data>();
if (data) {
data->final_value = 0;
data->yield_times = 0;
}
return test_rpc_generator{ data };
}
auto initial_suspend() {
return std::experimental::suspend_never{}; // STL提供了一些自带的awaiter实现，我们其实很多情况下也不需要另外写，直接用STL就好了
}
auto final_suspend() {
return std::experimental::suspend_always{}; // 和上面一样，也是STL自带的awaiter实现
}
void unhandled_exception() {
std::terminate();
}
// 用以支持 co_return
void return_value(int v) {
// 最终co_return时保存最终数据
if (data) {
data->final_value = v;
auto followers = std::move(data->follower);
for (auto& h : followers) {
h.resume();
}
}
}
// 用以支持 co_yield
auto yield_value(int* value) {
// 每次调用都会执行,创建handle用以后面恢复数据
g_test_rpc_manager.emplace_back(std::make_pair(value, std::experimental::coroutine_handle<>::from_address(
std::experimental::coroutine_handle<promise_type>::from_promise(*this).address()
)));
if (data) {
++data->yield_times;
}
return std::experimental::suspend_always{};
}
};
// 下面的接入用侵入式的方式支持 co_await test_rpc_generator
// MSVC 目前支持使用非侵入式的方式实现，但是clang不支持
bool await_ready() noexcept {
return value() > 0;
}
void await_resume() {
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": test_rpc_generator resume for " << yield_times() << " time(s)" << std::endl;
}
void await_suspend(std::experimental::coroutine_handle<> h) {
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": test_rpc_generator yield for " << yield_times() << " time(s), wait for " << h.address() << std::endl;
// 记录要恢复父协程
if (h) {
add_follower(h);
}
}
int value() const {
if (data) {
return data->final_value;
}
return 0;
}
int yield_times() const {
if (data) {
return data->yield_times;
}
return -1;
}
void add_follower(std::experimental::coroutine_handle<> h) {
if (data) {
data->follower.emplace_back(std::move(h));
}
}
private:
test_rpc_generator(data_ptr d) : data(d) {}
data_ptr data;
};
// 异步协程函数
test_rpc_generator f() {
int rpc_res1, rpc_res2;
co_yield &rpc_res1;
// _alloca(rpc_res1);
std::cout << "resumed got rpc_res1: " << rpc_res1 << "(@" << &rpc_res1 << ")" << std::endl;
co_yield &rpc_res2;
// _alloca(rpc_res2);
std::cout << "resumed got rpc_res1: " << rpc_res1 << "(@" << &rpc_res1 << ")" << ", rpc_res2: " << rpc_res2 << "(@" << &rpc_res2 << ")" << std::endl;
// 模拟多次RPC然后返回最终结果
co_return rpc_res1 * 100 + rpc_res2;
}
// 这里模拟生成数据
void test_rpc_manager_run() {
std::vector<std::pair<int*, std::experimental::coroutine_handle<> > > rpc_manager;
g_test_rpc_manager.swap(rpc_manager);
for (auto& generator : rpc_manager) {
if (generator.first) {
*generator.first = ++g_test_rpc_fake_data;
}
if (generator.second && !generator.second.done()) {
generator.second.resume();
}
}
}
struct test_task {
using ptr_t = std::shared_ptr<test_task>;
test_task(int ms) : status(0), max_status(ms) {}
bool is_ready() const noexcept {
return status >= max_status;
}
int status;
int max_status;
};
struct test_task_future {
struct promise_type;
using ptr_t = std::shared_ptr<test_task>;
using handle = std::experimental::coroutine_handle<promise_type>;
struct promise_type {
static auto get_return_object_on_allocation_failure() {
return test_task_future{ nullptr };
}
auto get_return_object() {
return test_task_future{ handle::from_promise(*this) };
}
auto initial_suspend() {
return std::experimental::suspend_never{};
}
auto final_suspend() {
return std::experimental::suspend_always{};
}
void unhandled_exception() {
std::terminate();
}
void return_void() {
for (auto& h : follower) {
h.resume();
}
}
// 这里是为了把task::ptr_t数据关联进去，目前版本不支持promise构造函数参数，没想到什么其他好方法
auto yield_value(ptr_t t) {
task = t;
return std::experimental::suspend_never{};
}
ptr_t task;
std::vector<handle> follower;
};
// 下面的接入用侵入式的方式支持 co_await test_task::ptr_t
struct awaitable {
awaitable(const test_task_future& pt) {
if (pt.coro) {
data = pt.coro.promise().task;
coro = pt.coro;
}
};
bool await_ready() const {
bool ret = !data || data->is_ready();
if (ret) {
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": task " << data.get() << " ready" << std::endl;
}
return ret;
}
void await_resume() {
if (data) {
++data->status;
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": task " << data.get() << " resume to " << (data ? data->status : -1) << std::endl;
}
}
void await_suspend(handle h) {
if (data) {
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": task " << data.get() << " suspend test_task_future::handle from " << (data ? data->status : -1) <<
", wait for " << h.address() << std::endl;
}
if (coro && h) {
coro.promise().follower.emplace_back(h);
}
}
ptr_t data;
handle coro;
};
bool done() const {
return !coro || coro.done();
}
private:
test_task_future(handle h) : coro(h) {}
handle coro;
};
// 接入 co_await test_task::ptr_t
auto operator co_await(const test_task_future & pt) noexcept {
return test_task_future::awaitable{ pt };
}
test_task_future h(test_task::ptr_t task) {
// 这里是为了把task::ptr_t数据关联进去，目前版本不支持promise构造函数参数，没想到什么其他好方法
co_yield task;
// 模拟任务内部流程并调用外部RPC
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": task " << task.get() << std::endl;
test_rpc_generator rpc_res = f();
co_await rpc_res;
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": task " << task.get() << " call f() ret: " << rpc_res.value() << std::endl;
}
test_task_future g(test_task::ptr_t task) {
// 这里是为了把task::ptr_t数据关联进去，目前版本不支持promise构造函数参数，没想到什么其他好方法
co_yield task;
// 等待子任务完成
while (task && !task->is_ready()) {
std::cout << std::setw(32) << __FUNCTION__ << std::setw(3) << ": " << __LINE__ << ": task " << task.get() << std::endl;
co_await h(task);
}
}
int main(int argc, char* argv[]) {
#ifdef __cpp_coroutines
std::cout << "__cpp_coroutines: " << __cpp_coroutines << std::endl;
#endif
// 创建一个任务
test_task::ptr_t task = std::make_shared<test_task>(3);
// 运行任务
auto fut = g(task);
// 模拟从外部获取数据然会恢复协程
while (!fut.done()) {
test_rpc_manager_run();
}
return 0;
}
这回贴一下linux内clang的输出吧:
__cpp_coroutines: 201703
g : 280: task 0x1d3c028
h : 268: task 0x1d3c028
await_suspend : 89: test_rpc_generator yield for 1 time(s), wait for 0x1d3c320
await_suspend : 235: task 0x1d3c028 suspend test_task_future::handle from 0, wait for 0x1d3c040
resumed got rpc_res1: 1(@0x1d3c730)
resumed got rpc_res1: 1(@0x1d3c730), rpc_res2: 2(@0x1d3c734)
await_resume : 85: test_rpc_generator resume for 2 time(s)
h : 271: task 0x1d3c028 call f() ret: 102
await_resume : 229: task 0x1d3c028 resume to 1
g : 280: task 0x1d3c028
h : 268: task 0x1d3c028
await_suspend : 89: test_rpc_generator yield for 1 time(s), wait for 0x1d3c850
await_suspend : 235: task 0x1d3c028 suspend test_task_future::handle from 1, wait for 0x1d3c040
resumed got rpc_res1: 3(@0x1d3cc60)
resumed got rpc_res1: 3(@0x1d3cc60), rpc_res2: 4(@0x1d3cc64)
await_resume : 85: test_rpc_generator resume for 2 time(s)
h : 271: task 0x1d3c028 call f() ret: 304
await_resume : 229: task 0x1d3c028 resume to 2
g : 280: task 0x1d3c028
h : 268: task 0x1d3c028
await_suspend : 89: test_rpc_generator yield for 1 time(s), wait for 0x1d3cd40
await_suspend : 235: task 0x1d3c028 suspend test_task_future::handle from 2, wait for 0x1d3c040
resumed got rpc_res1: 5(@0x1d3d150)
resumed got rpc_res1: 5(@0x1d3d150), rpc_res2: 6(@0x1d3d154)
await_resume : 85: test_rpc_generator resume for 2 time(s)
h : 271: task 0x1d3c028 call f() ret: 506
await_resume : 229: task 0x1d3c028 resume to 3
promise/future 支持
我本地测试的Clang版本（7.0.1）尚未实现支持。 MSVC的标准库用偏特化的方式对 std::promise 和 std::future 实现了协程的接入。它的接入代码如下（为了简短，精简掉了一个偏特化实现，流程和不特化的类似）:
namespace experimental {
template <class _Ty, class... _ArgTypes>
struct coroutine_traits<future<_Ty>, _ArgTypes...> { // defines resumable traits for functions returning future<_Ty>
struct promise_type {
promise<_Ty> _MyPromise;
future<_Ty> get_return_object() {
return _MyPromise.get_future();
}
bool initial_suspend() const {
return false;
}
bool final_suspend() const {
return false;
}
template <class _Ut>
void return_value(_Ut&& _Value) {
_MyPromise.set_value(_STD forward<_Ut>(_Value));
}
void set_exception(exception_ptr _Exc) {
_MyPromise.set_exception(_STD move(_Exc));
}
};
};
} // namespace experimental
template <class _Ty>
bool await_ready(future<_Ty>& _Fut) {
return _Fut._Is_ready();
}
template <class _Ty>
void await_suspend(future<_Ty>& _Fut,
experimental::coroutine_handle<> _ResumeCb) { // change to .then when future gets .then
thread _WaitingThread([&_Fut, _ResumeCb] {
_Fut.wait();
_ResumeCb();
});
_WaitingThread.detach();
}
template <class _Ty>
auto await_resume(future<_Ty>& _Fut) {
return _Fut.get();
}
可以看到，它的 co_await std::future<T> 挂起是开了个新线程来等待，真他喵暴力，建议不要用。但是还是可以比较容易地让自己的管理器接入 co_await 。
我们借助STL的协程接入， 可以实现一个最小化的自定义协程支持：
#include <iostream>
#include <future>
#include <thread>
#include <chrono>
#include <experimental/coroutine>
struct custom_rpc_generator {};
struct suspend_custom_rpc : public std::experimental::suspend_always {
void await_suspend(std::experimental::coroutine_handle<> h) noexcept {
std::thread thd{ [h]() {
using namespace std;
std::this_thread::sleep_for(2s);
std::cout << "start to resume coroutine" << std::endl;
h.resume();
} };
thd.detach();
}
};
auto operator co_await(const custom_rpc_generator&) {
return suspend_custom_rpc{};
}
std::future<void> outter_fn() {
co_await custom_rpc_generator{};
}
int main() {
auto fut = outter_fn();
std::cout << "start to wait future" << std::endl;
fut.wait();
std::cout << "future finished, ready to exit" << std::endl;
return 0;
}
总结
总体感觉上，C++20协程为了兼顾灵活和支持非侵入式接入，设计了好几个互相交织的大模块，函数级有处理协程函数内部的 promise_type 、 协程函数对外暴露的交互对象 （我们这里统称为 future） 、 用于协程上下文切换的 handle ，单次切换有用于支持await功能的 awaitable 和一些回调函数接口。几个对象之间的数据共享也并不是很方便。而且和传统有栈协程的区别仅仅是约束了返回值类型，并且可以依次在编译期推断出需要多少栈空间，从而减少浪费。
首先是如果业务有自己的线程池，其实还是要由管理层来控制resume，不能直接像STL那样开线程，那基本上就和 std::future<T> say Good-Bye 了。虽然在小心维护的情况下，避免 co_await std::future<T> 也是可以避免STL乱开线程的，但是我觉得一旦使用了，后面就很难控制住。特别是这个C++协程的对象关系互耦合插如此严重的情况下，本身就不容易理解。
第二个问题是调用链。C++协程接口设计是非对称的，我们实际业务中，肯定还是需要对称协程的支持，即子协程结束后能够自动恢复 co_await 它的父协程。这个需要我们自己去实现，上面 std::future<T> 也就是这里是开了个线程去实现的（都开线程了还用协程干啥）。在上面的sample代码中，我是开了一个共享数据区，在 await_suspend 的时候去追加等待链，在 return_void/return_value 的时候去执行父级协程的恢复切入，这里面写成vector是为了支持N个协程await一个协程的情况，实际上用链表会更好一些。这里有个比较“脏”的设计是handle的加入是在 awaitable 执行挂起的时候，而恢复是在 promise_type 的 return_void/return_value 事件里。这就分了两个地方。这里的 awaitable 对 promise_type 和N个协程等一个协程是保持一致的N:1关系。但是 awaitable 仅仅能访问 future ， 要让 promise_type 也能访问 future 的话，一种方法是开一个共享数据块，在 promise_type 创建 future 的时候传进去这样了。这也是上面sample代码里 using data_ptr = std::shared_ptr<test_rpc_data>; 的实现。 MSVC的STL的实现也是这样 promise::get_future() 是使用了promise内部的数据创建的future对象。 另外一种尝试的方法是后面task的，进入协程后先 co_yield 一次，然后 yield_value(VALUE) 函数返回 std::experimental::suspend_never。这样不会造成协程挂起，并且可以给 promise_type 注入任意数据。不过这样就有个约定式的规范了，也不是很严谨。这方面等后面支持了 promise_type 的带参数构造函数可能可以好一些。
第三个问题是handle提前结束的问题。在看了MSVC的实现，这个handle是可以copy也可以转换的。copy开销也很小，里面只包含一个和 promise_type 地址有关的指针（就是 promise_type 的地址然后加了 align和padding）。比如一个RPC任务，我可能copy一个handle用来在有数据的时候resume，然后我还会copy一个handle在超时的时候强行resume然后走失败流程。现在这种情况，就是需要在 awaitable 的 await_suspend 里添加这两个handle到两个manager里，然后在 await_resume 里要把这两个都移除。不管哪个流程，肯定有一个handle是处于resume回调中的，这还涉及递归调用的问题（删除正在resume过程中的handle）。 我目前的想法是，封装一个handle resumer，让多个事件manager都指向同一个 handle resumer的weak_ptr。这个resumer的生命周期可以和 awaitable 共存，一旦 awaitable 消失，这次的await也就结束了。
