tmc::qu_spsc_unbounded

tmc::qu_spsc_unbounded#

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

Producer operations:

post()
post_bulk()
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_tuple.hpp"
#include "tmc/qu_spsc_unbounded.hpp"

tmc::task<void> producer(tmc::qu_spsc_unbounded<int>& q) {
  for (int i = 0; i < 100; ++i) {
    q.post(i);
  }
  q.close();
}

tmc::task<void> consumer(tmc::qu_spsc_unbounded<int>& q) {
  // Loop automatically breaks once the queue drains 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() {
  tmc::qu_spsc_unbounded<int> q;
  co_await tmc::spawn_tuple(producer(q), consumer(q));
}

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_spsc_unbounded<int>& q) {
  while(true) {
    auto data = q.try_pull();
    switch (data.status()) {
      case tmc::qu_spsc_unbounded_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_spsc_unbounded_err::EMPTY:
        // No data available right now. Try again later.
        co_return true;
      case tmc::qu_spsc_unbounded_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_spsc_unbounded_default_config> class qu_spsc_unbounded#

Public Functions

template<typename ...Args> inline void post(Args&&... ConstructArgs) noexcept#

Enqueues a new value in the queue by in-place construction, forwarding ConstructArgs to T’s constructor. Only safe to call from the single producer.

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

You must not call this after calling close().

template<typename It> inline void post_bulk(It &&Items, size_t Count) noexcept#

Moves Count values from the iterator Items into the queue. Only safe to call from the single producer.

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

You must not call this after calling close().

template<typename It> inline void post_bulk(It &&Begin, It &&End) noexcept#

Calculates the number of elements via size_t Count = End - Begin; and moves them from the iterator Begin into the queue. Only safe to call from the single producer.

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

You must not call this after calling close().

template<typename Range> inline void post_bulk(Range &&R) noexcept#

Calculates the number of elements via size_t Count = Range.end() - Range.begin(); and moves them from the beginning of the range into the queue. Only safe to call from the single producer.

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

You must not call this after calling close().

inline void close() noexcept#

Closes the queue. May only be called from the single producer. After close() returns, the producer must not call post() or post_bulk() again. 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.

close() is idempotent.

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. May only be called from the single producer.

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_spsc_unbounded_err::OK - a value was dequeued
qu_spsc_unbounded_err::EMPTY - no value is currently available
qu_spsc_unbounded_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_spsc_unbounded()#

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

Before destroying this, you must ensure:

No producer is currently calling post() or post_bulk().
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 post() calls.
close() the queue.
Drain via pull() / try_pull() until CLOSED.
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.

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_spsc_unbounded_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_spsc_unbounded_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_spsc_unbounded_err#

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

Values:

enumerator OK#

enumerator EMPTY#

enumerator CLOSED#

struct qu_spsc_unbounded_default_config#

Public Static Attributes

static constexpr bool ConsumerCanSuspend = true#: If true, enables the suspending pull() operation. This costs the producer an additional locked operation to check for a waiting consumer.

static constexpr size_t BlockSize = 4096#: The number of elements that can be stored in each block in the qu_spsc_unbounded linked list.

static constexpr size_t PackingLevel = 1#: At level 0, queue elements will be padded up to the next increment of 64 bytes. This reduces false sharing between neighboring elements. At level 1, no padding will be applied. The SPSC queue is packed by default to improve cache coherency for the single producer.

static constexpr bool EmbedFirstBlock = false#: If true, the first storage block will be a member of the qu_spsc_unbounded object (instead of dynamically allocated). Subsequent storage blocks are always dynamically allocated.

tmc::qu_spsc_unbounded

Page Contents

tmc::qu_spsc_unbounded#

Usage Examples#

API Reference#