如何在其链的上下文中恢复堆栈协程的执行？

Question

using Yield = asio::yield_context;
using boost::system::error_code;
int Func(Yield yield) {
  error_code ec;
  asio::detail::async_result_init<Yield, void(error_code, int)> init(yield[ec]);
  std::thread th(std::bind(Process, init.handler));
  int result = init.result.get();  // <--- yield at here
  return result;
}

如何实现Process，以便Func 将在Func 最初产生的链的上下文中恢复？

Answer 1

您通过在 Boost Asio 提供的执行器框架之外创建线程来使事情复杂化。

因此，您不应该假设您想要的东西是可能的。我强烈建议向io_service 添加更多线程并让它为您管理线程。

或者，您可以扩展库并添加您显然想要的新功能。如果是这样，最好联系开发者邮件列表寻求建议。也许他们欢迎这个功能¹？

---

¹（有趣的是，你没有描述，所以我不会问它的目的是什么）

Answer 2

using CallbackHandler = boost::asio::handler_type<Yield, void (error_code, int)>::type;

void Process(CallbackHandler handler) {
  int the_result = 81;
  boost::asio::detail::asio_handler_invoke(
      std::bind(handler, error_code(), the_result), &handler);
}

在@sehe 的提示下，我提出了上述可行的解决方案。但我不确定这是否是正确的/惯用的/最好的方法。欢迎评论/编辑此答案。

Answer 3

Boost.Asio 使用辅助函数asio_handler_invoke 为调用策略提供自定义点。例如，当 Handler 被 strand 包裹时，调用策略将导致在调用时通过 strand 调度处理程序。如documentation 中所述，asio_handler_invoke 应通过依赖于参数的查找来调用。

using boost::asio::asio_handler_invoke;
asio_handler_invoke(nullary_functor, &handler);

---

对于堆栈式协程，在产生协程以及调用与yield_context 关联的handler_type 以恢复协程时，需要考虑各种重要的细节：

• 如果代码当前在协程中运行，那么它在与协程关联的strand 内。本质上，一个简单的处理程序被 strand 包裹，它恢复协程，导致执行跳转到协程，阻塞当前在strand 中的处理程序。当协程产生时，执行会跳回到strand 处理程序，让它完成。
• 虽然spawn() 将工作添加到io_service（将启动并跳转到协程的处理程序），但协程本身并不工作。为了防止 io_service 事件循环在协程未完成时结束，可能需要在屈服之前向 io_service 添加工作。
• 堆栈式协程使用strand 来帮助保证协程yields 在resume 被调用之前。 Asio 1.10.6 / Boost 1.58 启用能够从启动函数中安全地调用完成处理程序。以前的版本要求完成处理程序不能从启动函数中调用，因为它的调用策略会dispatch()，导致协程在被挂起之前尝试恢复。

这是一个完整的example，它说明了这些细节：

#include <iostream>    // std::cout, std::endl
#include <chrono>      // std::chrono::seconds
#include <functional>  // std::bind
#include <thread>      // std::thread
#include <utility>     // std::forward
#include <boost/asio.hpp>
#include <boost/asio/spawn.hpp>

template <typename CompletionToken, typename Signature>
using handler_type_t = typename boost::asio::handler_type<
  CompletionToken, Signature>::type;

template <typename Handler>
using async_result = boost::asio::async_result<Handler>;

/// @brief Helper type used to initialize the asnyc_result with the handler.
template <typename CompletionToken, typename Signature>
struct async_completion
{
  typedef handler_type_t<CompletionToken, Signature> handler_type;

  async_completion(CompletionToken&& token)
    : handler(std::forward<CompletionToken>(token)),
      result(handler)
  {}

  handler_type handler;
  async_result<handler_type> result;
};

template <typename Signature, typename CompletionToken>
typename async_result<
  handler_type_t<CompletionToken, Signature>
>::type
async_func(CompletionToken&& token, boost::asio::io_service& io_service)
{
  // The coroutine itself is not work, so guarantee the io_service has
  // work.
  boost::asio::io_service::work work(io_service);

  // Initialize the async completion handler and result.
  async_completion<CompletionToken, Signature> completion(
      std::forward<CompletionToken>(token));

  auto handler = completion.handler;
  std::cout << "Spawning thread" << std::endl;
  std::thread([](decltype(handler) handler)
    {
      // The handler will be dispatched to the coroutine's strand.
      // As this thread is not running within the strand, the handler
      // will actually be posted, guaranteeing that yield will occur
      // before the resume.
      std::cout << "Resume coroutine" << std::endl;
      using boost::asio::asio_handler_invoke;
      asio_handler_invoke(std::bind(handler, 42), &handler);
    }, handler).detach();

  // Demonstrate that the handler is serialized through the strand by
  // allowing the thread to run before suspending this coroutine.
  std::this_thread::sleep_for(std::chrono::seconds(2));

  // Yield the coroutine.  When this yields, execution transfers back to
  // a handler that is currently in the strand.  The handler will complete
  // allowing other handlers that have been posted to the strand to run.
  std::cout << "Suspend coroutine" << std::endl;
  return completion.result.get();
}

int main()
{
  boost::asio::io_service io_service;

  boost::asio::spawn(io_service,
    [&io_service](boost::asio::yield_context yield)
    {
      auto result = async_func<void(int)>(yield, io_service);
      std::cout << "Got: " << result << std::endl;
    });

  std::cout << "Running" << std::endl;
  io_service.run();
  std::cout << "Finish" << std::endl;
}

输出：

Running
Spawning thread
Resume coroutine
Suspend coroutine
Got: 42
Finish

---

有关更多详细信息，请考虑阅读Library Foundations for Asynchronous Operations。它更详细地介绍了异步操作的组成、Signature 如何影响async_result，以及async_result、handler_type 和async_completion 的整体设计。

Answer 4

以下是 Boost 1.66.0 的更新示例，基于 Tanner 的出色回答：

#include <iostream>    // std::cout, std::endl
#include <chrono>      // std::chrono::seconds
#include <functional>  // std::bind
#include <thread>      // std::thread
#include <utility>     // std::forward
#include <boost/asio.hpp>
#include <boost/asio/spawn.hpp>

template <typename Signature, typename CompletionToken>
auto async_add_one(CompletionToken token, int value) {
    // Initialize the async completion handler and result
    // Careful to make sure token is a copy, as completion's handler takes a reference
    using completion_type = boost::asio::async_completion<CompletionToken, Signature>;
    completion_type completion{ token };

    std::cout << "Spawning thread" << std::endl;
    std::thread([handler = completion.completion_handler, value]() {
        // The handler will be dispatched to the coroutine's strand.
        // As this thread is not running within the strand, the handler
        // will actually be posted, guaranteeing that yield will occur
        // before the resume.
        std::cout << "Resume coroutine" << std::endl;

        // separate using statement is important
        // as asio_handler_invoke is overloaded based on handler's type
        using boost::asio::asio_handler_invoke;
        asio_handler_invoke(std::bind(handler, value + 1), &handler);
    }).detach();

    // Demonstrate that the handler is serialized through the strand by
    // allowing the thread to run before suspending this coroutine.
    std::this_thread::sleep_for(std::chrono::seconds(2));

    // Yield the coroutine.  When this yields, execution transfers back to
    // a handler that is currently in the strand.  The handler will complete
    // allowing other handlers that have been posted to the strand to run.
    std::cout << "Suspend coroutine" << std::endl;
    return completion.result.get();
}

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

    boost::asio::spawn(
        io_context,
        [&io_context](boost::asio::yield_context yield) {
            // Here is your coroutine

            // The coroutine itself is not work, so guarantee the io_context
            // has work while the coroutine is running
            const auto work = boost::asio::make_work_guard(io_context);

            // add one to zero
            const auto result = async_add_one<void(int)>(yield, 0);
            std::cout << "Got: " << result << std::endl; // Got: 1

            // add one to one forty one
            const auto result2 = async_add_one<void(int)>(yield, 41);
            std::cout << "Got: " << result2 << std::endl; // Got: 42
        }
    );

    std::cout << "Running" << std::endl;
    io_context.run();
    std::cout << "Finish" << std::endl;
}

输出：

Running
Spawning thread
Resume coroutine
Suspend coroutine
Got: 1
Spawning thread
Resume coroutine
Suspend coroutine
Got: 42
Finish

备注：

• 充分利用 Tanner 的回答
• 首选网络 TS 命名（例如 io_context）
• boost::asio 提供了一个 async_completion 类，它封装了处理程序和 async_result。当处理程序引用 CompletionToken 时要小心，这就是现在显式复制令牌的原因。这是因为通过 async_result (completion.result.get) 让出将使相关的 CompletionToken 放弃其底层强引用。这最终可能导致协程意外提前终止。
• 明确指出单独的using boost::asio::asio_handler_invoke 语句非常重要。显式调用可以防止调用正确的重载。

-

我还要提到，我们的应用程序以两个 io_context 结束，协程可以与之交互。具体来说，一个用于 I/O 绑定工作的上下文，另一个用于 CPU。使用带有boost::asio::spawn 的显式链最终为我们提供了对协程运行/恢复的上下文的明确控制。这帮助我们避免了零星的 BOOST_ASSERT(!is_running()) 失败。

使用显式链创建协程：

auto strand = std::make_shared<strand_type>(io_context.get_executor());
boost::asio::spawn(
    *strand,
    [&io_context, strand](yield_context_type yield) {
        // coroutine
    }
);

调用显式调度到链（多 io_context 世界）：

boost::asio::dispatch(*strand, [handler = completion.completion_handler, value] {
    using boost::asio::asio_handler_invoke;
    asio_handler_invoke(std::bind(handler, value), &handler);
});

-

我们还发现，在 async_result 签名中使用未来允许异常传播回协程恢复。

using bound_function = void(std::future<RETURN_TYPE>);
using completion_type = boost::asio::async_completion<yield_context_type, bound_function>;

产量为：

auto future = completion.result.get();
return future.get(); // may rethrow exception in your coroutine's context