c++ - 从另一个线程恢复 asio 协程

Question

我在从另一个线程恢复 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;
}

当使用协程时，是否有可能出现类似于上一个示例的行为。

Answer 1

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

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

---

这是最小的完整 example基于原始示例，其中 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

如 this 所述回答，当boost::asio::yield_context检测到异步操作失败时，比如取消操作时，它会转换boost::system::error_code 进入 system_error 异常并抛出。上面的例子使用了 yield_context::operator[]允许 yield_context 在失败时填充提供的 error_code 而不是抛出 throwing。