.. _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 :literal_ref:`tmc::spawn_many()<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 :cpp:func:`fork()<tmc::mux_many::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 keep a constant level of concurrency, join several streams, or hold a background wait open while doing other work.

Use :literal_ref:`tmc::mux_tuple<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 :cpp:func:`end()<tmc::mux_many::end>` when no slots remain active)
* ``operator[]`` - access the result of the slot whose index was just returned from ``co_await``
* :cpp:func:`fork()<tmc::mux_many::fork>` - start (or re-arm) an awaitable in a slot, optionally on a specific executor / priority
* :cpp:func:`fork_clang()<tmc::mux_many::fork_clang>` - HALO-eligible :cpp:func:`fork()<tmc::mux_many::fork>` (Clang 20+ only)
* :cpp:func:`is_active()<tmc::mux_many::is_active>` / :cpp:func:`active_bitset()<tmc::mux_many::active_bitset>` - query which slots are currently live
* :cpp:func:`capacity()<tmc::mux_many::capacity>` - the number of slots (the ``Count`` template argument); the maximum number of awaitables that may be active at once
* :cpp:func:`end()<tmc::mux_many::end>` - the sentinel returned from ``co_await`` once every slot has been drained

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

* ``mux_many`` 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_many::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()<tmc::mux_many::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()<tmc::mux_many::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.
* :cpp:func:`fork()<tmc::mux_many::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:: 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.

   .. code-block:: cpp

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

.. tab:: 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 :cpp:func:`pull()<tmc::qu_spsc_unbounded::pull>`; re-arm it after each item, and leave it empty once its queue closes.

   .. code-block:: cpp

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

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

   .. code-block:: cpp

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