C++20 coroutine 探索III: 异步编程，Task＜T＞ 编写，boost asio 协程分析，C# async / await, cppcoro 源码分析

经过了前面对 coroutine 的反复学习，现在尝试写一些封装好的协程工具（reinventing the wheel）。本文先从最Promise 异步编程模型的最基本的 Task<T> 入手.

异步编程

Futures and promises - Wikipedia

异步的关键点是真正的无线程异步必须要 all the way down 到 O/S system call 层面甚至到硬件层面的异步支持才能实现 (对于 Linux 的驱动模型, 做 top half 工作的 softirq 的本质其实也就是硬件线程). 不过到底来说仍然是死循环, 因为硬件中断就是用一些电一直在检测某个点的电平从而让程序计数器不断执行指令的死循环停下来跳转到特定地方而已.

但是问题是程序还是不知道他委托给 O/S 的事件什么时候会完成.

一开始还是朴素的想法, 既然有了异步系统调用, 那就做 void 返回值的回调就行了, 完成的结果把他放到参数里. 这就是事件驱动 + 异步 I/O 的写法, 事件来了就触发回调函数就行了. 需要各项任务依次同步执行, 那就让一个异步调用运行完的回调里面又进行一个异步调用, 然后继续回调就行了. 当然回调不一定会在这个线程运行了, 但是由于这个异步完成之后的工作基本是不阻塞的 (回调函数里面是不能有阻塞行为的, 否则就会卡死). 这个做法的问题是回调地狱 (比如 js 里面一开始的异步调用都是写成回调地狱的).

然而根据 O/S 的内核隔离政策, 不可能让内核去运行一个应用态的回调函数. 这里的中间层肯定是要应用层上做的. 所以才需要询问的接口 (比如 IOCP的GetQueuedCompletionStatus ) 和 Asynchronous Completion Token 来标识你提交的任务. 这个时候自己如果要提交多个任务, 又要自己做一套 demux 了.

Future 和 Promise 则是直接在这个基础上构建出来的 (future, promise, delay, deferred 其实是同一个东西), 当程序向 O/S 委托一个任务的时候下层组件会构建一个 Promise, 自己保留一个 Future 作为钩子 (C++ 里 Future 的 Promise 的区别就是 Future 是一个 read-only 视图, 视图的概念参照数据库吧). 这个 Promise 将会作为一个 Completion Token, O/S 完成异步事件后会根据这个 token 找到你的 Future 并更新相应的状态 (然而这个 demux 其实是 runtime library 做的)…

比如等待一个硬件中断, 直接中断到达的时候马上就能以增量的方式根据 Promise 更新 Future 的状态, 从而让应用层能够决定拿这个 Future 做事情, 比如询问是否完成等. 如果有需要, 程序也能够完全阻塞在 Future 身上 (实际会 context switch 出去) 等待这个中断来了的时候再唤醒程序线程. 这个过程中没有轮询的空转 CPU .

有了 Future 和 Promise 的抽象之后, 我们不一定要依赖事件/信号机制 (类似硬件中断) 做异步编程, 我们可以自己做两个线程, 并且利用 Future 和 Promise 的关系来进行线程间的同步. 线程 A 只需要异步提交任务给线程 B ,自己则做别的事情, 等待的确要用 Future 的时候, 再获取值或者阻塞或者捕获错误.

但是实际情况是很多程序的主要工作流程都是流水线的, 必须需要各项任务依次同步执行, 然后某些特殊的控制点上不需要等待所有流程运行完成 (比如他只需要发起任务就行了). 一种朴素的想法是在这个控制点上开一个线程去同步完成全部的任务. 但是大部分任务比如 IO , 陷入到 O/S 的时候就会被挂起了, 这种线程一多系统就要维护很多 PCB.

然后是 javascript 的 ES6 提出的 Promise, 这个 Promise (比上面根据 C++ 讲的多了一个东西) 用来解决回调地狱, 我们知道 Future/ Promise 这个东西为了能询问提交的任务是否完成的一个钩子. 因为每个异步任务提交之后会返回一个那就可以在钩子上挂回调函数了.

但是这样写实际还是很难受的, 同步地写程序才是人一直想做的事情.

就想到了进行多线程的单线程复用, 其实就是用户态调度各个子程序. 这就是协程啊! 因为我们掌握了每个协程的所有的信息, 所以我们可以从容地手动实现调度 (虽然的确需要保留协程信息, 但是比陷入 kernel 开销小多了) , 也不再用一堆同步原语.

最简单的生产者消费者模型里, 说我们只有单核,如果没有东西可以消费了, 消费者可以直接 yield 出去 (这个 yield 和 co_yield 不是同一个东西) 把控制流转到生产者去继续生产, 如果生产者需要 sleep 就直接让 thread 都睡在他等待的条件上, 而不是让消费者睡觉 notify 生产者, 然后陷入内核, 然后内核再某个时间点调度到生产者, 生产者发现需要等待, 然后生产者又睡觉…

最基本的协程想法是这样的，假设我们在库里面实现了一个 yield_to 的函数负责 context switch：

def Consumer():
  while True:
    if notEmpty():
      consume()
    else: yield_to(Producer)

def Producer():
  while True:
    if size() > threshold:
      yield_to(Consumer)
    else: produce()

但是这样实际打乱了程序的逻辑, 实在是太原始了. 有必要在应用层完全做一套 Monitor 来做这些东西. 于是抽象出 awaitable 和 awaiter 的概念, 如果发起了一个异步调用, 就等到他完成. 而我实际是挂起了去运行这个异步调用的. 这个挂起切换上下文一路走到最后一个地方就会真的是一个异步操作等操作系统完成之后发起一个 Completion 之后再一步一步运行回来的.

def Consumer():
  while True:
    await get_product_async()
    consume()

def get_product_async():
  if(queue is not empty): return
  save current continuation
  switch to Producer

def Producer():
  while True:
    await empty_queue_async()
    produce_many()

def  empty_queue_async():
  if(queue is empty): return
  save current continuation
  switch to Consumer

所以 await 或者说协程到底要做些什么？对于 I/O 事件无非就是该调用异步动作等 completion 的时候就挂起协程让整个线程都睡觉就行了.

对于存在一个非 await 调用的异步任务调用链条, 可以在注册了事件之后不睡觉, 直接返回调用链条第一个非 await 调用者（具体让一个 Future 也作为 awaitable 的实现下文再讲）。等到 Completion 事件来到的时候恢复运行 (对于 CLR 来说, 这个恢复的线程可能已经是另一个线程了).

这里会有一个问题是非 await 调用异步任务的 Completion 到达的时候 continuation 会在哪里继续运行 ? 由于主线程已经在运行别的东西了, 这种情况被称为 Unsafe On Complete, 这种情况就让 I/O 完成线程来做了（区别原来的主线程可能也整个迁移到其他线程）, 而不是恢复主线程了.

比如 UI 的情况, UI 线程主要做的事情就是在等事件到达, 高响应的话可以用轮询, 不过一般不用因为 HID (鼠标键盘触摸屏等) 延迟还是挺大的. 对于 button click 事件的触发, 就是响应了一个 HID 的事件, 这个时候 UI thread 会直接调用 button_click 在本线程运行.

---

复习 C++ 20 协程组件

下面就来尝试写一个这样的 Task<T>。首先复习一下之前的 C++ 20 协程的全部要素吧。

首先是一个协程必须有一个 R 类型，然后具备 R::promise_type的条件，运行协程的时候，栈和寄存器等都会存到这个 R::promise_type 里面。

然后是能够被 co_xx 的表达式必须具备 Awaitable 的性质, 能够从他获取一个 Awaiter 无论是通过 promise_type 的 await_transform 转换还是隐性地类型转换，而 Aawaiter 即拥有 3 个以 await_ 开头的成员函数，分别是 ready, suspend, resume。

最后一个关键点就是一个协程的 promise_type 对象能够和协程对应 suspend 时候的 coroutine_handle<promise_type> 相互转换。

一个协程运行时，实际会产生这样的伪代码流程：

{
  co_await promise.initial_suspend();
  try
  {
    <body-statements>
  }
  catch (...)
  {
    promise.unhandled_exception();
  }
FinalSuspend:
  co_await promise.final_suspend();
}

而每一次 co_await expr 调用，会引发这样的伪代码流程（补充了一下前面笔记里漏掉的非 TS 的新部分, 根据 理解 co_await – Lewis Baker 的伪代码和 cppreference: coroutine 补充了 await_suspend 返回一个 coroutine_handle 的部分：

{
  auto&& value = <expr>;
  auto&& awaitable = get_awaitable(promise, static_cast<decltype(value)>(value));
  auto&& awaiter = get_awaiter(static_cast<decltype(awaitable)>(awaitable));
  if (!awaiter.await_ready()){
    using handle_t = std::coroutine_handle<P>;
    using await_suspend_result_t =
      decltype(awaiter.await_suspend(handle_t::from_promise(p)));
    <suspend-coroutine> -> save to a heap promise structure
    if constexpr (std::is_void_v<await_suspend_result_t>){
      awaiter.await_suspend(handle_t::from_promise(p));
      <return-to-caller-or-resumer> (get_return_object)
    }
    else if(std::is_same_v<await_suspend_result_t, bool>){
      if (awaiter.await_suspend(handle_t::from_promise(p)))
        <return-to-caller-or-resumer> (get_return_object)
    }else if(std::is_same_v<await_suspend_result_t, std::coroutine_handle>){
      auto h = awaiter.await_suspend(handle_t::from_promise(p));
      h.resume();
      <return-to-caller-or-resumer> (get_return_object)
    }else {throw "Wrong return type";}
    <resume-point>
  }
  return awaiter.await_resume();
}

其中 awaitable 和 awaiter 两个东西会根据这样的伪代码来转换：

template<typename P, typename T>
decltype(auto) get_awaitable(P& promise, T&& expr)
{
  if constexpr (has_any_await_transform_member_v<P>)
    return promise.await_transform(static_cast<T&&>(expr));
  else
    return static_cast<T&&>(expr);
}

template<typename Awaitable>
decltype(auto) get_awaiter(Awaitable&& awaitable)
{
  if constexpr (has_member_operator_co_await_v<Awaitable>)
    return static_cast<Awaitable&&>(awaitable).operator co_await();
  else if constexpr (has_non_member_operator_co_await_v<Awaitable&&>)
    return operator co_await(static_cast<Awaitable&&>(awaitable));
  else
    return static_cast<Awaitable&&>(awaitable);
}

---

C# 中的Task<T>

首先是异步编程里经典的 Task<T> 对象（Async in depth | Microsoft Docs）, 他是一种对 Future and Promise 异步模型的抽象。

在 C# 里面，Task<T> 主要有两种，一种是带 async 前缀的，一种是没有 async 前缀的， async的意义是说明这个 Task 里面会调用 await. 使用方法是对于 I/O bound 就直接 await 运行就行了, 他会直接挂起整个 await 调用链条. 对于 CPU bound 的应该开一个 Task.Run 进入到 CLR 的 thread pool 里面调度后台运行.

通过 await 的 async 方法就不会 block UI 的实现，下面的例子的 C# 代码 (例子来自StackOverflow)：

async private void button1_Click(object sender, EventArgs e)
{
  int contentlength = await AccessTheWebAsync();
  tbResult.Text = string.
      Format("Length of the downloaded string: {0}.", contentlength);
  Debug.WriteLine(tbResult.Text);
}    
async Task<int> AccessTheWebAsync()
{
  Debug.WriteLine("Call AccessTheWebAsync");
  await Task.Delay(5000);
  Debug.WriteLine("Call AccessTheWebAsync done");
  tbResult.Text = "I am in AccessTheWebAsync";
  return 1000;
}
public static void FakeMain(string[] args){
    for(;;){
        if(button1.isClicked()){ // 伪代码
            button1_Click(button1, e); // 不会被阻塞, FakeMain 得不到 Task 的钩子(返回 void)
        }
    }
}

带 async 的 Task 将会被认为是一个异步任务 (编译为状态机), 这个异步任务被 await 的时候, 让我们说这一行:

  int contentlength = await AccessTheWebAsync();

这里发生的事情就是由于 UI thread 没有 await 调用 button1_Click (button1_Click 本身就无法被 await, 因为他不是一个 Task ), 所以不会被阻塞, 由于他是一个 async 方法, Completion (timeout 可以和异步 I/O 同样的实现方法实现) 到达的时候他将会在后台(按 MSDN 的说法是分配到CLR thread pool 里面的一个线程) 继续运行.

button1_Click() 将会挂起并把 current continuation 传给 AccessTheWebAsync() , AccessTheWebAsync() 会注册一个 timer 请求后直接返回到 UI 线程.

随后AccessTheWebAsync() 和 button1_Click 的 continuation 链条将会在 Completion 到达后由 I/O Completion 线程继续执行.

上面讲的是 Async Task，普通的 Task 创建了（一个协程返回一个 Task，可以认为这个协程是一个 Task 工厂方法）之后并不会执行，只有当他被 run 的时候才执行。

---

Lazy Computation

cppcoro 这个 Task 的概念其实和上面讲的 async Task 有点不一样，但是在某种意义上的确是实现同样的功能：A task represents an asynchronous computation that is executed lazily in that the execution of the coroutine does not start until the task is awaited.

首先讲一下这里 cppcoro::task<T> 的用法先，为了切合上面的例子， 我模仿上面 C# UI 线程这个例子写的，包含了 awaitable 和 R::promise_type 的所有要素了。至于他和 C# 的 async Task 到底有什么不一样下一节就讲。

对于一个普通 Task 而言，可以创建它而不启动他（lazy 求值）。但是如果启动的话 (co_await)，就要一次运行完他，包括他内部调用的 await 运行的其他 Task， 并且会 all the way down 到第一个 co_await 非Task Awaitable 语句或者纯粹的普通函数（非协程）。为了方便看结果，代码示例里定义了一个同步的协程类型 SyncVoid ，他的意思是直接同步等待结果。（完整的可运行代码在本文最后写完 task 就得到了）。

#include <thread>
#include <iostream>
#include "task.hpp"

struct TaskDelay {
  constexpr bool await_ready() noexcept { return false; }
  void await_suspend(std::coroutine_handle<>) {
    // register the handle to the runtime library
    // register a timer to the o/s
    std::cout<<"        catch a delay request, block the all co_await expr on the way\n";
  }
  void await_resume() {}
  TaskDelay(int) {}
};

cppcoro::task<int> access_the_web_async() {
  std::cout << "      in task, ready to call co_await delay()\n";
  co_await TaskDelay(5000);
  std::cout << "      in task, ready to return 1\n";
  co_return 1;
}

cppcoro::task<> generate_task() {
  std::cout << "    in generate_task, get a task:\n";
  cppcoro::task<int> t = access_the_web_async();
  std::cout << "    in generate_task, got the task.\n";
  std::cout << "    in generate_task, co_await the task.\n";
  int ret = co_await t;
  std::cout << "    in generate_task, co_await return.\n" << ret << std::endl;
}

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

AsyncVoid button_click() {
  std::cout << "  in button click ready to assign a task\n";
  auto rt = generate_task();
  std::cout << "  in button click ready to co_await a task\n";
  co_await rt;
  std::cout << "  in button click, co_await ereturn\n";
}

int main(void) {
  std::cout << "in ui thread\n";
  button_click();
  std::cout << "in ui thread, button is clicked!\n";
  return 0;
}

他的运行结果是这样的：

in ui thread
  in button click ready to assign a task
  in button click ready to co_await a task
    in generate_task, get a task:
    in generate_task, got the task.
    in generate_task, co_await the task.
      in task, ready to call co_await delay()
        catch a delay request, block the all co_await expr on the way
in ui thread, button is clicked!

可以看到，的确 task 在没有 await的时候的确是不会运行的。而一旦 co_await rt 被调用的时候，其内部的全部 task<T> 类型都运行完毕了（指 all the way down 到第一个 co_await 非Task Awaitable 语句或者纯粹的普通函数）。

注意这个 AsyncVoid 类型, 我们必须理解的一点是对于 co_await 的行为， 不是由调用 co_await 的协程来控制的而是由后面的这个 awaiter 的 await_suspend 或者 operator co_await()返回的 awaiter 的 await_suspend 决定的。 所以 co_await rt 的时候 test 协程并没有被挂起（准确的说他被挂起了转而运行 Task rt 之后又被 resume 了 ）在 main 函数里，这一点和之前写 generator 的玩具版时的表现并不一样。

所以这样的 Task 协程有什么用呢?

第一个, 可以用它来实现协程的闭包函数 (当然有点大材小用了).

第二, 正如其名可以用来封装一个 Task 抽象供其他线程运行, 这样能够实现一个任务队列.

在 C# 里面, 常用的主要是用来建立一个后台任务让他在后台运行, 也就是当成一个 std::future 来使用, 这种时候我们可以在另一个线程里面调度运行这个 Task 而运行结束后其他线程再次 await 他的时候就不会再引发挂起或者阻塞 (相当于结果已经被 Cached 了).

当然, 对于在学的是网络编程, 怎么也得讲一个网络编程的应用吧. 既然学了 Proactor 的异步 I/O 模式, 那么当然也要说一下为什么协程和异步 I/O 是好搭档才行吧.

一个想法是, 如果我们知道一个 Task 内部某个地方会有一个 co_await 调用会被长时间的阻塞, 那么我们可以创建它, 而不运行它, 等到我们知道那个 co_await 的调用会马上测试得到 await_ready() == true 的结果, 我们再去 await 这个 Task, 结果就是这个 Task 一被运行就能顺畅的流转完毕, 从而不再需要阻塞, 这样结合任务队列 BlockingQueue 的用法, 我们就做出一个 IOCompletion 事件的 Callback 了.

---

await 和不 await，async 和不 async

先留白。。。这里应该讲 C# 的 async Task 和普通 Task 和 这个 cppcoro::task 的区别的。

---

EventLoop 和 coroutine (asio 实现分析)

例子是创建一个 Task, 并且同步地编写异步程序, 即 Task 内部有一个 co_await 某个 IO 操作, 但是他实际实现是提交一个 Async RecvEx 请求给系统 Async I/O 队列。

// asio demo: echo server with coroutine
// 请把 asio 的 awaitable 当成我们上面说的 Task
awaitable<void> echo(tcp::socket socket) {
  try {
    char data[1024];
    for (;;) {
      std::size_t n = co_await socket.async_read_some(boost::asio::buffer(data),
                                                      use_awaitable);
      co_await async_write(socket, boost::asio::buffer(data, n), use_awaitable);
    }
  } catch (std::exception& e) {
    std::printf("echo Exception: %s\n", e.what());
  }
}

awaitable<void> listener() {
  auto executor = co_await this_coro::executor;
  tcp::acceptor acceptor(executor, {tcp::v4(), 55555});
  for (;;) {
    tcp::socket socket = co_await acceptor.async_accept(use_awaitable);
    co_spawn(executor, echo(std::move(socket)), detached);
  }
}

int main() {
  try {
    boost::asio::io_context io_context(1);

    boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
    signals.async_wait([&](auto, auto) { io_context.stop(); });

    co_spawn(io_context, listener(), detached);

    io_context.run();
  } catch (std::exception& e) {
    std::printf("Exception: %s\n", e.what());
  }
}

这里 co_spawn 做的事情是把一个协程绑定到一个线程上下文上去然后运行他（从而得到一个协程）：Spawn a new coroutined-based thread of execution。可以看到 listener 里面所有的 echo 都会绑定到同一个 executor 里面：The first argument to co_spawn() is an executor that determines the context in which the coroutine is permitted to execute. For example, a server’s per-client object may consist of multiple coroutines; they should all run on the same strand so that no explicit synchronisation is required.

detached 是忽略结果，不需要异步 IO 返回的结果（这里的异步 IO 已经给 asio 封装过了，给 buffer 写的东西将会由 asio 负责全部读写完成）。

实际内部实现可以是这样的（具体的实现其实上一篇已经跟踪过一次了，写的不是一般地扭曲，我没有办法想到他们是怎么写出这么精妙复杂的代码出来的，可能是我态啋了），这里就抽象地复习一下吧：

• 首先 main 里面创建一个单一的单线程 io_context. 此后的所有逻辑都会单线程运行. (由于分析的是 WIN 下面的, 所以也就没有 asio 自己模拟的 aio 线程). 结果是之后的协程都会在同一个线程运行. 所以也完全不会有什么并发的同步原语的应用.

      boost::asio::io_context io_context(1);
  

• 然后启动一个 listener 协程. 这个 co_spawn 运行的时候马上就执行一个异步 IO 启动器：async_initiate。 这个 initiate 做的事情是运行这个 awaitable （awaitable as funtion），具体的机制是这样的，首先限定这个线程在 io_context 上面跑，然后由于 listener 启动的时候马上就捕获一个 promise 就结束了，所以真要启动到 co_await accept 那一行还要等到他运行 pump 把 awaitable_frame 拿出来 resume。于是 listener 被启动了！（boost 这里抽象好几层，差点把我搞晕了）

      co_spawn(io_context, listener(), detached);
  

• 上面说到 listener 刚被 resume 了，于是这个时候陷入到 async_accept 去了。此时再次进入到一个 async_initiate， 这次调用的编程了一个定制的函数 async_move_accept (这个值得参考因为他有 io_uring 的写法)，IOCP 的写法是运行一个 start_accept_op which 最后会运行 AcceptEx！问题是捕获 handler 的问题。

  tcp::socket socket = co_await acceptor.async_accept(use_awaitable);
  

• 上面说到 AcceptEx 已经被调用了，此时应该把 listener 的 coroutine_handle 存下来。这时候涉及到 operator co_await 和 await_transform 的东西。***这里太乱了，他的返回值全部用 auto 和类型推导或者 traits 的，我根本找不到 async_accept 返回了个啥，这怎么知道 await_suspend 做了个啥啊。（也没法找到 await_resume, 因为这个用的是 deduction ）只能合情推理这个东西会返回一个 awaitable 了（新1.74的代码直接会进行一个 await_transform 不过最后都是 push 进 frame）。如果是这样的话那么 co_await 之后，这个 handle 将会被 push_frame 进入到 io_context 里面的一个 coroutine 的一个 frame。此后这个 listen 协程东西就完全被挂起了。并且这个 handle 他以另一个方式封装好了丢到他原来的异步 IO 的 callback 里面去了，所以未来如果 Completion 来了就会恢复这个 coroutine。

  tcp::socket socket = co_await acceptor.async_accept(use_awaitable);
  

• 这时代码来到了 run，虽然上面协程这个的确有点模糊，但是 run 在 linux 下这个我的确分析清楚了（放在一篇讲 epoll 不支持普通文件以及 socket 文件读写最大缓冲限制的笔记的”I/O 最后的拼图“一篇，这篇没有发布博客）。不过这里还是讲 IOCP 的情况。他会循环运行一个 do_one 函数, 然后 do_one 函数会循环调用 GetQueuedCompletionStatus。（这里我又模糊了，这个 Completion 总是会保证 bytesTransfered 是我请求的数量吗？Stack Overflow ，不过实际可以不用 IOCP 端口提交请求，WIN32 提供了 WSA 系列函数直接对 OVERLAPPED 结构提交异步请求，这其中就包括了异步 Recv 函数等。还有一个问题是如果我不知道要读多少怎么办？（ io_uring 同样不明了，这个主要是针对 buffered i/o 来说的，[2/3] io_uring: short reads (kernel.org 根据 io_uring.c 的源码，一开始的实现是先非阻塞尝试一次，然后进行阻塞调用, 如果是这种实现，那么实际的 read write 里 short read 是允许的，而且是和 low-water-mark 相关的了，which 能读 1 byte 就 readable，而写是能写 2048 byte 就 writable）。

  io_context.run();
  

这里这个 echo 的循环实际是对单个客户的长连接服务 (断开连接。我暂时还没看怎么处理). 不过 asio 的写法的确有点混乱, 虽然实现上 Task 的确要做成是一个 Awaitable, 但是 boost 直接把他做成 Awaitable 的概念有点迷惑了, 就跟我追踪代码看到的那样十分精妙的实现, 大神代码, 但是太绕了. (虽然本文下面的内容也会很绕).

---

接下来进入实现 Task 的思考部分。

我们先梳理一下对 Task<T> 协程的需求:

• 首次调用完全不运行, 马上挂起返回一个 Task.
• 对于一个 Task<T>(1) 被一个 Task<T>(2) co_await 调用的时候, 要保证自己(1)运行完之后也运行他的 co_await 调用者(2). (保证一个 Task(2) 第一次调用 co_await 的时候展开他内部所有的(2) co_await.)
• 对于一个已经运行完的 Task<T>, 缓存他的结果, 当他第二次被 co_await 的时候直接返回结果.

思路: 控制流部分

首先，需要让 Task 运行的时候马上返回一个 Task 供后续的 lazy 调用运行。对于这个需求，很容易实现，我们想到直接让 promise_type::initial_suspend 返回一个 suspend_always 就行了.

而 handle 是否捕获都不重要了, 因为我们拿到 Task 就等于拿到 Promise, 有了 Promise 就能得到一个 handle.

{
  co_await suspend_always();
  try
  ...

OK, 此时直接 suspend, 进入到他的这个空的 await_suspend(std::coroutine_handle<>) 里面之后, invoke Task::promise_type 的 get_result_object() 返回到调用者. 结束.

对于第二个需求, 需要 Task 作为一个 Awaitable, 不过这个 Awaitable 并不难实现. 主要就是捕获一个 await 他的协程的 handle 并且存到某个地方. 至于这个地方是存到 Task 呢还是 Task::promise_type 呢还是 Awaitable 呢? 这个就需要思考一下了. 一个直接的想法当然是存到 promise_type 里面, 因为这个东西就是我们协程的上下文.

重复一下调用关系:

// 1
Task<> coro1(){
    //... body // synchronous until reach the first not Task awaitable
}
// 2
AsyncVoid coro2(){
    co_await coro1(); // synchronous
}
// 3
int main(void){
    coro2();  // asynchronous
}

首先思考一下我们什么时候恢复这个 handle (2) 的执行呢? 要等到我内部所有逻辑都执行完了才行! 但是我不知道我会什么时候执行完 (我 (1) 的 body 是不可控的).

作为 awaiter 来说(1), await_suspend 捕获了调用者(2)的 handle 之后如果什么都不做, 自然就会以 get_return_object 的形式 fall back 到第一个非 await 运行协程的人 (3) 身上.

所以当前我 (1) 在 await_suspend 里面保存了 await 调用者 (2) 的 handle 之后要做的事情是马上运行我自己的 handle. 由于新的 C++20 coroutine (非 TS) 支持了 std::coroutine_handle<> await_suspend( std::coroutine_handle<>) 我们要做的就很简单了:

template <typename T>
class task {
  using value_type = T
  // leave a handle for the co_await caller
  std::coroutine_handle<task_promise> my_continuation;
    
  // As Awaiter:
  // TODO: add cached result
  bool await_ready() const noexcept { return false; }
  std::coroutine_handle<> await_suspend(std::coroutine_handle<> who_await_me) {
    // save who await me for resuming
    my_continuation.promise().set_my_awaiter(who_await_me);
    return my_continuation;
  }
  // TODO: return a result
  value_type await_resume() {}
    
  // As coroutine:
  struct promise_type {
    std::coroutine_handle<> who_await_me_;
    task get_return_object() {
      return {std::coroutine_handle<promise_type>::from_promise(*this)};
    }
    std::suspend_always initial_suspend() noexcept { return {}; }
    std::suspend_never final_suspend() noexcept { return {}; }
    void unhandled_exception() noexcept {}
    void set_my_awaiter(std::coroutine_handle<> who_await_me) {
      who_await_me_ = who_await_me;
    }
  }
};

现在的功能已经实现了一部分了. 然而除了 result 还有一块重点没做的, 就是我们没有恢复我们的 awaiter.

注意, 这个时候如果执行 body 的话, 然后触发到一个非 Task 的 await awaitable, 我们的 handle 将会被他捕获. 随后将会触发一个 get_return_object. 下面理一下吧.

先重复一下调用关系:

// 1
Task<> coro1(){
    //... body // synchronous until reach the first not Task awaitable
    await 某个不会恢复的 awaitable
}
// 2
AsyncVoid coro2(){
    co_await coro1(); // synchronous
}
// 3
int main(void){
    coro2();  // asynchronous
    ...
}

暂时来理一下这个时候整个流程发生了什么. main 调用了 coro2, coro2 由于 initial 是 suspend_never, 所以他会直接进入 body.

第一行 co_await coro1() 此时 coro1 是一个右值 根据下面的表达式构造了一个 Task 返回:

    task get_return_object() {
      return {std::coroutine_handle<promise_type>::from_promise(*this)};
    }

然后 co_await 作用触发了 Task 作为 Awaiter 的属性, 于是触发了 await_suspend 捕获了 coro2 的 handle, 随后恢复 coro1 的运行:

  std::coroutine_handle<> await_suspend(std::coroutine_handle<> who_await_me) {
    // save who await me for resuming
    my_continuation.promise().set_my_awaiter(who_await_me);
    return my_continuation;
  }

恢复了 coro1 的运行后, await 某个不会恢复的 awaitable 被触发, 此时 coro1 的 handle 被某个不会恢复的 awaitable捕获, 随后 fall back 到 main 的 coro2() 触发 coro2() 作为协程的 AsyncVoid::promise_type 的 get_return_object 返回一个空的 AsynVoid.

为什么直接 fall back 到了 main 里面呢? 这个其实很好理解的, 这里的 Task 只有 coro1 和 coro2 是协程, 而 coro1 的 get_return_object 早就触发结束了(作为一个右值返回), 所以理论上 coro1 作为协程的首次运行已经结束了, 我们当前其实是在 coro2 的上下文里面恢复的 coro1.

如果此时再运行一个 coro1, 将会得到第二个 coro1 协程的 Task.(编译器在 heap 上再分配一个 Task::promise_type, 然后 co_await suspend_always 触发一次 get_return_object, 然后构造一个新的 Task).

下面我们补上 who_await_me 的 resume, 一个没有 result 的 Task 就基本写好了:

  struct promise_type {
    std::coroutine_handle<> who_await_me_;
    // As coroutine:
    task get_return_object() {
      return {std::coroutine_handle<promise_type>::from_promise(*this)};
    }
    std::suspend_always initial_suspend() noexcept { return {}; }
 +  struct final_awaitable {
 +    bool await_ready() const noexcept { return false; }
 + 
 +    std::coroutine_handle<> await_suspend(
 +        std::coroutine_handle<promise_type> me) noexcept {
 +      return me.promise().who_await_me_;
 +    }
 +    void await_resume() noexcept {}
 +  };
 -  std::suspend_never final_suspend() noexcept { return {}; }
 +  aoto final_suspend() noexcept { return final_awaitable; }
    void unhandled_exception() noexcept {}
    void set_my_awaiter(std::coroutine_handle<> who_await_me) {
      who_await_me_ = who_await_me;
    }
  }

当然, 协程的生命周期还没有进行管控, 此时的 Task 是一个会内存泄漏的协程类.

---

思路: 剩余的部分

时间关系暂时留白, …

---

附录

烂尾总是不好的。所以我补充一下直接从 cppcoro 里面改编一下（主要是适配 g+±11 以及可读性）的整个 task 的源码，至于 task 对异步的意义，我后面有时间再做几篇笔记。实际看他的代码就明白 co_return 是怎么被支持的了。异常是怎么处理的，以及后面的避免内存泄漏的析构函数。（理论上可以编译 g+±11 下）。很遗憾烂尾了。但是实际是 cppcoro 已经写好了，其实不要重复造轮子吧。写笔记太累了
（broken promise 也是 cppcoro 里写了，但是为了方便暂时改成抛字符串吧，单文件编译也能用，如果感兴趣直接 clone cppcoro 学习吧，这个代码主要可能还是用于g+±11 单文件编译学习 task 用的）

/// The original source code is from cppcoro, Copyright (c) Lewis Baker
/// Licenced under MIT license.

#include <atomic>
#include <exception>
#include <utility>
#include <type_traits>
#include <cstdint>
#include <cassert>
#include <coroutine>

namespace explore_coro {
template <typename T>
class task;

namespace detail {
// use base class is to support task<> (a.k.a. task<void>)
class task_promise_base {
  friend struct final_awaitable;

  // for future resumption of suspended routine
  struct final_awaitable {
    bool await_ready() const noexcept { return false; }

    template <typename Promise>
    std::coroutine_handle<> await_suspend(std::coroutine_handle<Promise> my_handle) noexcept {
      return my_handle.promise().who_await_me_;
    }
    void await_resume() noexcept {}
  };

 public:
  task_promise_base() noexcept {};

  auto initial_suspend() noexcept { return std::suspend_always{}; }

  auto final_suspend() noexcept { return final_awaitable{}; }

  void set_continuation(std::coroutine_handle<> h) noexcept { who_await_me_ = h; }

 private:
  std::coroutine_handle<> who_await_me_;
};

template <typename T>
class task_promise final : public task_promise_base {
 public:
  task_promise() noexcept {};

  ~task_promise() {
    switch (m_resultType) {
      case result_type::value:
        m_value.~T();
        break;
      case result_type::exception:
        m_exception.~exception_ptr();
        break;
      default:
        break;
    }
  }

  task<T> get_return_object() noexcept;

  void unhandled_exception() noexcept {
    ::new (static_cast<void *>(std::addressof(m_exception))) std::exception_ptr(std::current_exception());
    m_resultType = result_type::exception;
  }

  template <typename VALUE, typename = std::enable_if_t<std::is_convertible_v<VALUE &&, T>>>
  void return_value(VALUE &&value) noexcept(std::is_nothrow_constructible_v<T, VALUE &&>) {
    ::new (static_cast<void *>(std::addressof(m_value))) T(std::forward<VALUE>(value));
    m_resultType = result_type::value;
  }

  T &result() & {
    if (m_resultType == result_type::exception) {
      std::rethrow_exception(m_exception);
    }

    assert(m_resultType == result_type::value);

    return m_value;
  }

 private:
  enum class result_type { empty, value, exception };

  result_type m_resultType = result_type::empty;

  union {
    T m_value;
    std::exception_ptr m_exception;
  };
};

template <>
class task_promise<void> : public task_promise_base {
 public:
  task_promise() noexcept = default;

  task<void> get_return_object() noexcept;

  void return_void() noexcept {}

  void unhandled_exception() noexcept { m_exception = std::current_exception(); }

  void result() {
    if (m_exception) {
      std::rethrow_exception(m_exception);
    }
  }

 private:
  std::exception_ptr m_exception;
};

template <typename T>
class task_promise<T &> : public task_promise_base {
 public:
  task_promise() noexcept = default;

  task<T &> get_return_object() noexcept;

  void unhandled_exception() noexcept { m_exception = std::current_exception(); }

  void return_value(T &value) noexcept { m_value = std::addressof(value); }

  T &result() {
    if (m_exception) {
      std::rethrow_exception(m_exception);
    }

    return *m_value;
  }

 private:
  T *m_value = nullptr;
  std::exception_ptr m_exception;
};
}  // namespace detail

/// \brief
/// A task represents an operation that produces a result both lazily
/// and asynchronously.
///
/// When you call a coroutine that returns a task, the coroutine
/// simply captures any passed parameters and returns exeuction to the
/// caller. Execution of the coroutine body does not start until the
/// coroutine is first co_await'ed.
template <typename T = void>
class [[nodiscard]] task {
 public:
  using promise_type = detail::task_promise<T>;

  using value_type = T;

 private:
  struct awaitable_base {
    std::coroutine_handle<promise_type> my_continuation_;

    explicit awaitable_base(std::coroutine_handle<promise_type> coroutine) noexcept : my_continuation_(coroutine) {}

    bool await_ready() const noexcept { return !my_continuation_ || my_continuation_.done(); }

    std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaitingCoroutine) noexcept {
      my_continuation_.promise().set_continuation(awaitingCoroutine);
      return my_continuation_;
    }
  };

 public:
  template <typename Scheduler_>
  std::coroutine_handle<> bind_scheduler(Scheduler_ s) {
    s.register_task_handle(this->my_continuation_);
    return this->my_continuation_;
  }

  task() noexcept : my_continuation_(nullptr) {}

  explicit task(std::coroutine_handle<promise_type> coro) : my_continuation_(coro) {}

  task(task &&t) noexcept : my_continuation_(t.my_continuation_) { t.my_continuation_ = nullptr; }

  /// Disable copy construction/assignment.
  task(const task &) = delete;
  task &operator=(const task &) = delete;

  /// Frees resources used by this task.
  ~task() {
    if (my_continuation_) {
      my_continuation_.destroy();
    }
  }

  task &operator=(task &&other) noexcept {
    if (std::addressof(other) != this) {
      if (my_continuation_) {
        my_continuation_.destroy();
      }

      my_continuation_ = other.my_continuation_;
      other.my_continuation_ = nullptr;
    }

    return *this;
  }

  /// \brief
  /// Query if the task result is complete.
  ///
  /// Awaiting a task that is ready is guaranteed not to block/suspend.
  bool is_ready() const noexcept { return !my_continuation_ || my_continuation_.done(); }

  auto operator co_await() const &noexcept {
    struct awaitable : awaitable_base {
      using awaitable_base::awaitable_base;

      decltype(auto) await_resume() {
        if (!this->my_continuation_) {
          throw "broken_promise{}";
        }

        return this->my_continuation_.promise().result();
      }
    };

    return awaitable{my_continuation_};
  }

  auto operator co_await() const &&noexcept {
    struct awaitable : awaitable_base {
      using awaitable_base::awaitable_base;

      decltype(auto) await_resume() {
        if (!this->my_continuation_) {
          throw "broken_promise{}";
        }

        return std::move(this->my_continuation_.promise()).result();
      }
    };

    return awaitable{my_continuation_};
  }

  /// \brief
  /// Returns an awaitable that will await completion of the task without
  /// attempting to retrieve the result.
  auto when_ready() const noexcept {
    struct awaitable : awaitable_base {
      using awaitable_base::awaitable_base;

      void await_resume() const noexcept {}
    };

    return awaitable{my_continuation_};
  }

 private:
  std::coroutine_handle<promise_type> my_continuation_;
};
struct async_run {
  struct promise_type {
    async_run get_return_object() { return {}; }
    std::suspend_never initial_suspend() { return {}; }
    std::suspend_never final_suspend() noexcept { return {}; }
    void unhandled_exception() {}
    void return_void() {}
  };
};

template <typename Executor, typename Task>
void co_spawn(Executor&& executor, Task&&task){
  task.bind_scheduler(executor);
  executor.notify();
}

namespace detail {
template <typename T>
task<T> task_promise<T>::get_return_object() noexcept {
  return task<T>{std::coroutine_handle<task_promise>::from_promise(*this)};
}

inline task<void> task_promise<void>::get_return_object() noexcept { return task<void>{std::coroutine_handle<task_promise>::from_promise(*this)}; }

template <typename T>
task<T &> task_promise<T &>::get_return_object() noexcept {
  return task<T &>{std::coroutine_handle<task_promise>::from_promise(*this)};
}

}  // namespace detail
}  // namespace explore_coro

#endif