feat: add tmc::mux_many and tmc::mux_tuple

AGENTS.md

@@ -0,0 +1,130 @@
+# TooManyCooks — agent guide
+
+TooManyCooks (TMC) is a header-only C++20 coroutine runtime and concurrency
+library. Everything lives in the `tmc::` namespace. Include `"tmc/all_headers.hpp"` for the whole library, or the individual headers you need (e.g. `"tmc/spawn.hpp"`).
+
+## Core Concepts
+- `tmc::task` is the library coroutine type. It is a cold coroutine (initial_suspend = suspend_never).
+- Tasks run on executors and have a priority level. The current executor/priority are tracked via thread-local variables.
+- Child / spawned tasks inherit their parents executor/priority by default. If the child's executor/priority are customized, those customizations are transitive to the child's children. The inheritance flows downward only — parents are not affected by changes to their children's executor/priority. In general, whenever a task awaits an awaitable, it always resumes back on its original executor/priority afterward unless explicitly customized with `resume_on()`.
+
+## Awaitable Usage Rules
+- Most `tmc::` awaitables are single-use and must be `co_await`ed as a temporary or explicitly cast to rvalue before `co_await`ing. Awaitables that are designed to be awaited multiple times (e.g. `tmc::mutex`) must be awaited as an lvalue. Using the wrong value category for an awaitable will cause a compilation error.
+- Rvalue awaitables (including `tmc::task`) are linear types and *must* be awaited if not explicitly `detach()`ed. Constructing an awaitable that is not awaited will cause a leak.
+- Forked awaitables hold a pointer back to the awaitable object returned by `fork()`. They must be joined by `co_await` on that object before it goes out of scope. Failure to do so will cause a use-after-free. Usually this means they must be awaited within within the coroutine where they were created.  An escape hatch for this is to pass a persistent awaitable object (`fork_group` or `mux`) with an external lifetime by reference into a coroutine. You can `fork()` into that object and then return, as long as it is eventually awaited before it goes out of scope. You can also use this to `fork()` tasks from a function that is not a coroutine - only the eventual awaiter must be a coroutine.
+
+## Footgun: capturing coroutine lambdas
+
+**Rule: a lambda whose body is a coroutine (uses `co_await`/`co_return`) must not
+capture anything it needs at run time unless the closure object is guaranteed to
+outlive the task.**
+
+A capturing lambda that is itself a coroutine stores its captures in the *closure
+object*. The coroutine frame keeps only a `this` pointer back to that closure. If
+the closure dies before the task runs, the task reads freed memory — a
+use-after-free that often presents as a flaky hang, garbage result, or assert.
+
+Invoking such a lambda produces a **temporary closure** whose lifetime ends at the
+end of the full-expression. So whether it is safe depends entirely on *when the
+task runs relative to that closure temporary*.
+
+### Safe
+
+```cpp
+int x = 42;
+
+// 1. Invoke AND await in the same full-expression. The closure temporary lives
+//    to the end of the statement, which covers the whole await.
+co_await tmc::spawn([&x]() -> tmc::task<int> { co_return x; }());
+
+// 2. Non-capturing coroutine; pass references as PARAMETERS. Parameters are
+//    stored in the coroutine frame, so this is safe even when deferred.
+auto t = [](int& x) -> tmc::task<int> { co_return x; }(x);
+// ... arbitrary code ...
+co_await tmc::spawn(std::move(t));                 // OK
+
+// 3. Name the closure as an lvalue first, so it outlives the produced task.
+auto fn = [&x]() -> tmc::task<int> { co_return x; };
+auto t2 = fn();
+co_await tmc::spawn(std::move(t2));                // OK — fn still alive
+```
+
+### Unsafe
+
+```cpp
+int x = 42;
+
+// The closure is a TEMPORARY, destroyed at the `;`. The task is deferred.
+auto t = [&x]() -> tmc::task<int> { co_return x; }();
+// ... `t`'s closure is already gone here ...
+co_await tmc::spawn(std::move(t));                 // UAF: reads x via dangling `this`
+```
+
+The same trap applies whenever the produced task is **deferred** — stored in a
+variable, `std::array`/`std::vector`, or built into a `spawn_many` / `spawn_tuple`
+/ `mux_*` group that is awaited/`fork()`ed in a *later* statement. Prefer form 2
+(pass refs as parameters) for anything that isn't awaited on the spot.
+
+## Awaitable Wrappers
+
+These can all wrap one or more awaitables to provide custom dispatch behavior. They can all be customized before initiation (see next section).
+
+| Awaitable | Use when |
+|---|---|
+| `spawn` | Customize a single awaitable. |
+| `spawn_many` | Run N awaitables of the **same** type; results in a `std::array`/`std::vector`. |
+| `spawn_tuple` | Run several awaitables of **different** types; results in a `std::tuple`. |
+| `spawn_func` | Wrap a plain function/functor (not a coroutine) as a task. |
+| `spawn_func_many` | Run N plain functors. |
+| `spawn_group` | Imperatively build a group, then initiate all on `co_await` (lazy). Movable — can be returned/passed around. |
+| `fork_group` | Imperatively build a group, initiating **each awaitable immediately** (eager); join later. Not movable. |
+| `mux_many` | Homogeneous multiplexer: results delivered **as each becomes ready**; `co_await` returns the index of one ready slot. Slots are reusable - you can re-arm a new awaitable into a slot after its result has been consumed. Up to 63 slots (31 on 32-bit). |
+| `mux_tuple` | Heterogeneous multiplexer: same as `mux_many` but slots may have different result types. |
+| `select` | Await several awaitables, return the result of the **first** to complete, and cancel the rest. |
+
+`spawn_many` / `spawn_func_many` take iterators of tasks. A simple, lightweight tool to transform a sequence into an iterator these will accept is `tmc::iter_adapter` from `tmc/utils.hpp`. For more complex transformations, use the <ranges> library - but if it is not used elsewhere in the project, you should consult the user first, as <ranges> is a heavyweight include. If the user rejects the use of <ranges>, `spawn_group` / `fork_group` imperative construction avoids iterators entirely.
+
+### Customizing and initiating
+
+Most awaitable wrappers above expose fluent customizers (return `*this`, chainable). Some awaitables also provide them directly:
+
+- `.run_on(executor)` — where the work runs.
+- `.resume_on(executor)` — where the parent resumes afterward.
+- `.with_priority(p)` — priority level for the work.
+
+After customizing, initiate the work with **exactly one** of:
+
+- `co_await` — run and await the result inline.
+- `.fork()` — start eagerly now, await the returned handle later.
+- `.detach()` — fire-and-forget; no result is retrieved.
+
+### HALO (`*_clang()` variants) — Clang only
+
+TMC provides HALO (Heap Allocation Elision Optimization) entry points that use Clang-specific
+attributes to elide the child coroutine's frame allocation: `tmc::spawn_clang`,
+`tmc::fork_clang`, `tmc::fork_tuple_clang`, `mux.fork_clang(...)`,
+`sg.add_clang(...)`, `fg.fork_clang(...)`. Each **must be `co_await`ed
+immediately** in the same statement for elision to fire. The `fork_clang` functions
+return a dummy awaitable that you must await immediately for elision. The return
+value of that `co_await` expression is the real forked task handle, which you await
+later to join the forked task.
+
+**Before suggesting a `*_clang()` API, confirm the user's build environment
+actually targets Clang** (compiler flags, CMake toolchain, CI config).
+On other compilers they offer no benefit — use the plain variants.
+
+## Control structures
+
+Async equivalents of familiar primitives; behavior matches the name:
+
+`mutex`, `semaphore`, `rw_lock`, `barrier`, `latch`, `manual_reset_event`,
+`auto_reset_event`, `atomic_condvar`.
+
+## Queues
+
+- `qu_spsc_bounded`, `qu_spsc_unbounded` — single-producer, single-consumer.
+- `qu_mpsc_bounded`, `qu_mpsc_unbounded` — multi-producer, single-consumer.
+- `channel` — MPMC. Create with `tmc::make_channel<T>()`, which returns a
+  `chan_tok` — a hazard-pointer + shared-ownership handle to the channel. Access
+  the channel through the `chan_tok`; copy it (`new_token()`) to hand additional
+  producers/consumers their own token.

CLAUDE.md

@@ -0,0 +1 @@
+@AGENTS.md

include/tmc/all_headers.hpp

@@ -27,6 +27,8 @@
 #include "tmc/latch.hpp"              // IWYU pragma: export
 #include "tmc/manual_reset_event.hpp" // IWYU pragma: export
 #include "tmc/mutex.hpp"              // IWYU pragma: export
+#include "tmc/mux_many.hpp"           // IWYU pragma: export
+#include "tmc/mux_tuple.hpp"          // IWYU pragma: export
 #include "tmc/qu_mpsc_bounded.hpp"    // IWYU pragma: export
 #include "tmc/qu_mpsc_unbounded.hpp"  // IWYU pragma: export
 #include "tmc/qu_spsc_bounded.hpp"    // IWYU pragma: export

include/tmc/detail/result_each.hpp

@@ -18,12 +18,9 @@
 namespace tmc {
 namespace detail {
 
-TMC_DECL bool result_each_await_ready() noexcept;
-
 TMC_DECL bool result_each_await_suspend(
   ptrdiff_t remaining_count, std::coroutine_handle<> Outer,
-  std::coroutine_handle<>& continuation, tmc::ex_any* continuation_executor,
-  std::atomic<size_t>& sync_flags
+  std::coroutine_handle<>& continuation, std::atomic<size_t>& sync_flags
 ) noexcept;
 
 TMC_DECL size_t result_each_await_resume(

include/tmc/detail/result_each.ipp

@@ -8,8 +8,6 @@
 #include "tmc/detail/awaitable_customizer.hpp"
 #include "tmc/detail/impl.hpp" // IWYU pragma: keep
 #include "tmc/detail/result_each.hpp"
-#include "tmc/detail/thread_locals.hpp"
-#include "tmc/ex_any.hpp"
 
 #include <atomic>
 #include <bit>
@@ -19,54 +17,36 @@
 namespace tmc {
 namespace detail {
 
-bool result_each_await_ready() noexcept {
-  // Always suspends, due to the possibility to resume on another executor.
-  return false;
-}
-
 bool result_each_await_suspend(
   ptrdiff_t remaining_count, std::coroutine_handle<> Outer,
-  std::coroutine_handle<>& continuation, tmc::ex_any* continuation_executor,
-  std::atomic<size_t>& sync_flags
+  std::coroutine_handle<>& continuation, std::atomic<size_t>& sync_flags
 ) noexcept {
   continuation = Outer;
-  if (remaining_count != 0) {
-    // This logic is necessary because we submitted all child tasks before the
-    // parent suspended. Allowing parent to be resumed before it suspends
-    // would be UB. Therefore we need to block the resumption until here.
-    // WARNING: We can use fetch_sub here because we know this bit is set.
-    // It generates xadd instruction which is slightly more efficient than
-    // fetch_and. But not safe to use if the bit might not be set.
-    size_t resumeState;
-    do {
-      resumeState = sync_flags.fetch_sub(
-        tmc::detail::task_flags::EACH, std::memory_order_acq_rel
-      );
-      assert(0 != (resumeState & tmc::detail::task_flags::EACH));
-      if (0 == (resumeState & ~tmc::detail::task_flags::EACH)) {
-        return true; // we suspended and no tasks were ready
-      }
-      // A result became ready, so try to resume immediately.
-      resumeState = sync_flags.fetch_or(
-        tmc::detail::task_flags::EACH, std::memory_order_acq_rel
-      );
-      if (0 != (resumeState & tmc::detail::task_flags::EACH)) {
-        return true; // Another thread already resumed
-      }
-      // If we resumed, but another thread already consumed
-      // all the results, try again to suspend
-    } while (0 == (resumeState & ~tmc::detail::task_flags::EACH));
-  }
-  if (continuation_executor == nullptr ||
-      tmc::detail::this_thread::exec_is(continuation_executor)) {
+  if (remaining_count == 0) {
     return false;
-  } else {
-    // Need to resume on a different executor
-    tmc::detail::post_checked(
-      continuation_executor, std::move(Outer),
-      tmc::detail::this_thread::this_task().prio
-    );
-    return true;
+  }
+  // Children are submitted before the parent suspends. Allowing the parent to
+  // be resumed before it suspends would be UB, so children are blocked from
+  // resuming it by the EACH bit (lock bit), which is held for the entire time
+  // the parent is running. It may only be released here, atomically with the
+  // decision to suspend, and only if no results are already ready.
+  // This release must be the final operation via CAS rather than speculatively, because
+  // once released, another thread may resume this and destroy the coroutine frame.
+  size_t state = sync_flags.load(std::memory_order_acquire);
+  while (true) {
+    assert(0 != (state & tmc::detail::task_flags::EACH));
+    if (0 != (state & ~tmc::detail::task_flags::EACH)) {
+      // A result is ready. Resume immediately. The EACH bit was never
+      // released, so no child could have claimed the resumption.
+      return false;
+    }
+    if (sync_flags.compare_exchange_weak(
+          state, state & ~tmc::detail::task_flags::EACH, std::memory_order_acq_rel,
+          std::memory_order_acquire
+        )) {
+      return true;
+    }
+    // CAS failure means a child just set its result bit; retry to consume it.
   }
 }
 

include/tmc/detail/tsan.hpp

@@ -1,9 +1,14 @@
+// Copyright (c) 2023-2026 Logan McDougall
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt)
+
 #pragma once
 
 // The Chase-Lev deque has a racy read where it will memcpy data that may be
 // invalid, and then check the validity by an atomic load afterward. If invalid,
 // the potentially-garbage data is discarded. TSan doesn't like this.
-// When the data size is pointer-sized (std::coroutine_handle) we can solve this
+// When the data size is 1 pointer (std::coroutine_handle) we can solve this
 // by accessing it using a relaxed read from an atomic_ref, which TSan accepts.
 // When the data size is 2 pointers (tmc::coro_functor), we would have to use a
 // lot of workaround hacks. It's simpler to just disable TSan for the specific
@@ -20,3 +25,16 @@
 #else
 #define TMC_INLINE_OR_TSAN TMC_FORCE_INLINE
 #endif
+
+// Some compilers (observed on Clang 18/19 and Apple Clang) may convert
+// `Cond ? mux.get<0>() : mux.get<1>()` into a select of both branches.
+// The reads happen after the atomic acquire of Cond, so it's not a true race,
+// but it triggers a TSan false positive because the discarded value races
+// with a write from the completing task. Forcing this to be noinline under
+// TSan prevents the speculative load from being issued, while still allowing
+// the function to be instrumented by TSan for genuine races.
+#ifdef TMC_HAS_TSAN
+#define TMC_TSAN_NO_SPECULATE __attribute__((noinline))
+#else
+#define TMC_TSAN_NO_SPECULATE
+#endif

include/tmc/detail/tuple_helpers.hpp

@@ -0,0 +1,83 @@
+// Copyright (c) 2023-2026 Logan McDougall
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt)
+
+#pragma once
+
+// Common template helpers shared by the tuple-shaped awaitable groups
+// (tmc::spawn_tuple() and tmc::mux_tuple).
+
+#include "tmc/detail/concepts_awaitable.hpp" // IWYU pragma: keep
+
+#include <type_traits>
+#include <variant>
+
+namespace tmc {
+namespace detail {
+// Replace void with std::monostate (void is not a valid tuple element type)
+template <typename T>
+using void_to_monostate = std::conditional_t<std::is_void_v<T>, std::monostate, T>;
+
+// Get the last type of a parameter pack
+// In C++26 you can use pack indexing instead: T...[sizeof...(T) - 1]
+template <typename... T> struct last_type {
+  using type = typename decltype((std::type_identity<T>{}, ...))::type;
+};
+
+template <> struct last_type<> {
+  // workaround for empty tuples - the task object will be void / empty
+  using type = void;
+};
+
+template <typename... T> using last_type_t = typename last_type<T...>::type;
+
+// Create 2 instantiations of the Variadic, one where all of the types
+// satisfy the predicate, and one where none of the types satisfy the predicate.
+template <template <class> class Predicate, template <class...> class Variadic, class...>
+struct predicate_partition;
+
+// Definition for empty parameter pack
+template <template <class> class Predicate, template <class...> class Variadic>
+struct predicate_partition<Predicate, Variadic> {
+  using true_types = Variadic<>;
+  using false_types = Variadic<>;
+};
+
+template <
+  template <class> class Predicate, template <class...> class Variadic, class T,
+  class... Ts>
+struct predicate_partition<Predicate, Variadic, T, Ts...> {
+  template <class, class> struct Cons;
+  template <class Head, class... Tail> struct Cons<Head, Variadic<Tail...>> {
+    using type = Variadic<Head, Tail...>;
+  };
+
+  // Every type in the parameter pack satisfies the predicate
+  using true_types = typename std::conditional<
+    Predicate<T>::value,
+    typename Cons<
+      T, typename predicate_partition<Predicate, Variadic, Ts...>::true_types>::type,
+    typename predicate_partition<Predicate, Variadic, Ts...>::true_types>::type;
+
+  // No type in the parameter pack satisfies the predicate
+  using false_types = typename std::conditional<
+    !Predicate<T>::value,
+    typename Cons<
+      T, typename predicate_partition<Predicate, Variadic, Ts...>::false_types>::type,
+    typename predicate_partition<Predicate, Variadic, Ts...>::false_types>::type;
+};
+
+// Partition the awaitables into two groups:
+// 1. The awaitables that are coroutines, which will be submitted in bulk /
+// symmetric transfer.
+// 2. The awaitables that are not coroutines, which will be initiated by
+// async_initiate.
+template <typename T> struct treat_as_coroutine {
+  static constexpr bool value =
+    tmc::detail::get_awaitable_traits<T>::mode == tmc::detail::TMC_TASK ||
+    tmc::detail::get_awaitable_traits<T>::mode == tmc::detail::COROUTINE ||
+    tmc::detail::get_awaitable_traits<T>::mode == tmc::detail::WRAPPER;
+};
+} // namespace detail
+} // namespace tmc