Resuming asio coroutine from another thread

Question

I have a problem with resuming boost::asio coroutine from another thread. Here is sample code:

#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;
}

The problem is that the asio::steady_timer object is not thread safe and the program crashes. But if I try to use mutex to synchronize access to it then I have a deadlock because the scope of foo is not leaved.

#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;
}

There is no such a problem if I use standard completion handler instead of coroutines.

#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;
}

Is it possible to have behavior similar to the last example when couroutines are used.

score 6 · Accepted Answer · edited May 23 '17 at 12:09

A coroutine runs within the context of a strand. In spawn(), if one is not explicitly provided, a new strand will be created for the coroutine. By explicitly providing strand to spawn(), one can post work into the strand that will be synchronized with the coroutine.

Also, as noted by sehe, undefined behavior may occur if the coroutine is running in one thread, acquires a mutex lock, then suspends, but resumes and runs in a different thread and releases the lock. To avoid this, ideally one should not hold locks while the coroutine suspends. However, if it is necessary, one must guarantee that the coroutine runs within the same thread when it is resumed, such as by only running the io_service from a single thread.

Here is the minimal complete example based on the original example where bar() posts work into a strand to cancel the timer, causing the foo() coroutine to resume:

#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();
}

Which provides the following output:

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

As covered in this answer, when the boost::asio::yield_context detects that the asynchronous operation has failed, such as when the operation is canceled, it converts the boost::system::error_code into a system_error exception and throws. The above example uses yield_context::operator[] to allow the yield_context to populate the provided error_code on failure instead of throwing throwing.

Isn't it UB to hold the mutex across `async_wait` in the first place? You can't just unlock the same mutex from potentially a different thread (of course if a single thread runs the `io_service` that is only an academic issue) — sehe, Feb 02 '15 at 14:19
@sehe Thank you. The unlock certainly would invoke undefined behavior if the coroutine's resumes execution in a different thread. I have updated my answer to include that detail. — Tanner Sansbury, Feb 02 '15 at 14:46
Thank you for the answer. From the very beginning the problem was that I'm forgot to use error code handling in the coroutine but the timer cancel results in _Operation canceled_ error and then exception is thrown. Of course as it is pointed in the documentation the timer objects are not thread safe and post to `io_service` or `strand` is needed to execute the cancel operation in the context of the same thread which starts the timer. Instead of posting to `strand` the `io_service` itself is enough because the `strand` is needed only if multiple threads call `io_service::run`. — bobeff, Feb 02 '15 at 15:03

Resuming asio coroutine from another thread

1 Answers1