Unbuffered Channel

Boost.Fiber provides template unbuffered_channel suitable to synchonize fibers (running on same or different threads) via synchronous message passing. A fiber waiting to consume an value will block until the value is produced. If a fiber attempts to send a value through an unbuffered channel and no fiber is waiting to receive the value, the channel will block the sending fiber.

The unbuffered channel acts as an rendezvous point.

typedef boost::fibers::unbuffered_channel< int > channel_t;

void send( channel_t & chan) {
    for ( int i = 0; i < 5; ++i) {
        chan.push( i);
    }
    chan.close();
}

void recv( channel_t & chan) {
    int i;
    while ( boost::fibers::channel_op_status::success == chan.pop(i) ) {
        std::cout << "received " << i << std::endl;
    }
}

channel_t chan{ 1 };
boost::fibers::fiber f1( std::bind( send, std::ref( chan) ) );
boost::fibers::fiber f2( std::bind( recv, std::ref( chan) ) );

f1.join();
f2.join();

Range-for syntax is supported:

typedef boost::fibers::unbuffered_channel< int > channel_t;

void foo( channel_t & chan) {
    chan.push( 1);
    chan.push( 1);
    chan.push( 2);
    chan.push( 3);
    chan.push( 5);
    chan.push( 8);
    chan.push( 12);
    chan.close();
}

void bar( channel_t & chan) {
    for ( unsigned int value : chan) {
        std::cout << value << " ";
    }
    std::cout << std::endl;
}

Template `unbuffered_channel<>`

#include <boost/fiber/unbuffered_channel.hpp>

namespace boost {
namespace fibers {

template< typename T >
class unbuffered_channel {
public:
    typedef T   value_type;

    class iterator;

    unbuffered_channel();

    unbuffered_channel( unbuffered_channel const& other) = delete;
    unbuffered_channel & operator=( unbuffered_channel const& other) = delete;

    void close() noexcept;

    channel_op_status push( value_type const& va);
    channel_op_status push( value_type && va);
    template< typename Rep, typename Period >
    channel_op_status push_wait_for(
        value_type const& va,
        std::chrono::duration< Rep, Period > const& timeout_duration);
    channel_op_status push_wait_for( value_type && va,
        std::chrono::duration< Rep, Period > const& timeout_duration);
    template< typename Clock, typename Duration >
    channel_op_status push_wait_until(
        value_type const& va,
        std::chrono::time_point< Clock, Duration > const& timeout_time);
    template< typename Clock, typename Duration >
    channel_op_status push_wait_until(
        value_type && va,
        std::chrono::time_point< Clock, Duration > const& timeout_time);

    channel_op_status pop( value_type & va);
    value_type value_pop();
    template< typename Rep, typename Period >
    channel_op_status pop_wait_for(
        value_type & va,
        std::chrono::duration< Rep, Period > const& timeout_duration);
    template< typename Clock, typename Duration >
    channel_op_status pop_wait_until(
        value_type & va,
        std::chrono::time_point< Clock, Duration > const& timeout_time);
};

template< typename T >
unbuffered_channel< T >::iterator begin( unbuffered_channel< T > & chan);

template< typename T >
unbuffered_channel< T >::iterator end( unbuffered_channel< T > & chan);

}}

Constructor

unbuffered_channel();

Effects:: The constructor constructs an object of class unbuffered_channel.

Member function `close`()

void close() noexcept;

Effects:: Deactivates the channel. No values can be put after calling this->close(). Fibers blocked in this->pop(), this->pop_wait_for() or this->pop_wait_until() will return closed. Fibers blocked in this->value_pop() will receive an exception.
Throws:: Nothing.
Note:: close() is like closing a pipe. It informs waiting consumers that no more values will arrive.

Member function `push`()

channel_op_status push( value_type const& va);
channel_op_status push( value_type && va);

Effects:: If channel is closed, returns closed. Otherwise enqueues the value in the channel, wakes up a fiber blocked on this->pop(), this->value_pop(), this->pop_wait_for() or this->pop_wait_until() and returns success.
Throws:: Exceptions thrown by copy- or move-operations.

Member function `pop`()

channel_op_status pop( value_type & va);

Effects:: Dequeues a value from the channel. If the channel is empty, the fiber gets suspended until at least one new item is push()ed (return value success and va contains dequeued value) or the channel gets close()d (return value closed).
Throws:: Exceptions thrown by copy- or move-operations.

Member function `value_pop`()

value_type value_pop();

Effects:: Dequeues a value from the channel. If the channel is empty, the fiber gets suspended until at least one new item is push()ed or the channel gets close()d (which throws an exception).
Throws:: fiber_error if *this is closed or by copy- or move-operations.
Error conditions:: std::errc::operation_not_permitted

Member function `pop_wait_for`()

template< typename Rep, typename Period >
channel_op_status pop_wait_for(
    value_type & va,
    std::chrono::duration< Rep, Period > const& timeout_duration)

Effects:: Accepts std::chrono::duration and internally computes a timeout time as (system time + timeout_duration). If channel is not empty, immediately dequeues a value from the channel. Otherwise the fiber gets suspended until at least one new item is push()ed (return value success and va contains dequeued value), or the channel gets close()d (return value closed), or the system time reaches the computed timeout time (return value timeout).
Throws:: timeout-related exceptions or by copy- or move-operations.

Member function `pop_wait_until`()

template< typename Clock, typename Duration >
channel_op_status pop_wait_until(
    value_type & va,
    std::chrono::time_point< Clock, Duration > const& timeout_time)

Effects:: Accepts a std::chrono::time_point< Clock, Duration >. If channel is not empty, immediately dequeues a value from the channel. Otherwise the fiber gets suspended until at least one new item is push()ed (return value success and va contains dequeued value), or the channel gets close()d (return value closed), or the system time reaches the passed time_point (return value timeout).
Throws:: timeout-related exceptions or by copy- or move-operations.

Non-member function `begin( unbuffered_channel< T > &)`

template< typename T >
unbuffered_channel< T >::iterator begin( unbuffered_channel< T > &);

Returns:: Returns a range-iterator (input-iterator).

Non-member function `end( unbuffered_channel< T > &)`

template< typename T >
unbuffered_channel< R >::iterator end( unbuffered_channel< T > &);

Returns:: Returns an end range-iterator (input-iterator).