从另一个线程恢复 asio 协程

Resuming asio coroutine from another thread

本文关键字：asio 协程 恢复 线程 另一个      更新时间：2023-10-16

我从另一个线程恢复 boost：：asio 协程时遇到问题。下面是示例代码：

#include <iostream>
#include <thread>
#include <boost/asio.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/spawn.hpp>
using namespace std;
using namespace boost;
void foo(asio::steady_timer& timer, asio::yield_context yield)
{
    cout << "Enter foo" << endl;
    timer.expires_from_now(asio::steady_timer::clock_type::duration::max());
    timer.async_wait(yield);
    cout << "Leave foo" << endl;
}
void bar(asio::steady_timer& timer)
{
    cout << "Enter bar" << endl;
    sleep(1); // wait a little for asio::io_service::run to be executed
    timer.cancel();
    cout << "Leave bar" << endl;
}
int main()
{
    asio::io_service ioService;
    asio::steady_timer timer(ioService);
    asio::spawn(ioService, bind(foo, std::ref(timer), placeholders::_1));
    thread t(bar, std::ref(timer));
    ioService.run();
    t.join();
    return 0;
}

问题是 asio：：steady_timer 对象不是线程安全的，程序崩溃。但是，如果我尝试使用互斥锁来同步对它的访问，那么我就会死锁，因为 foo 的范围没有离开。

#include <iostream>
#include <thread>
#include <mutex>
#include <boost/asio.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/spawn.hpp>
using namespace std;
using namespace boost;
void foo(asio::steady_timer& timer, mutex& mtx, asio::yield_context yield)
{
    cout << "Enter foo" << endl;
    {
        lock_guard<mutex> lock(mtx);
        timer.expires_from_now(
            asio::steady_timer::clock_type::duration::max());
        timer.async_wait(yield);
    }
    cout << "Leave foo" << endl;
}
void bar(asio::steady_timer& timer, mutex& mtx)
{
    cout << "Enter bar" << endl;
    sleep(1); // wait a little for asio::io_service::run to be executed
    {
        lock_guard<mutex> lock(mtx);
        timer.cancel();
    }
    cout << "Leave bar" << endl;
}
int main()
{
    asio::io_service ioService;
    asio::steady_timer timer(ioService);
    mutex mtx;
    asio::spawn(ioService, bind(foo, std::ref(timer), std::ref(mtx),
        placeholders::_1));
    thread t(bar, std::ref(timer), std::ref(mtx));
    ioService.run();
    t.join();
    return 0;
}

如果我使用标准完成处理程序而不是协程，则没有这样的问题。

#include <iostream>
#include <thread>
#include <mutex>
#include <boost/asio.hpp>
#include <boost/asio/steady_timer.hpp>
using namespace std;
using namespace boost;
void baz(system::error_code ec)
{
    cout << "Baz: " << ec.message() << endl;
}
void foo(asio::steady_timer& timer, mutex& mtx)
{
    cout << "Enter foo" << endl;
    {
        lock_guard<mutex> lock(mtx);
        timer.expires_from_now(
            asio::steady_timer::clock_type::duration::max());
        timer.async_wait(baz);
    }
    cout << "Leave foo" << endl;
}
void bar(asio::steady_timer& timer, mutex& mtx)
{
    cout << "Enter bar" << endl;
    sleep(1); // wait a little for asio::io_service::run to be executed
    {
        lock_guard<mutex> lock(mtx);
        timer.cancel();
    }
    cout << "Leave bar" << endl;
}
int main()
{
    asio::io_service ioService;
    asio::steady_timer timer(ioService);
    mutex mtx;
    foo(std::ref(timer), std::ref(mtx));
    thread t(bar, std::ref(timer), std::ref(mtx));
    ioService.run();
    t.join();
    return 0;
}

使用couroutine时，是否有可能具有类似于上一个示例的行为。

协

程在strand的上下文中运行。 在spawn()中，如果没有显式提供，将为协程创建一个新的strand。 通过显式地向spawn()提供strand，可以将工作发布到将与协程同步的strand中。

此外，正如 sehe 所指出的，如果协程在一个线程中运行，获取互斥锁，然后挂起，但恢复并在另一个线程中运行并释放锁，则可能会出现未定义的行为。 为了避免这种情况，理想情况下，不应在协程挂起时保持锁。 但是，如有必要，必须保证协程在恢复时在同一线程中运行，例如仅从单个线程运行io_service。

以下是基于原始示例的最小完整示例，其中bar()帖子工作到strand中以取消计时器，从而导致foo()协程恢复：

#include <iostream>
#include <thread>
#include <boost/asio.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/steady_timer.hpp>
void foo(boost::asio::steady_timer& timer, boost::asio::yield_context yield)
{
  std::cout << "Enter foo" << std::endl;
  timer.expires_from_now(
      boost::asio::steady_timer::clock_type::duration::max());
  boost::system::error_code error;
  timer.async_wait(yield[error]);
  std::cout << "foo error: " << error.message() << std::endl;
  std::cout << "Leave foo" << std::endl;
}
void bar(
  boost::asio::io_service::strand& strand,
  boost::asio::steady_timer& timer
)
{
  std::cout << "Enter bar" << std::endl;
  // Wait a little for asio::io_service::run to be executed
  std::this_thread::sleep_for(std::chrono::seconds(1));
  // Post timer cancellation into the strand.
  strand.post([&timer]()
    {
      timer.cancel();
    });
  std::cout << "Leave bar" << std::endl;
}
int main()
{
  boost::asio::io_service io_service;
  boost::asio::steady_timer timer(io_service);
  boost::asio::io_service::strand strand(io_service);
  // Use an explicit strand, rather than having the io_service create.
  boost::asio::spawn(strand, std::bind(&foo, 
      std::ref(timer), std::placeholders::_1));
  // Pass the same strand to the thread, so that the thread may post
  // handlers synchronized with the foo coroutine.
  std::thread t(&bar, std::ref(strand), std::ref(timer));
  io_service.run();
  t.join();
}

它提供以下输出：

Enter foo
Enter bar
foo error: Operation canceled
Leave foo
Leave bar

如本答案所述，当boost::asio::yield_context检测到异步操作失败时（例如取消操作时），它会将boost::system::error_code转换为system_error异常并引发。 上面的示例使用 yield_context::operator[] 允许yield_context在失败时填充提供的error_code，而不是引发抛出。

相关文章：