tmc::mux_many

tmc::mux_many#

A reusable, homogeneous result-multiplexer over a fixed set of awaitable slots. All slots produce the same Result type.

Rather than delivering every result at once (like tmc::spawn_many()), each result is delivered as it becomes ready: co_await-ing the mux suspends until one slot completes, then returns that slot’s index. You read the value with operator[].

After a slot’s result has been consumed, you may fork() a fresh awaitable into it while the other slots keep running. Unlike tmc::select(), the still-running awaitables are never cancelled when a peer completes. This lets you keep a constant level of concurrency, join several streams, or hold a background wait open while doing other work.

Use tmc::mux_tuple instead when the slots have different result types.

Storage for up to 63 slots (31 on 32-bit) is held in a fixed-size std::array. Both template arguments - Result and Count - can be named explicitly to right-size the storage. If Count is not specified, it defaults to the maximum (63 or 31).

Member functions:

co_await the mux - suspends until one slot is ready, then returns its index (or end() when no slots remain active)
operator[] - access the result of the slot whose index was just returned from co_await
fork() - start (or re-arm) an awaitable in a slot, optionally on a specific executor / priority
fork_clang() - HALO-eligible fork() (Clang 20+ only)
is_active() / active_bitset() - query which slots are currently live
capacity() - the number of slots (the Count template argument); the maximum number of awaitables that may be active at once
end() - the sentinel returned from co_await once every slot has been drained

Usage Rules#

mux_many is an lvalue-only awaitable. Assign it to a named variable; it cannot be co_awaited as a temporary.
You may co_await it repeatedly. Each active slot’s index is returned exactly once per submission.
You must drain it - co_await until it returns end() - before it is destroyed. Forked awaitables hold a pointer back into the mux, so destroying it while a slot is still active is a use-after-free.
Only fork() a slot that is empty: one that was never started, or whose previous result has already been returned from co_await. Check with is_active() if unsure.
A replacement awaitable must produce the same Result type as the group.
Awaiting a partially-filled mux is OK; not every slot must be active.
Re-activating a completely empty mux (co_await returned end()) is also OK.
fork() and co_await are not thread-safe to call concurrently with each other or with themselves; drive the mux from a single coroutine.

Usage Examples#

Fixed concurrency

Keep exactly Width operations in flight over a large workload: fill the slots up front, then re-arm each slot with the next job as its result is consumed.

#include "tmc/task.hpp"
#include "tmc/mux_many.hpp"

#include <vector>

// An async operation we want to run over many inputs. Because `job` is a
// by-value parameter, this coroutine is safe to defer into a mux slot.
tmc::task<int> process(int job);

// Runs every job, but keeps at most `Width` of them running at once.
template <size_t Width>
tmc::task<void> run_bounded(std::vector<int> jobs) {
  // Storage for `Width` concurrent slots. Start empty, then fill it.
  tmc::mux_many<int, Width> mux;

  size_t next = 0;
  while (next < jobs.size() && next < mux.capacity()) {
    mux.fork(next, process(jobs[next]));
    ++next;
  }

  // As each slot completes, consume its result and immediately re-arm that
  // slot with the next job - keeping the pipeline full until the jobs run out.
  for (size_t i = co_await mux; i != mux.end(); i = co_await mux) {
    int result = mux[i];
    // ... use result ...

    if (next < jobs.size()) {
      mux.fork(i, process(jobs[next]));
      ++next;
    }
    // Otherwise leave slot `i` empty; the group drains as the last jobs finish.
  }
}

Join multiple queues

Consume from several queues with same result types at once, handling items in whatever order they arrive. Each slot holds one queue’s pull(); re-arm it after each item, and leave it empty once its queue closes.

#include "tmc/task.hpp"
#include "tmc/mux_many.hpp"
#include "tmc/qu_spsc_unbounded.hpp"

#include <span>

// Merges several queues into a single processing loop. Ends once every
// queue has been closed and drained.
tmc::task<void> merge_queues(std::span<tmc::qu_spsc_unbounded<int>> queues) {
  using scope = tmc::qu_spsc_unbounded<int>::pull_zc_scope;

  // One slot per queue; fork a pull() from each.
  tmc::mux_many<scope, 16> mux;
  for (size_t i = 0; i < queues.size(); ++i) {
    mux.fork(i, queues[i].pull());
  }

  for (size_t i = co_await mux; i != mux.end(); i = co_await mux) {
    if (auto& data = mux[i]) {
      // v is a zero-copy reference into queue `i`'s storage.
      int& v = data.value();
      // ... process v ...

      // This queue is still open; re-arm its pull.
      mux.fork(i, queues[i].pull());
    }
    // Otherwise queue `i` was closed & drained; leave its slot empty.
  }
}

Breaking the 63-slot cap

A single mux is capped at 63 slots (31 on 32-bit). To run more awaitables concurrently while still receiving each result as it becomes ready, nest muxes. Hold the real work in an array of inner muxes, then fork each inner mux - itself an lvalue awaitable - into a single outer mux. This increases the capacity to 63 x 63 = 3969 awaitables at once.

Awaiting a mux yields the size_t index of a ready slot, so the outer mux’s result type is size_t. Each outer result tells you which inner mux is ready, and outer[outerIdx] tells you which slot of that mux it is. Awaiting an inner mux through the outer is one-shot, so after each await, re-arm both halves: re-fork work into the inner mux, and re-fork the inner mux into the outer slot (whose previous await was just consumed).

#include "tmc/mux_many.hpp"
#include "tmc/task.hpp"

#include <array>

// The per-mux slot cap: 63 on 64-bit, 31 on 32-bit platforms.
constexpr size_t Cap = TMC_PLATFORM_BITS - 1;

// The real work. Taking `input` by value keeps this coroutine safe to defer.
tmc::task<int> work(int input) { co_return input * 2; }

int compute_index(size_t outerIdx, size_t innerIdx) {
  return static_cast<int>(outerIdx * Cap + innerIdx);
}

// Runs up to Cap * Cap awaitables concurrently, handling each result as soon as it is ready.
tmc::task<void> process_all() {
  // Inner muxes hold the actual work: Cap muxes of Cap slots each.
  // Total capacity: Cap * Cap = 3969 on 64-bit platforms.
  std::array<tmc::mux_many<int, Cap>, Cap> innerArr;
  // The outer mux drives the inner muxes. co_await-ing a mux yields a
  // size_t slot index, so the outer mux's result type is size_t.
  tmc::mux_many<size_t, Cap> outer;

  for (size_t i = 0; i < Cap; ++i) {
    for (size_t j = 0; j < Cap; ++j) {
      innerArr[i].fork(j, work(compute_index(i, j)));
    }
    outer.fork(i, innerArr[i]);
  }

  for (size_t outerIdx = co_await outer; outerIdx != outer.end();
       outerIdx = co_await outer) {
    // mux_many<int> is the same type as mux_many<int, Cap>, since Cap is
    // the default Count.
    tmc::mux_many<int>& inner = innerArr[outerIdx]; // the inner mux that is ready
    size_t innerIdx = outer[outerIdx];              // its slot that is ready

    if (innerIdx != inner.end()) {
      int resultIdx = compute_index(outerIdx, innerIdx); // overall awaitable index, for reference
      int result = inner[innerIdx];
      // ... use result ...

      // Re-arm both halves: refill the inner mux's slot with new work, and
      // re-subscribe the outer mux (its one-shot await was just consumed).
      inner.fork(innerIdx, work(compute_index(outerIdx, innerIdx)));
      outer.fork(outerIdx, inner);
    }
    // Otherwise inner mux `outerIdx` is fully drained; leave the outer slot empty.
  }
}

API Reference#

template<typename Result, size_t Count = TMC_PLATFORM_BITS - 1> class mux_many : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue#

A reusable result-multiplexer over a fixed set of awaitable slots.

Rather than returning all results at once, each result is made available as it becomes ready: co_await on the mux returns the index of a single ready slot.
All results must be of the same type.
Passing the awaitables up front is optional. You can construct this empty and fork() awaitables into it afterward. It is valid to await this while only a partial set of the slots are active.
Each fork()ed awaitable can be dispatched to a separate executor/priority.
You must ensure that all awaitables are completed (co_await mux == mux.end()) before destroying this.
Results are stored internally in a fixed-size std::array, limited to 63 (or 31, on 32-bit) slots. You can dispatch fewer awaitables than this (not every slot must be filled). If no size is specified, the default maximum of 63 or 31 will be used.

After a slot’s result has been consumed by co_await, you may call fork(i) to launch a fresh awaitable into that slot. This allows you to maintain a fixed level of concurrency (by always replacing a completed slot) or partial / conditional concurrency (see the batch_processor.cpp example).

There are two families of construction:

With awaitables (an iterator + count, a begin/end range, a begin/end range

max count, or a range object). The Result type is deduced from the awaitables via CTAD, and Count is left at its default. The awaitables are forked eagerly.
```
auto mux = tmc::mux_many(awaitables.begin(), awaitables.end());
for (size_t i = co_await mux; i != mux.end(); i = co_await mux) {
  use(mux[i]);
}
```

To right-size the std::array storage, name both template arguments explicitly:

auto mux = tmc::mux_many<int, N>(awaitables.begin());                   // exactly N
auto mux = tmc::mux_many<int, N>(awaitables.begin(), awaitables.end()); // up to N

// Allocate N storage slots. Eagerly dispatch awaitables equal to the smallest of:
// - awaitables.end() - awaitables.begin() (total awaitable count)
// - maxCount
// - N
auto mux = tmc::mux_many<int, N>(awaitables.begin(), awaitables.end(), maxCount);

If constructed with a partial workload (End - Begin < Count or MaxCount < Count) the initiated awaitables will be at the low indexes, and higher indexes will be empty and ready to fork() into immediately.

Empty (the Result type must be provided explicitly; Count defaults if not provided). This only allocates the result storage; no awaitables are initiated. You start individual slots with fork() whenever you like, optionally choosing a specific executor and priority for each one:
```
auto mux = tmc::mux_many<int, 2>(); // storage for 2 slots
auto mux = tmc::mux_many<int>();    // storage for TMC_PLATFORM_BITS - 1 slots
mux.fork(0, make_int_task());
mux.fork(1, make_int_task());
for (size_t i = co_await mux; i != mux.end(); i = co_await mux) { ... }
```

Public Functions

inline mux_many()#: Creates the result storage but does not initiate any awaitables. The Result type must be provided explicitly; Count is optional and defaults to TMC_PLATFORM_BITS - 1, e.g. tmc::mux_many<Result, Count>(). Use fork() to initiate work into individual slots after construction.

template<typename AwaitableIter> inline mux_many(AwaitableIter &&Iter, size_t AwaitableCount)#

Eagerly initiates awaitables from Iter, stopping at the earlier of:

initiatedCount == AwaitableCount
initiatedCount == Count

The Result type is deduced via CTAD; name both template arguments explicitly to right-size the fixed std::array storage.

template<typename AwaitableIter> inline mux_many(AwaitableIter &&Begin, AwaitableIter &&End, size_t MaxCount)#

Eagerly initiates awaitables from Begin, stopping at the earlier of:

iter == End
initiatedCount == MaxCount
initiatedCount == Count

The Result type is deduced via CTAD; name both template arguments explicitly to right-size the fixed std::array storage.

template<typename AwaitableIter> inline mux_many(AwaitableIter &&Begin)#: Eagerly initiates exactly Count awaitables from [Begin, Begin + Count). Name both template arguments explicitly, e.g. tmc::mux_many<Result, Count>(Begin).

template<typename AwaitableIter> inline mux_many(AwaitableIter &&Begin, AwaitableIter &&End)#

Eagerly initiates awaitables from Begin, stopping at the earlier of:

iter == End
initiatedCount == Count

The Result type is deduced via CTAD; name both template arguments explicitly to right-size the fixed std::array storage.

template<typename AwaitableRange> inline mux_many(AwaitableRange &&Range)#

Eagerly initiates awaitables from Range.begin(), stopping at the earlier of:

iter == Range.end()
initiatedCount == Count

The Result type is deduced via CTAD; name both template arguments explicitly to right-size the fixed std::array storage.

inline bool await_ready() const noexcept#: Always suspends.

inline bool await_suspend(std::coroutine_handle<> Outer) noexcept#: Suspends the outer coroutine until a result becomes ready (or resumes immediately if one is already ready).

inline size_t await_resume() noexcept#: Returns the index of a single ready slot. The result indexes correspond to the indexes of the originally submitted awaitables, and the values can be accessed using operator[]. Results may become ready in any order, but each consumed (or forked) slot index will be returned exactly once per submission. When no submitted results remain, the index returned will be equal to the value of end().

inline mux_many &operator co_await() & noexcept#: This type must be awaited as an lvalue (it is awaited repeatedly and is not movable). The awaiter is the group itself.

inline constexpr size_t end() const noexcept#: Provides a sentinel value that can be compared against the value returned from co_await.

inline constexpr size_t capacity() const noexcept#: Returns the capacity of the mux, equal to the Count template argument. This is the maximum number of awaitables that may be active concurrently.

TMC_TSAN_NO_SPECULATE inline std::add_lvalue_reference_t<Result> operator[](size_t Idx) noexcept#: Gets the ready result at the given index.

inline void operator[](size_t Idx) noexcept: Provided for convenience only - to expose the same API as the non-void version. Does nothing.

inline bool is_active(size_t Idx) const noexcept#: Returns true if slot idx is currently active: it has been forked but its result has not yet been returned from co_await. An active slot may not be re-forked. This is the negation of the precondition checked by fork().

inline size_t active_bitset() const noexcept#: Returns the raw bitmap of active slots: bit idx is set if slot idx has been forked but its result has not yet been returned from co_await.

template<typename Aw, typename Exec = tmc::ex_any*> inline void fork(size_t Idx, Aw &&Awaitable, Exec &&Executor = tmc::current_executor(), size_t Priority = tmc::current_priority())#

Starts a new awaitable in the given slot and initiates it immediately on the specified executor and priority. The slot must be empty - either it was never started (when constructed with an empty constructor), or its previous result has already been consumed by co_await. The replacement awaitable must produce the same Result type as the group.

fork() destroys the previous result in the given slot before initiating the replacement. If you need to use the result after this, you should move it out before calling fork().

Executor defaults to the current executor. Priority defaults to the current priority.

This method is not thread-safe to call concurrently with itself or co_await.

template<typename Aw, typename Exec = tmc::ex_any*> inline mux_many_fork_clang fork_clang(size_t Idx, Aw &&Awaitable, Exec &&Executor = tmc::current_executor(), size_t Priority = tmc::current_priority())#

Similar to fork() but allows the forked awaitable’s allocation to be elided by combining it into the parent’s allocation (HALO). This works by using specific attributes that are only available on Clang 20+. You can safely call this function on other compilers, but no HALO-specific optimizations will be applied.

This method is not thread-safe to call concurrently with itself or co_await.

WARNING: You may safely call this in a loop only if you use a switch statement so that each index has a unique call site. This ensures Clang will create independent awaitable storage for each slot (keyed to the call site).

IMPORTANT: This returns a dummy awaitable. For HALO to work, you should not store the dummy awaitable. Instead, co_await this expression immediately.

Proper usage, taking both of the above into account:

tmc::mux_many<int, 2> mux(tasks.begin(), tasks.end());
for (size_t i = co_await mux; i != mux.end(); i = co_await mux) {
  switch (i) {
  case 0:
    process(mux[0]);
    co_await mux.fork_clang(0, task(0));
    break;
  case 1:
    process(mux[1]);
    co_await mux.fork_clang(1, task(1));
    break;
  default:
    std::unreachable();
  }
}

class mux_many_fork_clang : private tmc::detail::AwaitTagNoGroupAsIs#

This is a dummy awaitable. Don’t store this in a variable. For HALO to work, you must co_await mux.fork_clang() immediately.

Public Functions

inline bool await_ready() const noexcept#: Never suspends.

inline void await_suspend(std::coroutine_handle<>) noexcept#: Does nothing.

inline void await_resume() noexcept#: Does nothing.

tmc::mux_many

Page Contents

tmc::mux_many#

Usage Rules#

Usage Examples#

API Reference#