tmc::qu_mpsc_bounded

tmc::qu_mpsc_bounded#

Async MPSC bounded queue using purely zero-copy operation. It is linearizable / FIFO. It can be configured by overriding the Config template parameter.

Producer (unrestricted) operations:

co_await push()
close()
close_resume_inline()

Consumer operations:

co_await pull()
try_pull()
empty()

Usage Examples#

Using pull()

pull() suspends until data is available. It returns a scoped zero-copy handle whose operator bool() / has_value() indicates the result.

#include "tmc/task.hpp"
#include "tmc/spawn.hpp"
#include "tmc/spawn_group.hpp"
#include "tmc/qu_mpsc_bounded.hpp"

tmc::task<void> producer(tmc::qu_mpsc_bounded<int>& q) {
  for (int i = 0; i < 100; ++i) {
    // If the queue is full, this will suspend until a slot becomes free
    co_await q.push(i);
  }
}

tmc::task<void> consumer(tmc::qu_mpsc_bounded<int>& q) {
  // Loop automatically breaks once the queue empties after close()
  while (auto data = co_await q.pull()) {
    // v is a zero-copy reference to a T located in the queue storage
    int& v = data.value();
    // do something with v
  }
}

tmc::task<void> queue_quickstart() {
  // Set queue capacity
  tmc::qu_mpsc_bounded<int> q(10);

  // Start the consumer first
  auto cons = tmc::spawn(consumer(q)).fork();

  // Start multiple producers and wait for them to finish
  auto sg = tmc::spawn_group();
  for (size_t i = 0; i < 10; ++i) {
    sg.add(producer(q));
  }
  co_await std::move(sg);

  // Close the queue and wait for the consumer to drain all data
  q.close();
  co_await std::move(cons);
}

Using try_pull()

try_pull() is non-suspending and must be polled. It returns a scoped zero-copy handle whose status() (or operator bool()) indicates the result.

// Drains all data currently in the queue. Returns true if the queue was empty afterward,
// and false if the queue was closed afterward.
tmc::task<bool> consume_all_data(tmc::qu_mpsc_bounded<int>& q) {
  while(true) {
    auto data = q.try_pull();
    switch (data.status()) {
      case tmc::qu_mpsc_bounded_err::OK: {
        // v is a zero-copy reference to a T located in the queue storage
        int& v = data.value();
        // do something with v
        break;
      }
      case tmc::qu_mpsc_bounded_err::EMPTY:
        // No data available right now. Try again later.
        co_return true;
      case tmc::qu_mpsc_bounded_err::CLOSED:
        // The queue has been closed and drained. Do not try again later.
        co_return false;
      default:
        std::unreachable();
    }
  }
}

API Reference#

template<typename T, typename Config = tmc::qu_mpsc_bounded_default_config> class qu_mpsc_bounded#

Public Functions

inline explicit qu_mpsc_bounded(size_t Capacity) noexcept#: Constructs a qu_mpsc_bounded with the given capacity.

template<typename ...Args> inline aw_push<Args...> push(Args&&... ConstructArgs) noexcept#

Enqueues a new value in the queue by in-place construction, forwarding ConstructArgs to T’s constructor.

Returns an awaitable. If the queue is full, the producer suspends until the consumer reads a value and frees a slot. Otherwise it completes synchronously.

The awaited result is bool: true if the value was enqueued, false if the queue was closed and the value was not enqueued.

If a consumer is currently suspended waiting for a value, it will be resumed once the the value is enqueued.

LIFETIME REQUIREMENT: the returned awaitable holds the arguments by reference (T& for lvalues, T&& for rvalues / temporaries). If you pass a temporary into this, you must co_await it immediately, so the lifetime of the argument can be extended to the end of the full-expression.

// Safe: the temporary T's lifetime is extended to the end of the
// full-expression.
co_await q.push(T{...});

// Unsafe: `a` holds a dangling reference to the temporary T
auto a = q.push(T{...});
co_await std::move(a);

// Safe: passing a reference to a named variable
auto v = T{...};
auto a = q.push(std::move(v));
co_await std::move(a);

inline void close() noexcept#

All future calls to co_await push() will immediately return false. Calls to pull() and try_pull() will continue to read data until all messages have been consumed, at which point all subsequent calls will immediately return an empty scope. If the queue was already empty, any waiting consumers will be awoken immediately and return an empty scope.

Producers that were already suspended in push() waiting for a slot to be freed will be allowed to complete their production normally.

close() is idempotent and safe to call from any thread.

inline void close_resume_inline() noexcept#

Closes the queue and resumes any waiting consumer inline on the caller’s thread instead of posting its continuation to its continuation executor. This should only be used when the caller knows that the waiting consumer may safely run on the caller’s thread.

Behaves like close() in all other respects. close_resume_inline() is idempotent and safe to call from any thread.

inline bool empty()#: Returns true if the queue appears to be empty. A closed-and-drained queue is considered non-empty, and this will return false so that the consumer will call try_pull() / pull() and observe the CLOSED status. This is an unsynchronized read (like try_pull()), so it is only a hint. Only safe to call from the single consumer.

inline aw_pull pull() noexcept#

Await to dequeue. Returns a pull_zc_scope which provides a scoped zero-copy reference to a value in the queue storage. When the scope is destroyed, the referenced value will be destroyed and the queue slot freed for reuse. Only safe to call from the single consumer.

The returned scope’s has_value() / operator bool() returns true if a value was dequeued, or false if the queue was closed and drained.

This scope must be released before the next call to try_pull() or pull(). It must also be released before the queue is destroyed.

May suspend until a value is available, or until close() is called.

inline try_pull_zc_scope try_pull()#

Attempts to immediately dequeue, returning a try_pull_zc_scope which provides a scoped zero-copy reference to a value in the queue storage. When the scope is destroyed, the referenced value will be destroyed and the queue slot freed for reuse. Only safe to call from the single consumer.

The returned scope’s status() returns:

qu_mpsc_bounded_err::OK - a value was dequeued
qu_mpsc_bounded_err::EMPTY - no value is currently available
qu_mpsc_bounded_err::CLOSED - the queue has been closed and drained

The returned scope’s has_value() / operator bool() returns true if a value was dequeued, or false if the queue was empty or closed.

This scope must be released before the next call to try_pull() or pull(). It must also be released before the queue is destroyed.

inline ~qu_mpsc_bounded()#

Destroys the queue and any contained values that have not yet been consumed.

Before destroying this, you must ensure:

No producer is currently calling or suspended in push().
No consumer is calling or suspended in pull() / try_pull().
No pull_zc_scope / try_pull_zc_scope from this queue is alive.
No other thread is calling any other member function.

The recommended teardown sequence is:

Stop submitting new push() calls.
close() the queue.
Drain via pull() / try_pull() until CLOSED. This naturally wakes every pre-close parked producer.
Ensure no further queue method calls will occur (e.g. by joining all producer and consumer coroutines).
Destroy the queue.

class aw_pull : private tmc::detail::AwaitTagNoGroupCoAwait#: Returns a pull_zc_scope when awaited.

template<typename ...Args> class aw_push : private tmc::detail::AwaitTagNoGroupCoAwait#: Returns a bool when awaited: true on successful enqueue, false if the queue was closed.

class pull_zc_scope#

A zero-copy handle to an object in the queue’s storage. The object is exclusively available to this handle. When this handle is destroyed, the queued object will be destroyed and the queue slot will be freed for reuse. Returned by co_await pull().

If the queue has been closed and is drained, pull() will resume with an empty pull_zc_scope (operator bool returns false).

Public Functions

inline pull_zc_scope() noexcept#: Constructs an empty scope. Evaluates to false when converted to bool.

inline explicit operator bool() const noexcept#: Returns true if this scope holds a value from the queue.

inline bool has_value() const noexcept#: Returns true if this scope holds a value from the queue.

inline T &value() noexcept#: Returns a reference to the object in the queue storage. Only valid to call if operator bool() is true.

inline T &operator*() noexcept#: Returns a reference to the object in the queue storage. Only valid to call if operator bool() is true.

inline T *operator->() noexcept#: Returns a pointer to the object in the queue storage. Only valid to call if operator bool() is true.

TMC_FORCE_INLINE inline ~pull_zc_scope()#: Destroys the object in the queue storage and releases the queue slot.

class try_pull_zc_scope#

The status of the pull is exposed via status(): qu_mpsc_bounded_err::OK if a value is held, EMPTY if no value was available, or CLOSED if the queue has been closed and drained.

Public Functions

inline try_pull_zc_scope() noexcept#: Constructs an empty scope (status EMPTY). Evaluates to false when converted to bool.

inline explicit operator bool() const noexcept#: Returns true if this scope holds a value from the queue (status == OK).

inline bool has_value() const noexcept#: Returns true if this scope holds a value from the queue (status == OK).

inline tmc::qu_mpsc_bounded_err status() const noexcept#: Returns the status of this pull: OK, EMPTY, or CLOSED.

inline T &value() noexcept#: Returns a reference to the object in the queue storage. Only valid to call if status() is OK / operator bool() is true.

inline T &operator*() noexcept#: Returns a reference to the object in the queue storage. Only valid to call if status() is OK / operator bool() is true.

inline T *operator->() noexcept#: Returns a pointer to the object in the queue storage. Only valid to call if status() is OK / operator bool() is true.

inline ~try_pull_zc_scope()#: Destroys the object in the queue storage and releases the queue slot.

enum class tmc::qu_mpsc_bounded_err#

Status code returned by qu_mpsc_bounded.try_pull().status()

Values:

enumerator OK#

enumerator EMPTY#

enumerator CLOSED#

struct qu_mpsc_bounded_default_config#

Public Static Attributes

static constexpr bool ConsumerCanSuspend = true#: If true, enables the suspending pull() operation.

static constexpr size_t PackingLevel = 0#: At level 0, queue elements will be padded up to the next increment of TMC_CACHE_LINE_SIZE bytes. This reduces false sharing between neighboring elements. At level 1, no padding will be applied.

tmc::qu_mpsc_bounded

Page Contents

tmc::qu_mpsc_bounded#

Usage Examples#

API Reference#