Async Scope

之前的章节只是讲解了协程中的各种概念

本章我们来实现一个async scope，即一个可以在其中使用co_await和co_return的函数

最后我们要在这个基础上实现一个单元素的回调转协程

Promise实现

这里我们直接把整个Scope建模为一个Task

先看看整体的样式

Task<int> simple_task2() {
  std::cout << __func__ << "\n";
  co_return 2;
}
Task<int> simple_task1() {
  std::cout << __func__ << "\n";
  co_return 1;
}

Task<int> simple_task() {
  // result2 == 2

  auto result2 = co_await simple_task1();
  std::cout << __func__ << " co_await task1 \n";

  // result3 == 3
  auto result3 = co_await simple_task2();
  std::cout << __func__ << " co_await task2 \n";

  co_return 1 + result2 + result3;
}

这里可以推断出两个要做的事情

• co_await 对应的await_transform允许接受Task作为参数并返回一个Awaiter
• co_return 对应的return_value允许接受T来作为整个任务的结果值

先考虑一个单独异步任务，即我们最常见的那种Promise/Future模型，promise负责支持complete触发回调，future负责挂载回调，那么我们就可以这样考虑：

promise::complete 的实现就是 return_value 的实现

  void return_value(ResultType value) {
      done = true;
      res = Result<ResultType>(std::move(value));
      for (auto &callback : completion_callbacks) {
        callback(res);
      }
    }

那么下一步的设计就该考虑completion_callbacks放在哪里了

首先按照习惯性设计外部的Future（这里是C++的协程概念中的future，就是promise外面那个）是会持有当前的coroutine_handle的通过这个很容易拿到promise，但是promise是不知道future的，所以我们直接把completion_callbacks放在promise内部即可。

这里跟co_await实现倒是没什么关系。。。只是给一个外部可以给task挂载回调的机制罢了

 Task &finally(std::function<void()> &&func) {
    handle.promise().on_completed([func](auto result) { func(); });
    return *this;
  }

接下来就是暂存结果值和实现三大件了

都很简单

struct promise_type {
    auto initial_suspend() { return std::suspend_never{}; }
    auto final_suspend() noexcept { return std::suspend_always{}; }
    auto unhandled_exception() {
      done = true;
      res = Result<ResultType>(std::current_exception());
      for (auto &callback : completion_callbacks) {
        callback(res);
      }
    }
    // 定制co_return行为 当调用这个方法时意味着此时
    void return_value(ResultType value) {
      done = true;
      res = Result<ResultType>(std::move(value));
      for (auto &callback : completion_callbacks) {
        callback(res);
      }
    }

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

    template <typename _ResultType>
    TaskAwaiter<_ResultType> await_transform(Task<_ResultType> &&task) {
      return TaskAwaiter<_ResultType>(std::move(task));
    }


    void on_completed(std::function<void(Result<ResultType>)> &&func) {
      if (done) {
        // result 已经有值
        auto value = res;
        // 解锁之后再调用 func
        func(value);
      } else {
        // 否则添加回调函数，等待调用
        completion_callbacks.push_back(func);
      }
    }

    auto get_result() { return res.get_or_throw(); }

  public:
    Result<ResultType> res;
    bool done = false;
    std::list<std::function<void(Result<ResultType>)>> completion_callbacks;
  };

Awaitable

剩下的就是awaitable实现，这个主要用于co_await这个操作符的返回值，此时我们需要根据其右侧的状态来决定是否挂起以及返回值了，就是等待子任务·

我们再看一眼这个函数的签名，注意Task是个C++的Future哦

template <typename _ResultType>
    TaskAwaiter<_ResultType> await_transform(Task<_ResultType> &&task) {
      return TaskAwaiter<_ResultType>(std::move(task));
    }

直接看代码吧 很简单

   //因为task是future所以我们有个handle指针再获取到promise再获取到done变量状态 
   bool await_ready() const noexcept {
      return task.handle.promise().done;
    }

//注意这个handle代指的是当前continuation，是调用方的句柄，当我resume时，等价于从co_await处向下执行 而task则是代指的子任务
    void await_suspend(std::coroutine_handle<> handle) noexcept {
      // 当 task 执行完之后调用 resume
      task.finally([handle]() { handle.resume(); });
    }

    // 协程恢复执行时，被等待的子 Task 已经执行完，调用 get_result 来获取结果
    R await_resume() noexcept { return task.get_result(); }

简单的一个async scope就完成了

回调转协程

这个就更简单啦

我们要实现的结构类似于js中的

fn(args,(i) => {fun1(i)});

fun1(await fn(args))

状态储存

这个就有点像我们Vert.x的Promise/Future实现了

template <typename T> struct AsyncResult {
  explicit AsyncResult() = default;

  // 当 Task 正常返回时用结果初始化 Result
  explicit AsyncResult(T &&value)
      : res(value),
        completion_callbacks(std::move(value.completion_callbacks)) {}

  // 这里先暂时一把大锁控制一下 好理解
  void complete(Result<T> &&async_value) {
    std::cout << __func__ << "AsyncResult 准备触发回调 \n";
    auto scope = std::lock_guard(queue_lock);
    done = true;
    res = async_value;
    for (auto fn : completion_callbacks) {
      fn(res);
    }
    completion_callbacks.clear();
  }

  void on_completed(std::function<void(Result<T>)> &&func) {
      auto scope = std::lock_guard(queue_lock);
    if (done) {
      // result 已经有值
      auto value = res;
      // 解锁之后再调用 func
      func(value);
    } else {
      // 否则添加回调函数，等待调用
      completion_callbacks.push_back(func);
    }
  }

  bool is_done() { return done; }

  Result<T> get_result() { return res; }

private:
  std::mutex queue_lock;
  std::list<std::function<void(Result<T>)>> completion_callbacks;
  Result<T> res;
  bool done = false;
};

awaitable

既然要允许await那就得把之前实现的AsyncResult添加点awaitable实现

这里我们采用组合的方式来做

这里是一个多所有权的场景 直接用share_ptr把这个玩意让协程和异步任务共同持有

我思考了一下，这里因为存在一定的顺序性 raw ptr也是可行的

异步任务持有指针不释放

等待awaitable析构时自然释放即可

template <typename T> struct AsyncResultAwaiter {
//省略构造参数
  bool await_ready() const noexcept { return res->is_done(); }

  void await_suspend(std::coroutine_handle<> handle) noexcept {
    // 当 task 执行完之后调用 resume
    res->on_completed([handle](auto a) {
      std::cout << __func__ << "AsyncResultAwaiter  回调 \n";
      std::cout << handle.address();
      handle.resume();
    });
  }

  Result<T> await_resume() noexcept {
    return res.get()->get_result();
  }

  std::shared_ptr<AsyncResult<T>> res;
  std::coroutine_handle<> handle;
};

这里就相当于kt的那种做法，直接拿到coroutine_handle来用

最后我们在之前实现的task的promise中添加对它的支持就行了

这里的参数名为future的参数就是与异步任务共享的那个回调，这里只是为了能拿到协程句柄丢到对应的异步任务回调中罢了

template <typename _ResultType>
    AsyncResultAwaiter<_ResultType>
    await_transform(std::shared_ptr<AsyncResult<_ResultType>> future) {
      return AsyncResultAwaiter(
          future, std::coroutine_handle<promise_type>::from_promise(*this));
    }

用法

Task<int> async_task() {
  // result2 == 2
  auto shared_ptr = std::make_shared<AsyncResult<int>>();
  auto thread = std::thread([shared_ptr]() {
    std::cout << "thread :in lambda \n";
    std::this_thread::sleep_for(std::chrono::seconds(1));
    shared_ptr->complete(Result(12));
  });
  thread.detach();
  std::cout << "before await count:" << shared_ptr.use_count() << "\n";
  auto res = co_await shared_ptr;
  std::cout << "res:" << res.get_or_throw() << "\n";
  co_return res.get_or_throw() + 3;
}

总结

这里面我只是实现了 而非打磨好了 其中还有一些内存所有权的问题还没有解决甚至存在内存泄漏

但是先学会再打磨

完整代码参考codepieces/async_scope_and_future.cpp at main · dreamlike-ocean/codepieces · GitHub