uvco::Channel< T > Class Template Reference

#include <channel.h>

Collaboration diagram for uvco::Channel< T >:

Classes
struct	ChannelAwaiter_

Public Member Functions
	Channel (unsigned capacity, unsigned max_waiters=16)
	Create a channel for up to capacity items.
template<typename U>
Promise< void >	put (U value)
Promise< T >	get ()
MultiPromise< T >	getAll ()

Private Member Functions
void	awake_reader ()
void	awake_writer ()

Private Attributes
BoundedQueue< T >	queue_
BoundedQueue< std::coroutine_handle<> >	read_waiting_
BoundedQueue< std::coroutine_handle<> >	write_waiting_

Detailed Description

template<typename T>
class uvco::Channel< T >

A Channel is similar to a Go channel: buffered, and blocking for reading and writing if empty/full respectively.

A bounded-capacity channel for items of type T. A channel can be waited on by at most max_waiters coroutines. If more coroutines want to wait, an exception is thrown.

A reader waits while the channel is empty, and is awoken by the first writer. A writer waits while the channel is full, and is awoken by the first reader.

When only using a channel to communicate small objects between coroutines, it takes about 50 ns to send a small value (e.g. int) on clang-21 and AMD Ryzen 7 PRO 7840U.

Caveat 1: the channel is obviously not thread safe. Only use within one loop.

Caveat 2: Channel is independent of a Loop. This means that runMain() may return despite there being channels in use and awaited on. Ensure that at least one uv operation (socket read/write/listen, timer, etc.) is running to keep the loop alive.

Constructor & Destructor Documentation

Channel()

template<typename T>

uvco::Channel< T >::Channel ( unsigned capacity, unsigned max_waiters = 16 )

inlineexplicit

Create a channel for up to capacity items.

      : queue_{capacity}, read_waiting_{max_waiters},
        write_waiting_{max_waiters} {}

Member Function Documentation

awake_reader()

template<typename T>

void uvco::Channel< T >::awake_reader ( )

inlineprivate

                      {
    while (!read_waiting_.empty()) {
      std::coroutine_handle<void> handle = read_waiting_.get();
      // Skip cancelled coroutines.
      if (handle != nullptr) {
        Loop::enqueue(handle);
        break;
      }
    }
  }

awake_writer()

template<typename T>

void uvco::Channel< T >::awake_writer ( )

inlineprivate

                      {
    while (!write_waiting_.empty()) {
      std::coroutine_handle<void> handle = write_waiting_.get();
      if (handle != nullptr) {
        Loop::enqueue(handle);
        break;
      }
    }
  }

get()

template<typename T>

Promise< T > uvco::Channel< T >::get ( )

inline

Take an item from the channel.

Suspends if no items are available. The suspended coroutine is resumed by the next writer adding an item.

                   {
    if (queue_.empty()) {
      ChannelAwaiter_ awaiter{queue_, read_waiting_};
      BOOST_VERIFY(co_await awaiter);
    }
    T item{queue_.get()};
    // NOTE: this will switch control to the writer until it suspends; keep this
    // in mind.
    awake_writer();
    co_return item;
  }

getAll()

template<typename T>

MultiPromise< T > uvco::Channel< T >::getAll ( )

inline

Continuously read items from channel by repeatedly co_awaiting the returned MultiPromise. (getAll() is a generator)

Remember to call MultiPromise<T>::cancel() when you are done with the channel, although this should happen automatically when the last MultiPromise object is dropped.

                           {
    while (true) {
      if (queue_.empty()) {
        ChannelAwaiter_ awaiter{queue_, read_waiting_};
        BOOST_VERIFY(co_await awaiter);
      }
      T item = queue_.get();
      awake_writer();
      // Suspends until consumer asks for next item.
      co_yield item;
    }
  }

put()

template<typename T>

template<typename U>

Promise< void > uvco::Channel< T >::put ( U value )

inline

Put an item into the channel.

Suspends if no space is available in the channel. The suspended coroutine is resumed by the next reader taking out an item.

Template method: implements perfect forwarding for both copy and move insertion.

The argument is only accepted by value, as it's safer to do so in a coroutine.

                                                   {
    if (!queue_.hasSpace()) {
      // Block until a reader has popped an item.
      ChannelAwaiter_ awaiter{queue_, write_waiting_};
      // Return value indicates if queue is filled with >= 1 item (true) or
      // empty (false).
      co_await awaiter;
      BOOST_VERIFY(queue_.hasSpace());
    }
    queue_.put(std::forward<U>(value));
 
    // NOTE: this will switch control to the reader until it suspends; keep this
    // in mind.
    //
    // For a filled queue, this will result in a nice lock-step switching back
    // and forth.
    awake_reader();
    co_return;
  }

Member Data Documentation

queue_

template<typename T>

BoundedQueue<T> uvco::Channel< T >::queue_

private

read_waiting_

template<typename T>

BoundedQueue<std::coroutine_handle<> > uvco::Channel< T >::read_waiting_

private

write_waiting_

template<typename T>

BoundedQueue<std::coroutine_handle<> > uvco::Channel< T >::write_waiting_

private

---

The documentation for this class was generated from the following file:

• uvco/channel.h