.. _mux_tuple:

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 :literal_ref:`tmc::spawn_tuple()<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 :cpp:func:`get\<I\>()<tmc::mux_tuple::get>`.

After a slot's result has been consumed, you may :cpp:func:`fork\<I\>()<tmc::mux_tuple::fork>` a fresh awaitable into it while the other slots keep running. Unlike :literal_ref:`tmc::select()<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 :literal_ref:`tmc::mux_many<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 :cpp:func:`fork\<I\>()<tmc::mux_tuple::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 :cpp:func:`end()<tmc::mux_tuple::end>` when no slots remain active)
* :cpp:func:`get\<I\>()<tmc::mux_tuple::get>` - access the result of the slot whose index was just returned from ``co_await``
* :cpp:func:`fork\<I\>()<tmc::mux_tuple::fork>` - start (or re-arm) an awaitable in slot ``I``, optionally on a specific executor / priority
* :cpp:func:`fork_clang\<I\>()<tmc::mux_tuple::fork_clang>` - HALO-eligible :cpp:func:`fork\<I\>()<tmc::mux_tuple::fork>` (Clang 20+ only)
* :cpp:func:`is_active\<I\>()<tmc::mux_tuple::is_active>` / :cpp:func:`active_bitset()<tmc::mux_tuple::active_bitset>` - query which slots are currently live
* :cpp:func:`capacity()<tmc::mux_tuple::capacity>` - the number of slots (the count of ``Result`` template arguments); the maximum number of awaitables that may be active at once
* :cpp:func:`end()<tmc::mux_tuple::end>` - the sentinel returned from ``co_await`` once every slot has been drained

Usage Rules
-----------------------------------------------------------------------------------

* ``mux_tuple`` is an :ref:`lvalue-only awaitable<lvalue_awaitables>`. Assign it to a named variable; it cannot be ``co_await``\ ed 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 :cpp:func:`end()<tmc::mux_tuple::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 :cpp:func:`fork\<I\>()<tmc::mux_tuple::fork>` a slot that is empty: one that was never started, or whose previous result has already been returned from ``co_await``. Check with :cpp:func:`is_active\<I\>()<tmc::mux_tuple::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.
* :cpp:func:`fork\<I\>()<tmc::mux_tuple::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
-----------------------------------------------------------------------------------

.. tab:: 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.

   .. code-block:: cpp

      #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
          }
        }
      }

.. tab:: 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 <https://github.com/tzcnt/tmc-examples/blob/main/examples/asio/batch_processor.cpp>`_ example.

   .. code-block:: cpp

      #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
-----------------------------------------------------------------------------------
.. doxygenclass:: tmc::mux_tuple
  :members: