tmc::mux_tuple#

A reusable, heterogeneous result-multiplexer over a fixed set of awaitable slots. Each slot may produce a different result type; the mux is templated on the result types, not the awaitable types (a void result becomes std::monostate).

Rather than delivering every result at once (like tmc::spawn_tuple()), 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 get().

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 hold a background wait open while doing other work, join heterogeneous streams, or maintain conditional concurrency.

Use tmc::mux_many instead when every slot produces the same result type.

Limited to 63 slots (31 on 32-bit). There are two ways to construct one:

With awaitables - the result types are deduced via CTAD and every awaitable is forked eagerly: tmc::mux_tuple mux(task0(), task1());
Empty - only the result-type template arguments are given; no awaitables are initiated. Start slots individually with fork(): tmc::mux_tuple<int, std::string> mux;

Member functions:

co_await the mux - suspends until one slot is ready, then returns its index (or end() when no slots remain active)
get() - access the result of the slot whose index was just returned from co_await
fork() - start (or re-arm) an awaitable in slot I, 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 of Result template arguments); 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_tuple 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 slot it’s replacing.
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#

Background signal

Aggregate a high-rate data stream while keeping a wait open on a low-rate control channel - reacting to commands without blocking the data path, and vice versa. The two channels carry different types, so the slots are heterogeneous.

#include "tmc/task.hpp"
#include "tmc/mux_tuple.hpp"
#include "tmc/qu_mpsc_unbounded.hpp"
#include "tmc/qu_spsc_unbounded.hpp"

struct Measurement { double v; };
enum class Command { Recalibrate, Report };

struct Stats; // exposition only

tmc::task<void> monitor(
  tmc::qu_mpsc_unbounded<Measurement>& data,
  tmc::qu_spsc_unbounded<Command>& control
) {
  // slot 0: the data stream (the "work")
  // slot 1: the background control signal
  tmc::mux_tuple<
    tmc::qu_mpsc_unbounded<Measurement>::pull_zc_scope,
    tmc::qu_spsc_unbounded<Command>::pull_zc_scope>
    mux;
  mux.fork<0>(data.pull());
  mux.fork<1>(control.pull());

  Stats stats;
  while (true) {
    switch (co_await mux) {
    case 0:
      if (auto& m = mux.get<0>()) {
        stats.add(m.value().v);
        mux.fork<0>(data.pull()); // keep pulling data
      }
      // else: data channel closed - stop pulling from it.
      break;
    case 1:
      if (auto& cmd = mux.get<1>()) {
        switch (cmd.value()) {
        case Command::Recalibrate: stats.reset(); break;
        case Command::Report:      stats.publish(); break;
        }
        mux.fork<1>(control.pull()); // keep watching the control channel
      }
      // else: control channel closed - stop watching it.
      break;
    default:
      co_return; // both channels closed and drained
    }
  }
}

Timeout batch processor

A timeout-based batch processor: collect items from a queue, start a timer once the first item of a batch arrives, and flush the batch when the timer fires. The queue pull (slot 0) and the timer (slot 1) have different result types, and the timer slot is only conditionally active - a natural fit for mux_tuple. This is condensed from the runnable batch_processor.cpp example.

#include "tmc/asio/aw_asio.hpp"
#include "tmc/asio/ex_asio.hpp"
#include "tmc/mux_tuple.hpp"
#include "tmc/qu_mpsc_unbounded.hpp"
#include "tmc/spawn.hpp"
#include "tmc/task.hpp"

#include <asio/error_code.hpp>
#include <asio/steady_timer.hpp>
#include <chrono>
#include <cstdio>
#include <vector>

// For exposition only
struct data {};
static void process(std::vector<data> batch) {
  std::printf("Processed a batch of size %zu\n", batch.size());
}
constexpr auto duration = std::chrono::milliseconds(100);

static tmc::task<void> timedBatchProcessor(tmc::qu_mpsc_unbounded<data>& q) {
  // slot 0: a pull from the queue - starts active
  // slot 1: a timeout - starts inactive
  // Once the first item arrives, start the timeout and keep collecting until
  // it fires; then process the batch and restart the timer on the next element.
  tmc::mux_tuple<
    tmc::qu_mpsc_unbounded<data>::pull_zc_scope, std::tuple<asio::error_code>>
    mux;
  mux.fork<0>(q.pull());

  asio::steady_timer timer{tmc::asio_executor()};
  std::vector<data> batch;

  while (true) {
    switch (co_await mux) {
    case 0:
      if (auto& elem = mux.get<0>()) {
        // Grab the data from the queue and restart the pull awaitable.
        batch.push_back(std::move(*elem));
        mux.fork<0>(q.pull());

        if (!mux.is_active<1>()) {
          // This is the first element of the batch; (re)start the timer.
          timer.expires_after(duration);
          mux.fork<1>(timer.async_wait(tmc::aw_asio));
        }
      } else {
        // Queue was closed. Cancel the timer if it's active.
        timer.cancel();
      }
      break;
    case 1:
      // Timer fired, either due to timeout or cancellation. Either way,
      // process the batch. The timer slot's bit was already cleared by
      // co_await, so is_active<1>() is now false and the next queue element
      // will restart the timer.
      process(std::move(batch));
      batch.clear();
      break;
    default:
      // idx == mux.end(); both awaitables are done and we can safely return.
      co_return;
    }
  }
}

int main() {
  tmc::asio_executor().init();
  return tmc::async_main([]() -> tmc::task<int> {
    tmc::qu_mpsc_unbounded<data> q;
    auto t = tmc::spawn(timedBatchProcessor(q)).fork();

    q.post(data{});
    q.post(data{});
    q.post(data{});
    q.close();

    co_await std::move(t);
    co_return 0;
  }());
}

API Reference#

template<typename ...Result> class mux_tuple : 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.
Templated on result types, not awaitable types. void becomes std::monostate.
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.
Limited to 63 (or 31, on 32-bit) slots.

After a slot’s result has been consumed by co_await, you may call fork() to launch a fresh awaitable into that slot. The replacement awaitable must produce the same result type as the 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 constructors: one taking all the awaitables, and deducing the result types, and the other taking no awaitables, which requires you to specify the result type for each slot:

// Result types deduced to <int, data>. Both awaitables are forked eagerly.
mux_tuple mux0(int_awaitable, data_task);

// Result types explicitly specified. Individual awaitables can be forked afterward.
mux_tuple<int, data> mux1;

Public Functions

inline mux_tuple()#: Creates the result storage but does not initiate any awaitables. The Result template types must be provided explicitly. Use fork() to initiate work into individual slots.

template<typename ...Awaitable> inline mux_tuple(Awaitable&&... Awaitables)#: Eagerly initiates all of the provided awaitables. The Result template types are deduced from the awaitable arguments.

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. 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_tuple &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 number of Result template arguments. This is the maximum number of awaitables that may be active concurrently.

template<size_t I> TMC_TSAN_NO_SPECULATE inline std::tuple_element_t<I, ValueTuple> &get() noexcept#: Gets the ready result at the given index.

template<size_t I> inline bool is_active() const noexcept#: Returns true if slot I 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.

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

template<size_t I, typename Aw, typename Exec = tmc::ex_any*> inline void fork(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 the empty constructor), or its previous result has already been consumed by co_await. The replacement awaitable must produce the same result type as the slot’s declared awaitable.

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<size_t I, typename Aw, typename Exec = tmc::ex_any*> inline mux_tuple_fork_clang fork_clang(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_tuple mux(task(0), task(1));
for (size_t i = co_await mux; i != mux.end(); i = co_await mux) {
  switch (i) {
  case 0:
    process(mux.get<0>());
    co_await mux.fork_clang<0>(task(0));
    break;
  case 1:
    process(mux.get<1>());
    co_await mux.fork_clang<1>(task(1));
    break;
  default:
    std::unreachable();
  }
}

class mux_tuple_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_tuple

Page Contents

tmc::mux_tuple#

Usage Rules#

Usage Examples#

API Reference#