Events

Introduction

Previously we have seen a simple way to deal with the blocking behavior of coroutines when used as cooperative tasks.

A task is blocked if it is waiting for a some operation to complete. Examples are waiting for I/O, waiting for timers to expire, waiting for external signals etc..

The problem of handling blocking function can be generalized as the problem of waiting for some events to be signaled: in fact a function that blocks can also be modeled as a function that starts an asynchronous operation and then waits for it to complete. The completion of the operation is the event to be signaled.

Simple events are simply On/Off. They have been signaled or they haven't. More complex events also carry information. An event that signal the completion of a read operation may communicate the amount of data read and whether an error has occurred of not.

Boost.Coroutine provides generalized functionalities for event waiting.

Futures

A future object holds the result of an asynchronous computation. When an asynchronous computation is started it returns a future. At any time, a task can query the future object to detect if the operation has completed. If the operation is completed, the task can retrieve any extra information provided by the operation completion. If the operation has not completed yet, the task can wait for the operation to complete.

The future interface is modeled in a way that it act as a substitute for the result of an operation. Only when the result is actually needed, the future causes the task to wait for an operation to complete. The act of waiting for the result of an operation through a future is called resolving a future.

In Boost.Coroutine a future is bound to a specific coroutine on creation. When this coroutine wants to wait for an event, it binds the future, with a callback, with the asynchronous operation that is responsible of signaling the event by invoking the callback.

Then the coroutine can use the future to wait for the operation completion. When the coroutine tries to resolve the future, the latter causes the former to yield to the scheduler.

When the operation completes, the callback is invoked, with the results of the operation as parameter, and causes the coroutine to be resumed. From the point of view of the coroutine it is as if the future had returned these values immediately.

Boost.Coroutine also provides the ability to wait for more futures at the same time, increasing efficiency and potentially simplifying some tasks.

The future class template

The following pipe class provides a mean of sending data, of type int, to a listener. A consumer that wants to receive data from the pipe registers a callback with the listen member function. Whenever a producer sends data into the pipe the callback is invoked with the data as parameter:

class pipe {
public:

  void send(int x) {
    m_callback (x);
  }

  template<typename Callback>
  void listen(Callback c) {
    m_callback = c;
  }
private:
  boost::function<void(int)> m_callback;
};

While this class is extremely simple and not really useful, the method of registering a callback to be notified of an event is a very general and common pattern. In the following example a coroutine will be created and a int sent to it through the pipe:

typedef coro::coroutine<void()> coroutine_type;
void consumer_body(coroutine_type::self&, pipe&);
pipe my_pipe;
coroutine_type consumer(boost::bind(consumer_body, _1, my_pipe));
...
consumer_body(std::nothrow);
my_pipe.send(1);
...

A coroutine of type coroutine<void()> is initialized with consumer_body. When the coroutine returns (we will see later why the std::nothrow is needed), an integer is sent trough the pipe to the coroutine.

Let's see how the future class template can be used to wait for the pipe to produce data. This is the implementation of consumer_body

void consumer_body(coroutine_type::self& self, pipe& my_pipe) {
  typedef coro::future<int> future_type;
  future_type future(self);

  my_pipe.listen(coro::make_callback(future));
  assert(!future);
  coro::wait(future);
  assert(future);
  assert(*future == 1);
}

consumer_body creates an instance of future<int> initializing it with a reference self. Then it invokes pipe::listen(), passing as a callback the result of invoking coro::make_callback(). This function returns a function object responsible of assigning a value to the future.

After the asynchronous call to listen has been done, the future is guaranteed not to be resolved until the following call to coro::wait(). This function is responsible of resolving the future. The current coroutine is marked as waiting and control is returned to the caller. It is as if the coroutine had yielded, but no value is returned. In fact coroutine::operator() would throw an exception of type waiting to signal that the current coroutine did not return a value. Passing std::nothrow, as usual, prevents operator() from throwing an exception.

While coroutine<void()> are usually used for cooperative multitasking, Boost.Coroutine doesn't limit in any way the signature of coroutines used with futures.

A waiting coroutine cannot be resumed with operator() and its conversion to bool will return false. Also coroutine::waiting() will return true.

Finally you can't invoke yield(), yield_to(), coroutine::exit() nor coroutine::self::exit() while there are operation pending. Both coroutine::pending() and coroutine::pending() will return the number of pending operations.

An operation is said to be pending if make_callback has been used to create a callback function object from a future for that operation. Also as more experience is gained with this functionality, the restriction of what member functions may be called when there are pending operations might be relaxed.

make_callback() works by returning a function object that when invoked pass its parameter to the future object. Then, if the future is being waited, the associated coroutine will be waken up directly from inside the callback.

The function object returned by make_callback will extend the life time of the coroutine until the callback is signaled. If the signaling causes the coroutine to be resumed, its life time will be extended until the coroutine relinquishes control again. The lifetime is extended by internally using reference counting, thus if the coroutine stores a copy of the callback a cycle can be formed.

A future can only be realized synchronously with the owner coroutine execution. That is, while the operation it is bound to can execute asynchronously, it can only be signaled when the coroutine is not running. This means that a coroutine must enter the wait state for a future to be signaled. It isn't necessarily required that it waits for that specific future to be signaled, only that some events is being waited.

Semantics of future

A future is not _Copyable_ but is Movable.

The future class template models the OptionalPointee concept, that is, has a similar interface to boost::optional.

The conversion to a safe-bool can be used to detect if the future has been signaled or not.

future::operator* returns the realized value. If the future has not been signaled yet, this operator will cause the current coroutine to wait as if it had invoked wait(*this)

future::pending() returns true if the future has been bound to an asynchronous operation.

Assigning an instance of type boost::none_t to a future, causes it to be reseted and return to the non-signaled state. Such a future can be rebound to another asynchronous operation. Resetting a pending() future is undefined behavior.

Multiple parameter futures

It is possible to have futures that represent a tuple of values instead of a single value. For example:

coro::future<int, void*> my_future;

In this operator* will return a tuple of type boost::tuple<int, void*>:

int a;
void * b;

boost::tie(a, b) = *my_future;

If my_future is passed as parameter to make_callback() the equivalent signature of the function object returned by this function will be:

void(int, void*)

This is useful whenever a an asynchronous function may return more than one parameter.

Waiting for multiple futures

Boost.Coroutine allows multiple futures to be waited at the same time. Overloads of wait() are provided that take multiple futures as arguments. Up to BOOST_COROUTINE_WAIT_MAX futures can be waited at the same time. wait will return when at least one future has been signaled. See also the rationale for a variable argument wait.

Boost.Coroutine also provides a variable argument wait_all that blocks until all future arguments have been signaled.