From 21dc3d5ffe06578c3c8ab1fd2c4233be6473dae9 Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 19:02:47 -0700
Subject: [PATCH 1/9] add mux_many / mux_tuple

---
 include/tmc/all_headers.hpp          |   2 +
 include/tmc/detail/tuple_helpers.hpp |  83 +++++
 include/tmc/mux_many.hpp             | 535 +++++++++++++++++++++++++++
 include/tmc/mux_tuple.hpp            | 400 ++++++++++++++++++++
 4 files changed, 1020 insertions(+)
 create mode 100644 include/tmc/detail/tuple_helpers.hpp
 create mode 100644 include/tmc/mux_many.hpp
 create mode 100644 include/tmc/mux_tuple.hpp

diff --git a/include/tmc/all_headers.hpp b/include/tmc/all_headers.hpp
index 86936690..9f14633b 100644
--- a/include/tmc/all_headers.hpp
+++ b/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
diff --git a/include/tmc/detail/tuple_helpers.hpp b/include/tmc/detail/tuple_helpers.hpp
new file mode 100644
index 00000000..b4999c39
--- /dev/null
+++ b/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
diff --git a/include/tmc/mux_many.hpp b/include/tmc/mux_many.hpp
new file mode 100644
index 00000000..070d6d39
--- /dev/null
+++ b/include/tmc/mux_many.hpp
@@ -0,0 +1,535 @@
+// 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
+
+#include "tmc/current.hpp"
+#include "tmc/detail/awaitable_customizer.hpp"
+#include "tmc/detail/compat.hpp"
+#include "tmc/detail/concepts_awaitable.hpp" // IWYU pragma: keep
+#include "tmc/detail/concepts_work_item.hpp"
+#include "tmc/detail/result_each.hpp"
+#include "tmc/detail/thread_locals.hpp"
+#include "tmc/ex_any.hpp"
+#include "tmc/work_item.hpp"
+
+#include <array>
+#include <atomic>
+#include <cassert>
+#include <coroutine>
+#include <cstddef>
+#include <iterator>
+#include <type_traits>
+#include <utility>
+
+namespace tmc {
+/// 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:
+///
+/// 1. 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.
+///
+/// 2. 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) { ... }
+/// ```
+template <typename Result, size_t Count = TMC_PLATFORM_BITS - 1>
+class mux_many : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
+  // Tasks are synchronized via an atomic bitmask with only 63 (or 31, on 32-bit)
+  // slots for tasks.
+  static_assert(
+    Count < TMC_PLATFORM_BITS,
+    "mux_many supports up to 63 awaitables (31 on 32-bit platforms)."
+  );
+
+  // The count of submitted-but-not-yet-consumed results.
+  ptrdiff_t remaining_count;
+  std::coroutine_handle<> continuation;
+  tmc::ex_any* continuation_executor;
+  // Bitmap of slots that have been forked but not yet returned from co_await.
+  // A set bit means the slot is active: its result is pending or ready but has
+  // not been consumed, so the slot may not be re-forked. Non-atomic; only
+  // mutated by the (single-threaded) owner.
+  size_t active_slots;
+  // the atomic synchronization variable that coordinates between this and
+  // awaitable_customizer (task's final_suspend)
+  std::atomic<size_t> sync_flags = tmc::detail::task_flags::EACH;
+
+  // Non-default-constructible results are stored in a std::optional. This is not visible
+  // to the user; when get() is called, a reference to the constructed object is returned.
+  using ResultStorage = tmc::detail::result_storage_t<Result>;
+  struct empty {};
+  using ResultArray =
+    std::conditional_t<std::is_void_v<Result>, empty, std::array<ResultStorage, Count>>;
+  TMC_NO_UNIQUE_ADDRESS ResultArray result_arr;
+
+  // Prepares the work item but does not initiate it.
+  template <typename T>
+  TMC_FORCE_INLINE inline void
+  prepare_work(T& Task, size_t Idx, size_t ContinuationPrio) {
+    tmc::detail::get_awaitable_traits<T>::set_continuation(Task, &continuation);
+    tmc::detail::get_awaitable_traits<T>::set_continuation_executor(
+      Task, &continuation_executor
+    );
+    tmc::detail::get_awaitable_traits<T>::set_done_count(Task, &sync_flags);
+    tmc::detail::get_awaitable_traits<T>::set_flags(
+      Task, tmc::detail::task_flags::EACH |
+              (Idx << tmc::detail::task_flags::TASKNUM_LOW_OFF) | ContinuationPrio
+    );
+    if constexpr (!std::is_void_v<Result>) {
+      tmc::detail::get_awaitable_traits<T>::set_result_ptr(Task, &result_arr[Idx]);
+    }
+  }
+
+  // Sets the owner-only (non-atomic) tracking variables.
+  void init_active_counts(size_t NumTasks) {
+    remaining_count = static_cast<ptrdiff_t>(NumTasks);
+    active_slots = (TMC_ONE_BIT << NumTasks) - 1;
+  }
+
+  // A sentinel for initiate_eager() iterations that are bounded only by the
+  // count; it never compares equal to the iterator.
+  struct unbounded {
+    template <typename TaskIter>
+    friend bool operator!=(const TaskIter&, unbounded) noexcept {
+      return true;
+    }
+  };
+
+  // Eagerly initiates awaitables from `Iter` until `Size` have been initiated
+  // or `Iter == End` (pass `unbounded{}` when there is no end iterator).
+  // Returns the number initiated. Initiated awaitables interact only with
+  // sync_flags as they complete, so the caller may set the tracking counts
+  // from the returned value afterward.
+  template <typename TaskIter, typename EndIter>
+  size_t initiate_eager(TaskIter Iter, EndIter End, size_t Size) {
+    using Awaitable = std::remove_cvref_t<std::iter_value_t<TaskIter>>;
+    constexpr auto mode = tmc::detail::get_awaitable_traits<Awaitable>::mode;
+    static_assert(
+      mode != tmc::detail::UNKNOWN, "This doesn't appear to be an awaitable."
+    );
+
+    tmc::ex_any* executor = tmc::detail::this_thread::executor();
+    size_t prio = tmc::detail::this_thread::this_task().prio;
+
+    size_t taskCount = 0;
+    if constexpr (mode == tmc::detail::ASYNC_INITIATE) {
+      // ASYNC_INITIATE types may possibly not be stored in an array
+      // (no default/copy constructor), so initiate them individually.
+      while (Iter != End && taskCount < Size) {
+        TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
+        auto t = std::move(*Iter);
+        TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
+        prepare_work(t, taskCount, prio);
+        tmc::detail::get_awaitable_traits<Awaitable>::async_initiate(
+          std::move(t), executor, prio
+        );
+        ++Iter;
+        ++taskCount;
+      }
+    } else {
+      // Batch other types of awaitables into a work_item array and submit them
+      // in bulk. The iterator could produce fewer than `Size` awaitables, so
+      // count them; it could also produce more, so stop after taking `Size`.
+      // The result storage is a fixed-size array, so result pointers are stable
+      // and can be bound as each awaitable is prepared - no second pass or
+      // separate type-preserving buffer (as an unsized std::vector would need)
+      // is required, even for COROUTINE-mode awaitables.
+      std::array<work_item, Count> taskArr;
+      while (Iter != End && taskCount < Size) {
+        TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
+        auto t = tmc::detail::into_known<false>(std::move(*Iter));
+        TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
+        prepare_work(t, taskCount, prio);
+        taskArr[taskCount] = tmc::detail::into_initiate(std::move(t));
+        ++Iter;
+        ++taskCount;
+      }
+
+      if (taskCount != 0) {
+        tmc::detail::post_bulk_checked(executor, taskArr.data(), taskCount, prio);
+      }
+    }
+    return taskCount;
+  }
+
+public:
+  /// 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.
+  mux_many() { init_active_counts(0); }
+
+  /// 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>
+    requires(requires(AwaitableIter a) {
+      ++a;
+      *a;
+    })
+  inline mux_many(AwaitableIter&& Iter, size_t AwaitableCount) {
+    size_t size = AwaitableCount < Count ? AwaitableCount : Count;
+    init_active_counts(
+      initiate_eager(std::forward<AwaitableIter>(Iter), unbounded{}, size)
+    );
+  }
+
+  /// 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>
+    requires(requires(AwaitableIter a, AwaitableIter b) {
+      ++a;
+      *a;
+      a != b;
+    })
+  inline mux_many(AwaitableIter&& Begin, AwaitableIter&& End, size_t MaxCount) {
+    size_t size = MaxCount < Count ? MaxCount : Count;
+    init_active_counts(initiate_eager(
+      std::forward<AwaitableIter>(Begin), std::forward<AwaitableIter>(End), size
+    ));
+  }
+
+  /// 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>
+    requires(
+      !requires(std::remove_reference_t<AwaitableIter>& r) {
+        r.begin();
+        r.end();
+      } &&
+      requires(AwaitableIter a) {
+        ++a;
+        *a;
+      }
+    )
+  inline mux_many(AwaitableIter&& Begin)
+      : mux_many(std::forward<AwaitableIter>(Begin), Count) {}
+
+  /// 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 AwaitableIter>
+    requires(requires(AwaitableIter a, AwaitableIter b) {
+      ++a;
+      *a;
+      a != b;
+    })
+  inline mux_many(AwaitableIter&& Begin, AwaitableIter&& End)
+      : mux_many(
+          std::forward<AwaitableIter>(Begin), std::forward<AwaitableIter>(End), Count
+        ) {}
+
+  /// 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.
+  template <typename AwaitableRange>
+    requires(requires(std::remove_reference_t<AwaitableRange>& r) {
+      r.begin();
+      r.end();
+    })
+  inline mux_many(AwaitableRange&& Range) : mux_many(Range.begin(), Range.end(), Count) {}
+
+  /// Always suspends.
+  inline bool await_ready() const noexcept { return false; }
+
+  /// Suspends the outer coroutine until a result becomes ready (or resumes
+  /// immediately if one is already ready).
+  inline bool await_suspend(std::coroutine_handle<> Outer) noexcept {
+    // mux_many can be awaited in a different context than where it was
+    // created. Therefore, capture the continuation_executor at the await point.
+    continuation_executor = tmc::detail::this_thread::executor();
+    return tmc::detail::result_each_await_suspend(
+      remaining_count, Outer, continuation, sync_flags
+    );
+  }
+
+  /// 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()`.
+  TMC_AWAIT_RESUME inline size_t await_resume() noexcept {
+    auto slot = tmc::detail::result_each_await_resume(remaining_count, sync_flags);
+    if (slot != end()) {
+      active_slots &= ~(TMC_ONE_BIT << slot);
+    }
+    return slot;
+  }
+
+  /// This type must be awaited as an lvalue (it is awaited repeatedly and is not
+  /// movable). The awaiter is the group itself.
+  mux_many& operator co_await() & noexcept { return *this; }
+
+  /// Provides a sentinel value that can be compared against the value returned
+  /// from co_await.
+  inline size_t end() noexcept { return 64; }
+
+  /// Gets the ready result at the given index.
+  inline std::add_lvalue_reference_t<Result> operator[](size_t Idx) noexcept
+    requires(!std::is_void_v<Result>)
+  {
+    assert(Idx < result_arr.size());
+    assert(
+      !is_active(Idx) && "You may only call get() on a slot after its result "
+                         "has been returned from co_await."
+    );
+    if constexpr (std::is_default_constructible_v<Result>) {
+      return result_arr[Idx];
+    } else {
+      return *result_arr[Idx];
+    }
+  }
+
+  /// Provided for convenience only - to expose the same API as the non-void
+  /// version. Does nothing.
+  inline void operator[]([[maybe_unused]] size_t Idx) noexcept
+    requires(std::is_void_v<Result>)
+  {}
+
+  /// 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 bool is_active(size_t Idx) const noexcept {
+    assert(Idx < Count && "is_active() index out of range.");
+    return 0 != (active_slots & (TMC_ONE_BIT << Idx));
+  }
+
+  /// 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`.
+  inline size_t active_bitset() const noexcept { return active_slots; }
+
+  /// 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.
+  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()
+  ) {
+    assert(Idx < Count && "fork() index out of range.");
+
+    auto slotBit = TMC_ONE_BIT << Idx;
+    assert(
+      0 == (active_slots & slotBit) &&
+      "You may only fork a slot after its previous result has been awaited."
+    );
+    active_slots |= slotBit;
+    ++remaining_count;
+
+    decltype(auto) known = tmc::detail::into_known<false>(static_cast<Aw&&>(Awaitable));
+    using KnownAwaitable = std::remove_reference_t<decltype(known)>;
+    static_assert(
+      std::is_same_v<
+        typename tmc::detail::get_awaitable_traits<KnownAwaitable>::result_type, Result>,
+      "Replacement awaitable must have the same result type as the mux_many "
+      "group."
+    );
+
+    // Destroy the previously-consumed result before initiating the replacement. This
+    // prevents issues with results that own a resource which the replacement awaitable
+    // needs to re-acquire, such as zero-copy queue scopes.
+    if constexpr (!std::is_void_v<Result>) {
+      result_arr[Idx] = ResultStorage{};
+    }
+
+    tmc::ex_any* exec = tmc::detail::get_executor_traits<Exec>::type_erased(Executor);
+    auto continuationPriority = tmc::detail::this_thread::this_task().prio;
+
+    prepare_work(known, Idx, continuationPriority);
+
+    constexpr auto mode = tmc::detail::get_awaitable_traits<KnownAwaitable>::mode;
+    if constexpr (mode == tmc::detail::ASYNC_INITIATE) {
+      // Forward known with its original value category.
+      tmc::detail::get_awaitable_traits<KnownAwaitable>::async_initiate(
+        static_cast<decltype(known)&&>(known), exec, Priority
+      );
+    } else {
+      auto item = tmc::detail::into_initiate(static_cast<decltype(known)&&>(known));
+      tmc::detail::post_checked(exec, std::move(item), Priority);
+    }
+  }
+
+  /// This is a dummy awaitable. Don't store this in a variable.
+  /// For HALO to work, you must `co_await mux.fork_clang()` immediately.
+  class TMC_CORO_AWAIT_ELIDABLE mux_many_fork_clang : tmc::detail::AwaitTagNoGroupAsIs {
+  public:
+    mux_many_fork_clang() {}
+
+    /// Never suspends.
+    bool await_ready() const noexcept { return true; }
+
+    /// Does nothing.
+    void await_suspend(std::coroutine_handle<>) noexcept {}
+
+    /// Does nothing.
+    void await_resume() noexcept {}
+  };
+
+  /// 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.
+  ///
+  /// WARNING: Don't allow coroutines passed into this to cross a loop boundary,
+  /// or Clang will try to reuse the same allocation for multiple active
+  /// coroutines.
+  ///
+  /// 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:
+  /// ```
+  /// auto mux = tmc::mux_many<int, 2>();
+  /// co_await mux.fork_clang(0, task(0));
+  /// co_await mux.fork_clang(1, task(1));
+  /// for (size_t i = co_await mux; i != mux.end(); i = co_await mux) { ... }
+  /// ```
+  template <typename Aw, typename Exec = tmc::ex_any*>
+  [[nodiscard("You must co_await fork_clang() immediately for HALO to be possible.")]]
+  mux_many_fork_clang fork_clang(
+    size_t Idx, TMC_CORO_AWAIT_ELIDABLE_ARGUMENT Aw&& Awaitable,
+    Exec&& Executor = tmc::current_executor(), size_t Priority = tmc::current_priority()
+  ) {
+    fork(Idx, static_cast<Aw&&>(Awaitable), static_cast<Exec&&>(Executor), Priority);
+    return mux_many_fork_clang{};
+  }
+
+  // This must be awaited and all child awaitables completed before destruction.
+#ifndef NDEBUG
+  ~mux_many() noexcept {
+    assert(
+      remaining_count == 0 &&
+      "You must wait for all mux_many awaitables to complete before destroying it."
+    );
+  }
+#endif
+
+  // Not movable or copyable due to awaitables being initiated with pointers to
+  // this.
+  mux_many& operator=(const mux_many& other) = delete;
+  mux_many(const mux_many& other) = delete;
+  mux_many& operator=(mux_many&& other) = delete;
+  mux_many(mux_many&& other) = delete;
+};
+
+// Deduction guides for the eager construction forms. Each deduces the `Result`
+// type from the awaitables and leaves `Count` at its default
+// (`TMC_PLATFORM_BITS - 1`), so a full-size fixed `std::array<Result, Count>` is
+// used. To right-size the storage, name both template arguments explicitly -
+// CTAD cannot deduce only `Result` while you fix `Count`, so there is no guide
+// for the explicit-`Count` forms.
+//
+// The eager constructors are member templates whose own parameters do not appear
+// in the class's `Result`/`Count` parameter list, so their implicit deduction
+// guides cannot deduce those parameters and are discarded - these written guides
+// are the only viable ones.
+
+/// Eagerly initiates items in range [Begin, Begin + AwaitableCount).
+template <
+  typename AwaitableIter, typename Awaitable = std::iter_value_t<AwaitableIter>,
+  typename Result = tmc::detail::awaitable_result_t<Awaitable>>
+mux_many(AwaitableIter&&, size_t) -> mux_many<Result>;
+
+/// Eagerly initiates items in range [Begin, End).
+template <
+  typename AwaitableIter, typename Awaitable = std::iter_value_t<AwaitableIter>,
+  typename Result = tmc::detail::awaitable_result_t<Awaitable>>
+mux_many(AwaitableIter&&, AwaitableIter&&) -> mux_many<Result>;
+
+/// Eagerly initiates items in range [Begin, min(Begin + MaxCount, End)).
+template <
+  typename AwaitableIter, typename Awaitable = std::iter_value_t<AwaitableIter>,
+  typename Result = tmc::detail::awaitable_result_t<Awaitable>>
+mux_many(AwaitableIter&&, AwaitableIter&&, size_t) -> mux_many<Result>;
+
+/// Eagerly initiates items in range [Range.begin(), Range.end()). The
+/// `range_iter` default template argument keeps this guide from matching a
+/// non-range single argument.
+template <
+  typename AwaitableRange,
+  typename AwaitableIter = tmc::detail::range_iter<AwaitableRange>::type,
+  typename Awaitable = std::iter_value_t<AwaitableIter>,
+  typename Result = tmc::detail::awaitable_result_t<Awaitable>>
+mux_many(AwaitableRange&&) -> mux_many<Result>;
+
+} // namespace tmc
diff --git a/include/tmc/mux_tuple.hpp b/include/tmc/mux_tuple.hpp
new file mode 100644
index 00000000..6f53f125
--- /dev/null
+++ b/include/tmc/mux_tuple.hpp
@@ -0,0 +1,400 @@
+// 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
+
+#include "tmc/current.hpp"
+#include "tmc/detail/awaitable_customizer.hpp"
+#include "tmc/detail/compat.hpp"
+#include "tmc/detail/concepts_awaitable.hpp" // IWYU pragma: keep
+#include "tmc/detail/concepts_work_item.hpp"
+#include "tmc/detail/result_each.hpp"
+#include "tmc/detail/thread_locals.hpp"
+#include "tmc/detail/tuple_helpers.hpp"
+#include "tmc/ex_any.hpp"
+#include "tmc/work_item.hpp"
+
+#include <array>
+#include <atomic>
+#include <cassert>
+#include <coroutine>
+#include <cstddef>
+#include <tuple>
+#include <type_traits>
+#include <utility>
+
+namespace tmc {
+/// 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<I>()` 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;
+/// ```
+template <typename... Result>
+class mux_tuple : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
+  static constexpr auto Count = sizeof...(Result);
+
+  // Tasks are synchronized via an atomic bitmask with only 63 (or 31, on 32-bit)
+  // slots for tasks.
+  static_assert(
+    Count < TMC_PLATFORM_BITS,
+    "mux_tuple supports at most 63 awaitables (31 on 32-bit platforms)."
+  );
+
+  std::coroutine_handle<> continuation;
+  tmc::ex_any* continuation_executor;
+
+  // Bitmap of slots that have been forked but not yet returned from co_await.
+  // A set bit means the slot is active: its result is pending or ready but has
+  // not been consumed, so the slot may not be re-forked. Non-atomic; only
+  // mutated by the (single-threaded) owner.
+  size_t active_slots;
+  // equal to popcount(active_slots)
+  ptrdiff_t remaining_count;
+
+  // the atomic synchronization variable that coordinates between this and
+  // awaitable_customizer (task's final_suspend)
+  std::atomic<size_t> sync_flags = tmc::detail::task_flags::EACH;
+
+  // Non-default-constructible results are stored in a std::optional. This is not
+  // visible to the user; when get() is called, a reference to the constructed object is
+  // returned.
+  template <typename R>
+  using ResultStorage = tmc::detail::result_storage_t<tmc::detail::void_to_monostate<R>>;
+  using ResultTuple = std::tuple<ResultStorage<Result>...>;
+  // The underlying value type of each slot (before optional-wrapping), with void
+  // slots represented as std::monostate. get<I>() returns a reference to this
+  // type, unwrapping the optional storage for non-default-constructible values.
+  using ValueTuple = std::tuple<tmc::detail::void_to_monostate<Result>...>;
+  ResultTuple result;
+
+  // Prepares the work item but does not initiate it.
+  template <typename T, typename R>
+  TMC_FORCE_INLINE inline void
+  prepare_work(T& Task, R* TaskResult, size_t Idx, size_t ContinuationPrio) {
+    tmc::detail::get_awaitable_traits<T>::set_continuation(Task, &continuation);
+    tmc::detail::get_awaitable_traits<T>::set_continuation_executor(
+      Task, &continuation_executor
+    );
+    tmc::detail::get_awaitable_traits<T>::set_done_count(Task, &sync_flags);
+    tmc::detail::get_awaitable_traits<T>::set_flags(
+      Task, tmc::detail::task_flags::EACH |
+              (Idx << tmc::detail::task_flags::TASKNUM_LOW_OFF) | ContinuationPrio
+    );
+    if constexpr (!std::is_void_v<
+                    typename tmc::detail::get_awaitable_traits<T>::result_type>) {
+      tmc::detail::get_awaitable_traits<T>::set_result_ptr(Task, TaskResult);
+    }
+  }
+
+  // Sets the owner-only (non-atomic) tracking variables.
+  void init_active_counts(size_t NumTasks) {
+    remaining_count = static_cast<ptrdiff_t>(NumTasks);
+    active_slots = (TMC_ONE_BIT << NumTasks) - 1;
+  }
+
+public:
+  /// Creates the result storage but does not initiate any awaitables. The
+  /// Result template types must be provided explicitly. Use `fork<I>()` to
+  /// initiate work into individual slots.
+  mux_tuple() { init_active_counts(0); }
+
+  /// Eagerly initiates all of the provided awaitables. The Result template types
+  /// are deduced from the awaitable arguments.
+  template <typename... Awaitable>
+    requires(sizeof...(Awaitable) != 0)
+  mux_tuple(Awaitable&&... Awaitables) {
+    static_assert(
+      std::is_same_v<
+        ResultTuple, std::tuple<ResultStorage<typename tmc::detail::get_awaitable_traits<
+                       Awaitable>::result_type>...>>,
+      "The provided awaitables' result types (and their count) must match the "
+      "mux_tuple's result-type template arguments."
+    );
+
+    // Hold awaitables by reference, as if by std::forward_as_tuple.
+    // This is safe because the awaitables are initiated within this constructor,
+    // not permanently stored.
+    using AwaitableTuple = std::tuple<Awaitable&&...>;
+    AwaitableTuple Tasks(static_cast<Awaitable&&>(Awaitables)...);
+
+    // Compile-time count of the number of non-ASYNC_INITIATE awaitables so they can be
+    // batched and submitted with a single post_bulk.
+    constexpr size_t WorkItemCount =
+      std::tuple_size_v<typename tmc::detail::predicate_partition<
+        tmc::detail::treat_as_coroutine, std::tuple, Awaitable&&...>::true_types>;
+    std::array<work_item, WorkItemCount> taskArr;
+
+    init_active_counts(Count);
+    tmc::ex_any* executor = tmc::detail::this_thread::executor();
+    size_t prio = tmc::detail::this_thread::this_task().prio;
+
+    // Prepare each task as if I loops from [0..Count),
+    // but using compile-time indexes and types.
+    size_t taskIdx = 0;
+    [&]<std::size_t... I>(std::index_sequence<I...>) {
+      (([&]() {
+         using Elem = std::tuple_element_t<I, AwaitableTuple>;
+         constexpr auto mode = tmc::detail::get_awaitable_traits<Elem>::mode;
+         static_assert(
+           mode != tmc::detail::UNKNOWN, "This doesn't appear to be an awaitable."
+         );
+         // The std::move is a bit misleading here - std::get actually forwards
+         // the awaitable from within the tuple with its original value category.
+         if constexpr (mode == tmc::detail::ASYNC_INITIATE) {
+           // Individually initiate the awaitables.
+           decltype(auto) t = std::get<I>(std::move(Tasks));
+           prepare_work(t, &std::get<I>(result), I, prio);
+           tmc::detail::get_awaitable_traits<Elem>::async_initiate(
+             static_cast<decltype(t)&&>(t), executor, prio
+           );
+         } else {
+           // Batch the coroutines into an array so they can be submitted in bulk.
+           decltype(auto) known =
+             tmc::detail::into_known<false>(std::get<I>(std::move(Tasks)));
+           prepare_work(known, &std::get<I>(result), I, prio);
+           taskArr[taskIdx] =
+             tmc::detail::into_initiate(static_cast<decltype(known)&&>(known));
+           ++taskIdx;
+         }
+       }()),
+       ...);
+    }(std::make_index_sequence<Count>{});
+
+    // Bulk submit the coroutines
+    if constexpr (WorkItemCount != 0) {
+      tmc::detail::post_bulk_checked(executor, taskArr.data(), WorkItemCount, prio);
+    }
+  }
+
+  /// Always suspends.
+  inline bool await_ready() const noexcept { return false; }
+
+  /// Suspends the outer coroutine until a result becomes ready (or resumes
+  /// immediately if one is already ready).
+  inline bool await_suspend(std::coroutine_handle<> Outer) noexcept {
+    // mux_tuple can be awaited in a different context than where it was
+    // created. Therefore, capture the continuation_executor at the await point.
+    continuation_executor = tmc::detail::this_thread::executor();
+    return tmc::detail::result_each_await_suspend(
+      remaining_count, Outer, continuation, sync_flags
+    );
+  }
+
+  /// 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()`.
+  TMC_AWAIT_RESUME inline size_t await_resume() noexcept {
+    auto slot = tmc::detail::result_each_await_resume(remaining_count, sync_flags);
+    if (slot != end()) {
+      active_slots &= ~(TMC_ONE_BIT << slot);
+    }
+    return slot;
+  }
+
+  /// This type must be awaited as an lvalue (it is awaited repeatedly and is not
+  /// movable). The awaiter is the group itself.
+  mux_tuple& operator co_await() & noexcept { return *this; }
+
+  /// Provides a sentinel value that can be compared against the value returned
+  /// from co_await.
+  inline size_t end() noexcept { return 64; }
+
+  /// Gets the ready result at the given index.
+  template <size_t I> inline std::tuple_element_t<I, ValueTuple>& get() noexcept {
+    assert(
+      !is_active<I>() && "You may only call get() on a slot after its result "
+                         "has been returned from co_await."
+    );
+    using Value = std::tuple_element_t<I, ValueTuple>;
+    if constexpr (std::is_default_constructible_v<Value>) {
+      return std::get<I>(result);
+    } else {
+      return *std::get<I>(result);
+    }
+  }
+
+  /// 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.
+  template <size_t I> inline bool is_active() const noexcept {
+    static_assert(I < Count, "is_active() index out of range.");
+    return 0 != (active_slots & (TMC_ONE_BIT << I));
+  }
+
+  /// 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`.
+  inline size_t active_bitset() const noexcept { return active_slots; }
+
+  /// 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 void fork(
+    Aw&& Awaitable, Exec&& Executor = tmc::current_executor(),
+    size_t Priority = tmc::current_priority()
+  ) {
+    decltype(auto) known = tmc::detail::into_known<false>(static_cast<Aw&&>(Awaitable));
+    using KnownAwaitable = std::remove_reference_t<decltype(known)>;
+    using KnownResult =
+      typename tmc::detail::get_awaitable_traits<KnownAwaitable>::result_type;
+    using SlotResult = std::tuple_element_t<I, ResultTuple>;
+    static_assert(
+      std::is_same_v<ResultStorage<KnownResult>, SlotResult>,
+      "Replacement awaitable must have the same result type as the original "
+      "slot."
+    );
+
+    constexpr size_t slotBit = TMC_ONE_BIT << I;
+    assert(
+      0 == (active_slots & slotBit) &&
+      "You may only fork a slot after its previous result has been "
+      "awaited."
+    );
+    active_slots |= slotBit;
+    ++remaining_count;
+
+    // Destroy the previously-consumed result before initiating the replacement. This
+    // prevents issues with results that own a resource which the replacement awaitable
+    // needs to re-acquire, such as zero-copy queue scopes.
+    std::get<I>(result) = SlotResult{};
+
+    tmc::ex_any* exec = tmc::detail::get_executor_traits<Exec>::type_erased(Executor);
+    auto continuationPriority = tmc::detail::this_thread::this_task().prio;
+
+    prepare_work(known, &std::get<I>(result), I, continuationPriority);
+
+    constexpr auto mode = tmc::detail::get_awaitable_traits<KnownAwaitable>::mode;
+    if constexpr (mode == tmc::detail::ASYNC_INITIATE) {
+      // Forward known with its original value category.
+      tmc::detail::get_awaitable_traits<KnownAwaitable>::async_initiate(
+        static_cast<decltype(known)&&>(known), exec, Priority
+      );
+    } else {
+      auto item = tmc::detail::into_initiate(static_cast<decltype(known)&&>(known));
+      tmc::detail::post_checked(exec, std::move(item), Priority);
+    }
+  }
+
+  /// This is a dummy awaitable. Don't store this in a variable.
+  /// For HALO to work, you must `co_await mux.fork_clang<I>()` immediately.
+  class TMC_CORO_AWAIT_ELIDABLE mux_tuple_fork_clang : tmc::detail::AwaitTagNoGroupAsIs {
+  public:
+    mux_tuple_fork_clang() {}
+
+    /// Never suspends.
+    bool await_ready() const noexcept { return true; }
+
+    /// Does nothing.
+    void await_suspend(std::coroutine_handle<>) noexcept {}
+
+    /// Does nothing.
+    void await_resume() noexcept {}
+  };
+
+  /// Similar to `fork<I>()` 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();
+  ///   }
+  /// }
+  /// ```
+  template <size_t I, typename Aw, typename Exec = tmc::ex_any*>
+  [[nodiscard("You must co_await fork_clang() immediately for HALO to be possible.")]]
+  mux_tuple_fork_clang fork_clang(
+    TMC_CORO_AWAIT_ELIDABLE_ARGUMENT Aw&& Awaitable,
+    Exec&& Executor = tmc::current_executor(), size_t Priority = tmc::current_priority()
+  ) {
+    fork<I>(static_cast<Aw&&>(Awaitable), static_cast<Exec&&>(Executor), Priority);
+    return mux_tuple_fork_clang{};
+  }
+
+// This must be awaited and all child awaitables completed before destruction.
+#ifndef NDEBUG
+  ~mux_tuple() noexcept {
+    assert(
+      remaining_count == 0 &&
+      "You must wait for all mux_tuple awaitables to complete before destroying it."
+    );
+  }
+#endif
+
+  // Not movable or copyable due to awaitables being initiated with pointers to this.
+  mux_tuple& operator=(const mux_tuple& other) = delete;
+  mux_tuple(const mux_tuple& other) = delete;
+  mux_tuple& operator=(mux_tuple&& other) = delete;
+  mux_tuple(mux_tuple&& other) = delete;
+};
+
+// Deduces the slots' result-type template parameters from the constructor
+// arguments. Each slot's type is the result the corresponding awaitable produces when
+// awaited.
+template <typename... Awaitable>
+  requires(sizeof...(Awaitable) != 0)
+mux_tuple(Awaitable&&...)
+  -> mux_tuple<typename tmc::detail::get_awaitable_traits<Awaitable>::result_type...>;
+
+} // namespace tmc

From c71ab519b22447b563b0c23d1f2559abf0f4489c Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 19:03:10 -0700
Subject: [PATCH 2/9] migrate select to use mux_tuple

---
 include/tmc/select.hpp | 99 +++++++++++++++++++++++++++++-------------
 1 file changed, 68 insertions(+), 31 deletions(-)

diff --git a/include/tmc/select.hpp b/include/tmc/select.hpp
index e3327fab..d9f237a9 100644
--- a/include/tmc/select.hpp
+++ b/include/tmc/select.hpp
@@ -7,7 +7,8 @@
 
 #include "tmc/detail/compat.hpp"
 #include "tmc/detail/concepts_awaitable.hpp" // IWYU pragma: keep
-#include "tmc/spawn_tuple.hpp"
+#include "tmc/detail/tuple_helpers.hpp"
+#include "tmc/mux_tuple.hpp"
 #include "tmc/task.hpp"
 #include "tmc/traits.hpp"
 
@@ -25,27 +26,29 @@ namespace detail {
 // storage.
 struct cancel_self_t {};
 
-// A nullary canceller that holds a cancellation target and invokes its
-// `.cancel()` method, forwarding whatever that method returns. `Target` is
-// instantiated as `forward_awaitable<T>`, so the target is owned by value when
-// it was passed as a movable rvalue, and held by reference (lvalue or
-// non-movable rvalue) otherwise. Produced by the object-taking constructor
-// (form 2) of `tmc::cancellable`. The forwarded result lets `select()` detect an
+// Wraps an object passed into `tmc::cancellable` constructor form 2 into a functor so
+// that it matches the behavior of form 1. The forwarded result lets `select()` detect an
 // awaitable `.cancel()` (an async cancel) and await it.
 template <typename Target> struct cancel_caller {
   Target target;
   decltype(auto) operator()() { return target.cancel(); }
 };
 
-// A trivial awaiter used by `select()` only in its async cancel branch for any sync
-// canceller mixed in. The fold expression cannot handle mixed await / no-await, so we
-// must provide an awaitable object for syntactic compatibility.
-struct cancel_noop : tmc::detail::AwaitTagNoGroupAsIs {
+// An awaitable that does nothing. Used by `select()` in the async cancel branch for any
+// sync canceller mixed in. The fold expression cannot handle mixed await / no-await, so
+// we must provide an awaitable object for syntactic compatibility.
+struct noop_awaitable : tmc::detail::AwaitTagNoGroupAsIs {
   static constexpr bool await_ready() noexcept { return true; }
   static void await_suspend(std::coroutine_handle<>) noexcept {}
   static void await_resume() noexcept {}
 };
 
+// A functor that does nothing. Used in the diagnostic `cancellable` overload so that
+// everything else compiles, and the only error is the human-readable static_assert.
+struct noop_func {
+  void operator()() const noexcept {}
+};
+
 // Forward-declared here so `tmc::cancellable` can grant it friendship.
 template <typename Awaitable, typename Canceller> consteval bool canceller_is_async();
 
@@ -60,7 +63,9 @@ template <typename Awaitable, typename Canceller> consteval bool canceller_is_as
 /// that cancels the operation when invoked. It may return void (a sync cancel) or
 /// an awaitable (an async cancel, which `select()` awaits). The canceller is
 /// stored by value and invoked only for losers, so an awaitable-returning canceller
-/// builds its awaitable only when the cancel is actually needed.
+/// builds its awaitable only when the cancel is actually needed. Therefore, it is safe to
+/// pass a capturing coroutine lambda here — it will be stored by value and outlive the
+/// awaited invocation.
 /// ```
 /// // sync cancel
 /// tmc::cancellable(timer.async_wait(tmc::aw_asio), [&]{ timer.cancel(); })
@@ -77,7 +82,7 @@ template <typename Awaitable, typename Canceller> consteval bool canceller_is_as
 /// 2. `tmc::cancellable(awaitable, object)` where `object` exposes a `.cancel()` method.
 /// The `.cancel()` method may be sync or async (awaitable).
 /// ```
-/// // sync cancel - timer.cancel() runs syncly (an Asio I/O object)
+/// // sync cancel - timer.cancel() runs synchronously (an Asio I/O object)
 /// tmc::cancellable(timer.async_wait(tmc::aw_asio), timer)
 ///
 /// // async cancel - op.cancel() returns an awaitable, which select() awaits
@@ -90,7 +95,7 @@ template <typename Awaitable, typename Canceller> consteval bool canceller_is_as
 /// automatically await it if needed. This constructor only accepts lvalues or non-movable
 /// rvalues. Movable rvalues are rejected to prevent lifetime issues.
 /// ```
-/// // sync cancel - op.cancel() runs syncly
+/// // sync cancel - op.cancel() runs synchronously
 /// tmc::cancellable(op)
 ///
 /// // async cancel - op2.cancel() returns an awaitable, which select() awaits
@@ -121,9 +126,12 @@ template <typename Awaitable, typename Canceller> class cancellable {
 public:
   /// `tmc::cancellable(awaitable, canceller)` where `canceller` is a nullary functor
   /// that cancels the operation when invoked. It may return void (a sync cancel)
-  /// or an awaitable (an async cancel, which `select()` awaits). The canceller is
-  /// stored by value and invoked only for losers, so an awaitable-returning canceller
-  /// builds its awaitable only when the cancel is actually needed.
+  /// or an awaitable (an async cancel, which `select()` awaits).
+  ///
+  /// The canceller is stored by value and invoked only for losers, so an
+  /// awaitable-returning canceller builds its awaitable only when the cancel is actually
+  /// needed. Therefore, it is safe to pass a capturing coroutine lambda here — it will be
+  /// stored by value and outlive the awaited invocation.
   /// ```
   /// // sync cancel
   /// tmc::cancellable(timer.async_wait(tmc::aw_asio), [&]{ timer.cancel(); })
@@ -145,7 +153,7 @@ template <typename Awaitable, typename Canceller> class cancellable {
   /// `tmc::cancellable(awaitable, object)` where `object` exposes a `.cancel()`
   /// method. The `.cancel()` method may be sync or async (awaitable).
   /// ```
-  /// // sync cancel - timer.cancel() runs syncly (an Asio I/O object)
+  /// // sync cancel - timer.cancel() runs synchronously (an Asio I/O object)
   /// tmc::cancellable(timer.async_wait(tmc::aw_asio), timer)
   ///
   /// // async cancel - op.cancel() returns an awaitable, which select() awaits
@@ -168,6 +176,24 @@ template <typename Awaitable, typename Canceller> class cancellable {
              requires(A& Obj) { Obj.cancel(); } &&
              std::is_reference_v<tmc::detail::forward_awaitable<A>>)
   explicit cancellable(A&& Obj) : awaitable(static_cast<A&&>(Obj)), canceller{} {}
+
+  /// This overload is invalid; it exists only to provide a useful compilation
+  /// error. It rejects passing an awaitable directly as the canceller (the second
+  /// argument). The canceller should instead be a nullary functor that returns the
+  /// awaitable, e.g. `[&]{ return op.async_cancel(); }`. It will be lazily invoked as
+  /// needed.
+  template <typename A, typename C>
+    requires(tmc::traits::executable_kind_v<std::decay_t<C>> ==
+             tmc::traits::executable_kind::AWAITABLE)
+  cancellable(A&& Aw, C&&) : awaitable(static_cast<A&&>(Aw)), canceller{} {
+    static_assert(
+      tmc::traits::executable_kind_v<std::decay_t<C>> !=
+        tmc::traits::executable_kind::AWAITABLE,
+      "The second argument to tmc::cancellable() should not be an awaitable. "
+      "Instead, pass a nullary functor that returns the awaitable, e.g. "
+      "[&]{ return op.async_cancel(); }. It will be lazily invoked as needed."
+    );
+  }
 };
 
 // Deduction guides pick the storage types that the matching constructor fills:
@@ -201,9 +227,19 @@ template <typename A>
 cancellable(A&&)
   -> cancellable<tmc::detail::forward_awaitable<A>, tmc::detail::cancel_self_t>;
 
+// Diagnostic guide: routes the "awaitable passed as the canceller" mistake to
+// the diagnostic constructor, which emits a readable static_assert. Uses a valid no-op
+// canceller type so that the only error emitted is the static_assert.
+template <typename A, typename C>
+  requires(
+    tmc::traits::executable_kind_v<std::decay_t<C>> ==
+    tmc::traits::executable_kind::AWAITABLE
+  )
+cancellable(A&&, C&&)
+  -> cancellable<tmc::detail::forward_awaitable<A>, tmc::detail::noop_func>;
+
 namespace detail {
-// Compile-time check whether the canceller of a `tmc::cancellable` is
-// async (returns an awaitable rather than performing the cancellation directly).
+// Compile-time check whether the canceller of a `tmc::cancellable` returns an awaitable.
 template <typename Awaitable, typename Canceller> consteval bool canceller_is_async() {
   using Pair = tmc::cancellable<Awaitable, Canceller>;
   if constexpr (std::is_same_v<Canceller, cancel_self_t>) {
@@ -249,23 +285,24 @@ select(tmc::cancellable<Awaitable, Canceller>... Pairs) {
   constexpr size_t Count = sizeof...(Awaitable);
 
   // Forward each awaitable with its original value category so the awaitable's own
-  // lvalue/rvalue qualification is respected.
-  auto each =
-    tmc::spawn_tuple(static_cast<Awaitable&&>(Pairs.awaitable)...).result_each();
+  // lvalue/rvalue qualification is respected. The mux_tuple makes each result
+  // available as it becomes ready, so we can return as soon as the first one
+  // completes.
+  tmc::mux_tuple mux(static_cast<Awaitable&&>(Pairs.awaitable)...);
 
   // Wait for at least one operation to complete.
-  size_t winner = co_await each;
+  size_t winner = co_await mux;
 
   // Move the winner's result into the variant slot for its index.
   std::optional<variant_type> result;
   auto storeWinner = [&]<size_t I>(std::integral_constant<size_t, I>) {
     using VarElem = std::variant_alternative_t<I, variant_type>;
-    using Stored = std::remove_reference_t<decltype(each.template get<I>())>;
+    using Stored = std::remove_reference_t<decltype(mux.template get<I>())>;
     if constexpr (std::is_same_v<Stored, VarElem>) {
-      result.emplace(std::in_place_index<I>, std::move(each.template get<I>()));
+      result.emplace(std::in_place_index<I>, std::move(mux.template get<I>()));
     } else {
       // Non-default-constructible results are stored wrapped in std::optional.
-      result.emplace(std::in_place_index<I>, std::move(*each.template get<I>()));
+      result.emplace(std::in_place_index<I>, std::move(*mux.template get<I>()));
     }
   };
   [&]<size_t... I>(std::index_sequence<I...>) {
@@ -305,14 +342,14 @@ select(tmc::cancellable<Awaitable, Canceller>... Pairs) {
           return Pair.awaitable.cancel();
         } else {
           Pair.awaitable.cancel();
-          return tmc::detail::cancel_noop{};
+          return tmc::detail::noop_awaitable{};
         }
       } else {
         if constexpr (tmc::traits::is_awaitable<decltype(Pair.canceller())>) {
           return Pair.canceller();
         } else {
           Pair.canceller();
-          return tmc::detail::cancel_noop{};
+          return tmc::detail::noop_awaitable{};
         }
       }
     };
@@ -320,8 +357,8 @@ select(tmc::cancellable<Awaitable, Canceller>... Pairs) {
   }
 
   // Drain the remaining (now-cancelled) awaitables. Their results are
-  // discarded, but they must all complete before `each` is destroyed.
-  for (size_t i = co_await each; i != each.end(); i = co_await each) {
+  // discarded, but they must all complete before `mux` is destroyed.
+  for (size_t i = co_await mux; i != mux.end(); i = co_await mux) {
     // discard
   }
 

From 90aa06fc611f81c5f6c61e6e5adf8da0595001d1 Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 19:03:59 -0700
Subject: [PATCH 3/9] remove result_each()

---
 include/tmc/detail/result_each.hpp |   5 +-
 include/tmc/detail/result_each.ipp |  46 +--
 include/tmc/spawn_group.hpp        |   2 +-
 include/tmc/spawn_many.hpp         | 453 ++++++++++-------------------
 include/tmc/spawn_tuple.hpp        | 264 +++--------------
 5 files changed, 196 insertions(+), 574 deletions(-)

diff --git a/include/tmc/detail/result_each.hpp b/include/tmc/detail/result_each.hpp
index 9d9e2c78..845f5c5b 100644
--- a/include/tmc/detail/result_each.hpp
+++ b/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(
diff --git a/include/tmc/detail/result_each.ipp b/include/tmc/detail/result_each.ipp
index 37c753b2..982b5d94 100644
--- a/include/tmc/detail/result_each.ipp
+++ b/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,37 +17,30 @@
 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.
+    // 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 by holding the EACH bit (lock bit).
+    //
+    // 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
-      );
+      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
-      );
+      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
       }
@@ -57,17 +48,8 @@ bool result_each_await_suspend(
       // 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)) {
-    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;
-  }
+  // We got a result.
+  return false;
 }
 
 size_t result_each_await_resume(
diff --git a/include/tmc/spawn_group.hpp b/include/tmc/spawn_group.hpp
index 5a550a4a..c2cbbb1e 100644
--- a/include/tmc/spawn_group.hpp
+++ b/include/tmc/spawn_group.hpp
@@ -170,7 +170,7 @@ class [[nodiscard(
   }
 
   /// Initiates all of the wrapped awaitables and waits for them to complete.
-  aw_spawn_many_impl<Result, MaxCount, false, false>
+  aw_spawn_many_impl<Result, MaxCount, false>
   operator co_await() && noexcept {
     // spawn_group can be awaited in a different context than where it was
     // created. Therefore, capture the continuation_executor at the await point,
diff --git a/include/tmc/spawn_many.hpp b/include/tmc/spawn_many.hpp
index ed3fdc9f..6def2c14 100644
--- a/include/tmc/spawn_many.hpp
+++ b/include/tmc/spawn_many.hpp
@@ -5,12 +5,10 @@
 
 #pragma once
 
-#include "tmc/detail/awaitable_customizer.hpp"
 #include "tmc/detail/compat.hpp"
 #include "tmc/detail/concepts_awaitable.hpp" // IWYU pragma: keep
 #include "tmc/detail/concepts_work_item.hpp"
 #include "tmc/detail/mixins.hpp"
-#include "tmc/detail/result_each.hpp"
 #include "tmc/detail/task_unsafe.hpp"
 #include "tmc/detail/thread_locals.hpp"
 #include "tmc/ex_any.hpp"
@@ -29,8 +27,7 @@ namespace tmc {
 /// The customizable task wrapper / awaitable type returned by
 /// `tmc::spawn_many()` / `tmc::spawn_func_many()`.
 template <
-  typename Result, size_t Count, typename IterBegin, typename IterEnd,
-  bool IsFunc>
+  typename Result, size_t Count, typename IterBegin, typename IterEnd, bool IsFunc>
 class aw_spawn_many;
 
 /// The single-argument form of spawn_many() has two overloads.
@@ -55,7 +52,13 @@ template <
   typename Result = tmc::detail::awaitable_result_t<Awaitable>>
 aw_spawn_many<Result, Count, AwaitableIter, size_t, false>
 spawn_many(AwaitableIter&& Begin)
-  requires(Count != 0)
+  requires(
+    Count != 0 &&
+    requires(AwaitableIter a) {
+      ++a;
+      *a;
+    }
+  )
 {
   return aw_spawn_many<Result, Count, AwaitableIter, size_t, false>(
     std::forward<AwaitableIter>(Begin), 0, 0
@@ -110,7 +113,12 @@ template <
   typename AwaitableIter, typename Awaitable = std::iter_value_t<AwaitableIter>,
   typename Result = tmc::detail::awaitable_result_t<Awaitable>>
 aw_spawn_many<Result, 0, AwaitableIter, size_t, false>
-spawn_many(AwaitableIter&& Begin, size_t TaskCount) {
+spawn_many(AwaitableIter&& Begin, size_t TaskCount)
+  requires(requires(AwaitableIter a) {
+    ++a;
+    *a;
+  })
+{
   return aw_spawn_many<Result, 0, AwaitableIter, size_t, false>(
     std::forward<AwaitableIter>(Begin), TaskCount, 0
   );
@@ -144,10 +152,15 @@ template <
   typename Awaitable = std::iter_value_t<AwaitableIter>,
   typename Result = tmc::detail::awaitable_result_t<Awaitable>>
 aw_spawn_many<Result, MaxCount, AwaitableIter, AwaitableIter, false>
-spawn_many(AwaitableIter&& Begin, AwaitableIter&& End) {
+spawn_many(AwaitableIter&& Begin, AwaitableIter&& End)
+  requires(requires(AwaitableIter a, AwaitableIter b) {
+    ++a;
+    *a;
+    a != b;
+  })
+{
   return aw_spawn_many<Result, MaxCount, AwaitableIter, AwaitableIter, false>(
-    std::forward<AwaitableIter>(Begin), std::forward<AwaitableIter>(End),
-    TMC_ALL_ONES
+    std::forward<AwaitableIter>(Begin), std::forward<AwaitableIter>(End), TMC_ALL_ONES
   );
 }
 
@@ -172,10 +185,15 @@ template <
   typename AwaitableIter, typename Awaitable = std::iter_value_t<AwaitableIter>,
   typename Result = tmc::detail::awaitable_result_t<Awaitable>>
 aw_spawn_many<Result, 0, AwaitableIter, AwaitableIter, false>
-spawn_many(AwaitableIter&& Begin, AwaitableIter&& End, size_t MaxCount) {
+spawn_many(AwaitableIter&& Begin, AwaitableIter&& End, size_t MaxCount)
+  requires(requires(AwaitableIter a, AwaitableIter b) {
+    ++a;
+    *a;
+    a != b;
+  })
+{
   return aw_spawn_many<Result, 0, AwaitableIter, AwaitableIter, false>(
-    std::forward<AwaitableIter>(Begin), std::forward<AwaitableIter>(End),
-    MaxCount
+    std::forward<AwaitableIter>(Begin), std::forward<AwaitableIter>(End), MaxCount
   );
 }
 
@@ -194,12 +212,17 @@ spawn_many(AwaitableIter&& Begin, AwaitableIter&& End, size_t MaxCount) {
 /// been `co_await` ed. A temporary iterator is OK only if this
 /// is `co_await` ed immediately.
 template <
-  size_t Count = 0, typename FuncIter,
-  typename Functor = std::iter_value_t<FuncIter>,
+  size_t Count = 0, typename FuncIter, typename Functor = std::iter_value_t<FuncIter>,
   typename Result = std::invoke_result_t<Functor&>>
 aw_spawn_many<Result, Count, FuncIter, size_t, true>
 spawn_func_many(FuncIter&& FunctorIterator)
-  requires(Count != 0)
+  requires(
+    Count != 0 &&
+    requires(FuncIter a) {
+      ++a;
+      *a;
+    }
+  )
 {
   return aw_spawn_many<Result, Count, FuncIter, size_t, true>(
     std::forward<FuncIter>(FunctorIterator), 0, 0
@@ -226,8 +249,7 @@ template <
   typename FuncIter = tmc::detail::range_iter<FuncRange>::type,
   typename Functor = std::iter_value_t<FuncIter>,
   typename Result = std::invoke_result_t<Functor&>>
-aw_spawn_many<Result, 0, FuncIter, FuncIter, true>
-spawn_func_many(FuncRange&& Range)
+aw_spawn_many<Result, 0, FuncIter, FuncIter, true> spawn_func_many(FuncRange&& Range)
   requires(Count == 0)
 {
   return aw_spawn_many<Result, 0, FuncIter, FuncIter, true>(
@@ -250,7 +272,12 @@ template <
   typename FuncIter, typename Functor = std::iter_value_t<FuncIter>,
   typename Result = std::invoke_result_t<Functor&>>
 aw_spawn_many<Result, 0, FuncIter, size_t, true>
-spawn_func_many(FuncIter&& FunctorIterator, size_t FunctorCount) {
+spawn_func_many(FuncIter&& FunctorIterator, size_t FunctorCount)
+  requires(requires(FuncIter a) {
+    ++a;
+    *a;
+  })
+{
   return aw_spawn_many<Result, 0, FuncIter, size_t, true>(
     std::forward<FuncIter>(FunctorIterator), FunctorCount, 0
   );
@@ -277,11 +304,16 @@ spawn_func_many(FuncIter&& FunctorIterator, size_t FunctorCount) {
 /// been `co_await` ed.
 ///
 template <
-  size_t MaxCount = 0, typename FuncIter,
-  typename Functor = std::iter_value_t<FuncIter>,
+  size_t MaxCount = 0, typename FuncIter, typename Functor = std::iter_value_t<FuncIter>,
   typename Result = std::invoke_result_t<Functor&>>
 aw_spawn_many<Result, MaxCount, FuncIter, FuncIter, true>
-spawn_func_many(FuncIter&& Begin, FuncIter&& End) {
+spawn_func_many(FuncIter&& Begin, FuncIter&& End)
+  requires(requires(FuncIter a, FuncIter b) {
+    ++a;
+    *a;
+    a != b;
+  })
+{
   return aw_spawn_many<Result, MaxCount, FuncIter, FuncIter, true>(
     std::forward<FuncIter>(Begin), std::forward<FuncIter>(End), TMC_ALL_ONES
   );
@@ -305,25 +337,28 @@ template <
   typename FuncIter, typename Functor = std::iter_value_t<FuncIter>,
   typename Result = std::invoke_result_t<Functor&>>
 aw_spawn_many<Result, 0, FuncIter, FuncIter, true>
-spawn_func_many(FuncIter&& Begin, FuncIter&& End, size_t MaxCount) {
+spawn_func_many(FuncIter&& Begin, FuncIter&& End, size_t MaxCount)
+  requires(requires(FuncIter a, FuncIter b) {
+    ++a;
+    *a;
+    a != b;
+  })
+{
   return aw_spawn_many<Result, 0, FuncIter, FuncIter, true>(
     std::forward<FuncIter>(Begin), std::forward<FuncIter>(End), MaxCount
   );
 }
 
-template <typename Result, size_t Count, bool IsEach, bool IsFunc>
-class aw_spawn_many_impl {
+template <typename Result, size_t Count, bool IsFunc> class aw_spawn_many_impl {
 public:
-  // result_each does not symmetric transfer, so we store the
-  // count of consumed results here instead
-  std::conditional_t<IsEach, ptrdiff_t, std::coroutine_handle<>>
-    remaining_count_or_symmetric_task;
+  // If the last task is to be run immediately via symmetric transfer, it is
+  // stored here instead of being posted to the executor.
+  std::coroutine_handle<> symmetric_task;
   std::coroutine_handle<> continuation;
   tmc::ex_any* continuation_executor;
   // the atomic synchronization variable that coordinates between this and
   // awaitable_customizer (task's final_suspend)
-  std::conditional_t<IsEach, std::atomic<size_t>, std::atomic<ptrdiff_t>>
-    sync_flags_or_done_count;
+  std::atomic<ptrdiff_t> done_count;
 
   struct empty {};
   using ResultArray = std::conditional_t<
@@ -333,51 +368,22 @@ class aw_spawn_many_impl {
       std::array<tmc::detail::result_storage_t<Result>, Count>>>;
   TMC_NO_UNIQUE_ADDRESS ResultArray result_arr;
 
-  template <typename, size_t, typename, typename, bool>
-  friend class aw_spawn_many;
-
-  // When result_each() is called, tasks are synchronized via an atomic bitmask
-  // with only 63 (or 31, on 32-bit) slots for tasks. result_each() doesn't seem
-  // like a good fit for larger task groups anyway. If you really need more
-  // room, please open a GitHub issue explaining why...
-  static_assert(!IsEach || Count < TMC_PLATFORM_BITS);
+  template <typename, size_t, typename, typename, bool> friend class aw_spawn_many;
 
   // Prepares the work item but does not initiate it.
-  template <typename T>
-  TMC_FORCE_INLINE inline void
-  prepare_work(T& Task, size_t Idx, [[maybe_unused]] size_t ContinuationPrio) {
+  template <typename T> TMC_FORCE_INLINE inline void prepare_work(T& Task, size_t Idx) {
     tmc::detail::get_awaitable_traits<T>::set_continuation(Task, &continuation);
     tmc::detail::get_awaitable_traits<T>::set_continuation_executor(
       Task, &continuation_executor
     );
-    tmc::detail::get_awaitable_traits<T>::set_done_count(
-      Task, &sync_flags_or_done_count
-    );
-    if constexpr (IsEach) {
-      tmc::detail::get_awaitable_traits<T>::set_flags(
-        Task, tmc::detail::task_flags::EACH |
-                (Idx << tmc::detail::task_flags::TASKNUM_LOW_OFF) |
-                ContinuationPrio
-      );
-    }
+    tmc::detail::get_awaitable_traits<T>::set_done_count(Task, &done_count);
     if constexpr (!std::is_void_v<Result>) {
-      tmc::detail::get_awaitable_traits<T>::set_result_ptr(
-        Task, &result_arr[Idx]
-      );
+      tmc::detail::get_awaitable_traits<T>::set_result_ptr(Task, &result_arr[Idx]);
     }
   }
 
   void set_done_count(size_t NumTasks) {
-    if constexpr (IsEach) {
-      remaining_count_or_symmetric_task = static_cast<ptrdiff_t>(NumTasks);
-      sync_flags_or_done_count.store(
-        tmc::detail::task_flags::EACH, std::memory_order_release
-      );
-    } else {
-      sync_flags_or_done_count.store(
-        static_cast<ptrdiff_t>(NumTasks), std::memory_order_release
-      );
-    }
+    done_count.store(static_cast<ptrdiff_t>(NumTasks), std::memory_order_release);
   }
 
   template <typename TaskIter>
@@ -385,42 +391,35 @@ class aw_spawn_many_impl {
     TaskIter Iter, size_t TaskCount, tmc::ex_any* Executor,
     tmc::ex_any* ContinuationExecutor, size_t Prio, bool DoSymmetricTransfer
   )
-      : continuation_executor{ContinuationExecutor} {
-    if constexpr (!IsEach) {
-      remaining_count_or_symmetric_task = nullptr;
-    }
-
+    requires(requires(TaskIter a) {
+              ++a;
+              *a;
+            })
+      : symmetric_task{nullptr}, continuation_executor{ContinuationExecutor} {
     // Wrap unknown (WRAPPER) awaitables into work_items (tasks). Preserve the
     // type of known awaitables.
     using Awaitable = std::conditional_t<
-      IsFunc, tmc::task<Result>,
-      std::remove_cvref_t<std::iter_value_t<TaskIter>>>;
+      IsFunc, tmc::task<Result>, std::remove_cvref_t<std::iter_value_t<TaskIter>>>;
 
     constexpr auto mode = tmc::detail::get_awaitable_traits<Awaitable>::mode;
     static_assert(
       mode != tmc::detail::UNKNOWN, "This doesn't appear to be an awaitable."
     );
 
-    using WorkItem = std::conditional_t<
-      mode == tmc::detail::ASYNC_INITIATE, Awaitable, work_item>;
-    using WorkItemArray = std::conditional_t<
-      Count == 0, std::vector<WorkItem>, std::array<WorkItem, Count>>;
+    using WorkItem =
+      std::conditional_t<mode == tmc::detail::ASYNC_INITIATE, Awaitable, work_item>;
+    using WorkItemArray =
+      std::conditional_t<Count == 0, std::vector<WorkItem>, std::array<WorkItem, Count>>;
 
     size_t size;
     if constexpr (Count != 0) {
       size = Count;
     } else {
       size = TaskCount;
-      if constexpr (IsEach) {
-        if (size > TMC_PLATFORM_BITS - 1) {
-          size = TMC_PLATFORM_BITS - 1;
-        }
-      }
       if constexpr (!std::is_void_v<Result>) {
         result_arr.resize(size);
       }
     }
-    size_t continuationPriority = tmc::detail::this_thread::this_task().prio;
 
     if constexpr (mode == tmc::detail::ASYNC_INITIATE) {
       // ASYNC_INITIATE types may possibly not be stored in a vector or array
@@ -434,7 +433,7 @@ class aw_spawn_many_impl {
         TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
         auto t = std::move(*Iter);
         TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
-        prepare_work(t, i, continuationPriority);
+        prepare_work(t, i);
         tmc::detail::get_awaitable_traits<Awaitable>::async_initiate(
           std::move(t), Executor, Prio
         );
@@ -453,7 +452,7 @@ class aw_spawn_many_impl {
         TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
         auto t = tmc::detail::into_known<IsFunc>(std::move(*Iter));
         TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
-        prepare_work(t, i, continuationPriority);
+        prepare_work(t, i);
         taskArr[i] = tmc::detail::into_initiate(std::move(t));
         ++Iter;
       }
@@ -466,11 +465,8 @@ class aw_spawn_many_impl {
       auto postCount = DoSymmetricTransfer ? size - 1 : size;
       set_done_count(postCount);
 
-      if constexpr (!IsEach) {
-        if (DoSymmetricTransfer) {
-          remaining_count_or_symmetric_task =
-            TMC_WORK_ITEM_AS_STD_CORO(taskArr[size - 1]);
-        }
+      if (DoSymmetricTransfer) {
+        symmetric_task = TMC_WORK_ITEM_AS_STD_CORO(taskArr[size - 1]);
       }
       tmc::detail::post_bulk_checked(Executor, taskArr.data(), postCount, Prio);
     }
@@ -482,25 +478,16 @@ class aw_spawn_many_impl {
     tmc::ex_any* ContinuationExecutor, size_t Prio, bool DoSymmetricTransfer
   )
     requires(requires(TaskIter a, TaskIter b) {
-      ++a;
-      *a;
-      a != b;
-    })
-      : continuation_executor{ContinuationExecutor} {
-    if constexpr (!IsEach) {
-      remaining_count_or_symmetric_task = nullptr;
-    }
-
+              ++a;
+              *a;
+              a != b;
+            })
+      : symmetric_task{nullptr}, continuation_executor{ContinuationExecutor} {
     size_t size;
     if constexpr (Count != 0) {
       size = Count;
     } else {
       size = MaxCount;
-      if constexpr (IsEach) {
-        if (size > TMC_PLATFORM_BITS - 1) {
-          size = TMC_PLATFORM_BITS - 1;
-        }
-      }
       if constexpr (requires(TaskIter a, TaskIter b) { a - b; }) {
         // Caller didn't specify capacity to preallocate, but we can calculate
         size_t iterSize = static_cast<size_t>(End - Begin);
@@ -517,20 +504,17 @@ class aw_spawn_many_impl {
     // type of known awaitables.
 
     using Awaitable = std::conditional_t<
-      IsFunc, tmc::task<Result>,
-      std::remove_cvref_t<std::iter_value_t<TaskIter>>>;
+      IsFunc, tmc::task<Result>, std::remove_cvref_t<std::iter_value_t<TaskIter>>>;
 
     constexpr auto mode = tmc::detail::get_awaitable_traits<Awaitable>::mode;
     static_assert(
       mode != tmc::detail::UNKNOWN, "This doesn't appear to be an awaitable."
     );
 
-    using WorkItem = std::conditional_t<
-      mode == tmc::detail::ASYNC_INITIATE, Awaitable, work_item>;
-    using WorkItemArray = std::conditional_t<
-      Count == 0, std::vector<WorkItem>, std::array<WorkItem, Count>>;
-
-    size_t continuationPriority = tmc::detail::this_thread::this_task().prio;
+    using WorkItem =
+      std::conditional_t<mode == tmc::detail::ASYNC_INITIATE, Awaitable, work_item>;
+    using WorkItemArray =
+      std::conditional_t<Count == 0, std::vector<WorkItem>, std::array<WorkItem, Count>>;
 
     // Collect and prepare the tasks
     size_t taskCount = 0;
@@ -550,7 +534,7 @@ class aw_spawn_many_impl {
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
           auto t = std::move(*Begin);
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
-          prepare_work(t, taskCount, continuationPriority);
+          prepare_work(t, taskCount);
           tmc::detail::get_awaitable_traits<Awaitable>::async_initiate(
             std::move(t), Executor, Prio
           );
@@ -569,7 +553,7 @@ class aw_spawn_many_impl {
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
           auto t = tmc::detail::into_known<IsFunc>(std::move(*Begin));
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
-          prepare_work(t, taskCount, continuationPriority);
+          prepare_work(t, taskCount);
           taskArr[taskCount] = tmc::detail::into_initiate(std::move(t));
           ++Begin;
           ++taskCount;
@@ -590,15 +574,10 @@ class aw_spawn_many_impl {
         } else {
           auto postCount = DoSymmetricTransfer ? taskCount - 1 : taskCount;
           set_done_count(postCount);
-          if constexpr (!IsEach) {
-            if (DoSymmetricTransfer) {
-              remaining_count_or_symmetric_task =
-                TMC_WORK_ITEM_AS_STD_CORO(taskArr[taskCount - 1]);
-            }
+          if (DoSymmetricTransfer) {
+            symmetric_task = TMC_WORK_ITEM_AS_STD_CORO(taskArr[taskCount - 1]);
           }
-          tmc::detail::post_bulk_checked(
-            Executor, taskArr.data(), postCount, Prio
-          );
+          tmc::detail::post_bulk_checked(Executor, taskArr.data(), postCount, Prio);
         }
       }
     } else {
@@ -610,16 +589,13 @@ class aw_spawn_many_impl {
       // vector so that they can be configured afterward. If the awaitable
       // type is not copy-constructible, this will not compile.
       if constexpr (mode == tmc::detail::TMC_TASK ||
-                    mode == tmc::detail::ASYNC_INITIATE ||
-                    mode == tmc::detail::WRAPPER) {
+                    mode == tmc::detail::ASYNC_INITIATE || mode == tmc::detail::WRAPPER) {
         // These types can be processed using a single vector
         WorkItemArray taskArr;
         while (Begin != End && taskCount < size) {
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
           taskArr.emplace_back(
-            tmc::detail::into_initiate(
-              tmc::detail::into_known<IsFunc>(std::move(*Begin))
-            )
+            tmc::detail::into_initiate(tmc::detail::into_known<IsFunc>(std::move(*Begin)))
           );
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
           ++Begin;
@@ -636,12 +612,12 @@ class aw_spawn_many_impl {
         }
         for (size_t i = 0; i < taskCount; ++i) {
           if constexpr (mode == tmc::detail::ASYNC_INITIATE) {
-            prepare_work(taskArr[i], i, continuationPriority);
+            prepare_work(taskArr[i], i);
           } else { // TMC_TASK or WRAPPER
             auto t = tmc::detail::task_unsafe<Result>::from_address(
               TMC_WORK_ITEM_AS_STD_CORO(taskArr[i]).address()
             );
-            prepare_work(t, i, continuationPriority);
+            prepare_work(t, i);
           }
         }
 
@@ -660,15 +636,10 @@ class aw_spawn_many_impl {
         } else {
           auto postCount = DoSymmetricTransfer ? taskCount - 1 : taskCount;
           set_done_count(postCount);
-          if constexpr (!IsEach) {
-            if (DoSymmetricTransfer) {
-              remaining_count_or_symmetric_task =
-                TMC_WORK_ITEM_AS_STD_CORO(taskArr[taskCount - 1]);
-            }
+          if (DoSymmetricTransfer) {
+            symmetric_task = TMC_WORK_ITEM_AS_STD_CORO(taskArr[taskCount - 1]);
           }
-          tmc::detail::post_bulk_checked(
-            Executor, taskArr.data(), postCount, Prio
-          );
+          tmc::detail::post_bulk_checked(Executor, taskArr.data(), postCount, Prio);
         }
       } else if constexpr (mode == tmc::detail::COROUTINE) {
         // These types must be stored in a separate vector that preserves the
@@ -697,16 +668,14 @@ class aw_spawn_many_impl {
           result_arr.resize(taskCount);
         }
         for (size_t i = 0; i < taskCount; ++i) {
-          prepare_work(originalCoroArr[i], i, continuationPriority);
+          prepare_work(originalCoroArr[i], i);
         }
 
         // Initiate the tasks
         auto postCount = DoSymmetricTransfer ? taskCount - 1 : taskCount;
         set_done_count(postCount);
-        if constexpr (!IsEach) {
-          if (DoSymmetricTransfer) {
-            remaining_count_or_symmetric_task = originalCoroArr[taskCount - 1];
-          }
+        if (DoSymmetricTransfer) {
+          symmetric_task = originalCoroArr[taskCount - 1];
         }
 
         std::array<tmc::work_item, 64> workItemArr;
@@ -714,53 +683,41 @@ class aw_spawn_many_impl {
         while (totalCount < postCount) {
           size_t submitCount = 0;
           while (submitCount < workItemArr.size() && totalCount < postCount) {
-            workItemArr[submitCount] = tmc::detail::into_initiate(
-              std::move(originalCoroArr[totalCount])
-            );
+            workItemArr[submitCount] =
+              tmc::detail::into_initiate(std::move(originalCoroArr[totalCount]));
             ++totalCount;
             ++submitCount;
           }
-          tmc::detail::post_bulk_checked(
-            Executor, workItemArr.data(), submitCount, Prio
-          );
+          tmc::detail::post_bulk_checked(Executor, workItemArr.data(), submitCount, Prio);
         }
       }
     }
   }
 
 public:
-  /*** SUPPORTS REGULAR AWAIT ***/
   /// Always suspends.
-  inline bool await_ready() const noexcept
-    requires(!IsEach)
-  {
+  inline bool await_ready() const noexcept {
     // Always suspends, due to the possibility to resume on another executor.
     return false;
   }
 
   /// Suspends the outer coroutine, submits the wrapped task to the
   /// executor, and waits for it to complete.
-  inline std::coroutine_handle<>
-  await_suspend(std::coroutine_handle<> Outer) noexcept
-    requires(!IsEach)
-  {
+  inline std::coroutine_handle<> await_suspend(std::coroutine_handle<> Outer) noexcept {
 #ifndef NDEBUG
-    assert(
-      sync_flags_or_done_count.load() >= 0 && "You may only co_await this once."
-    );
+    assert(done_count.load() >= 0 && "You may only co_await this once.");
 #endif
     continuation = Outer;
     std::coroutine_handle<> next;
-    if (remaining_count_or_symmetric_task != nullptr) {
+    if (symmetric_task != nullptr) {
       // symmetric transfer to the last task IF it should run immediately
-      next = remaining_count_or_symmetric_task;
+      next = symmetric_task;
     } else {
       // 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.
-      auto remaining =
-        sync_flags_or_done_count.fetch_sub(1, std::memory_order_acq_rel);
+      auto remaining = done_count.fetch_sub(1, std::memory_order_acq_rel);
       // No symmetric transfer - all tasks were already posted.
       // Suspend if remaining > 0 (task is still running)
       if (remaining > 0) {
@@ -786,89 +743,21 @@ class aw_spawn_many_impl {
   /// `std::array<Result, Count>`. If `Count` is a runtime parameter, returns
   /// a `std::vector<Result>` with capacity `Count`. If `Result` is not
   /// default-constructible, it will be wrapped in an optional.
-  TMC_AWAIT_RESUME inline std::add_rvalue_reference_t<ResultArray>
-  await_resume() noexcept
-    requires(!IsEach && !std::is_void_v<Result>)
+  TMC_AWAIT_RESUME inline std::add_rvalue_reference_t<ResultArray> await_resume() noexcept
+    requires(!std::is_void_v<Result>)
   {
     return std::move(result_arr);
   }
 
   /// Does nothing.
   inline void await_resume() noexcept
-    requires(!IsEach && std::is_void_v<Result>)
-  {}
-  /*** END REGULAR AWAIT ***/
-
-  /*** SUPPORTS EACH() ***/
-  /// Suspends if there are no ready results.
-  inline bool await_ready() const noexcept
-    requires(IsEach)
-  {
-    return tmc::detail::result_each_await_ready();
-  }
-
-  /// Suspends if there are no ready results.
-  inline bool await_suspend(std::coroutine_handle<> Outer) noexcept
-    requires(IsEach)
-  {
-    return tmc::detail::result_each_await_suspend(
-      remaining_count_or_symmetric_task, Outer, continuation,
-      continuation_executor, sync_flags_or_done_count
-    );
-  }
-
-  /// Returns the index of the current ready result. The result indexes
-  /// correspond to the indexes of the originally submitted tasks. Results may
-  /// become ready in any order, but when awaited repeatedly, each index from
-  /// `[0..task_count)` will be returned exactly once. When there are no
-  /// more results to be returned, the returned index will be equal to
-  /// `end()`.
-  TMC_AWAIT_RESUME inline size_t await_resume() noexcept
-    requires(IsEach)
-  {
-    return tmc::detail::result_each_await_resume(
-      remaining_count_or_symmetric_task, sync_flags_or_done_count
-    );
-  }
-
-  /// Provides a sentinel value that can be compared against the value
-  /// returned from co_await.
-  inline size_t end() noexcept
-    requires(IsEach)
-  {
-    return 64;
-  }
-
-  // Gets the ready result at the given index.
-  inline std::add_lvalue_reference_t<tmc::detail::result_storage_t<Result>>
-  operator[](size_t idx) noexcept
-    requires(IsEach && !std::is_void_v<Result>)
-  {
-    assert(idx < result_arr.size());
-    return result_arr[idx];
-  }
-
-  // Provided for convenience only - to expose the same API as the
-  // Result-returning awaitable version. Does nothing.
-  inline void operator[]([[maybe_unused]] size_t idx) noexcept
-    requires(IsEach && std::is_void_v<Result>)
+    requires(std::is_void_v<Result>)
   {}
-  /*** END EACH() ***/
 
   // This must be awaited and all child tasks completed before destruction.
 #ifndef NDEBUG
   ~aw_spawn_many_impl() noexcept {
-    if constexpr (IsEach) {
-      assert(
-        remaining_count_or_symmetric_task == 0 &&
-        "You must submit or co_await this."
-      );
-    } else {
-      assert(
-        sync_flags_or_done_count.load() < 0 &&
-        "You must submit or co_await this."
-      );
-    }
+    assert(done_count.load() < 0 && "You must submit or co_await this.");
   }
 #endif
 
@@ -881,16 +770,11 @@ class aw_spawn_many_impl {
 };
 
 template <typename Result, size_t Count, bool IsFunc>
-using aw_spawn_many_fork = tmc::detail::rvalue_only_awaitable<
-  aw_spawn_many_impl<Result, Count, false, IsFunc>>;
-
-template <typename Result, size_t Count, bool IsFunc>
-using aw_spawn_many_each = tmc::detail::lvalue_only_awaitable<
-  aw_spawn_many_impl<Result, Count, true, IsFunc>>;
+using aw_spawn_many_fork =
+  tmc::detail::rvalue_only_awaitable<aw_spawn_many_impl<Result, Count, IsFunc>>;
 
 template <
-  typename Result, size_t Count, typename IterBegin, typename IterEnd,
-  bool IsFunc>
+  typename Result, size_t Count, typename IterBegin, typename IterEnd, bool IsFunc>
 class [[nodiscard("You must await or initiate the result of spawn_many().")]]
 aw_spawn_many : public tmc::detail::run_on_mixin<
                   aw_spawn_many<Result, Count, IterBegin, IterEnd, IsFunc>>,
@@ -930,18 +814,15 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
   {
   }
 
-  aw_spawn_many_impl<Result, Count, false, IsFunc>
-  operator co_await() && noexcept {
+  aw_spawn_many_impl<Result, Count, IsFunc> operator co_await() && noexcept {
 #ifndef NDEBUG
     assert(!is_empty && "You may only submit or co_await this once.");
     is_empty = true;
 #endif
-    bool doSymmetricTransfer =
-      tmc::detail::this_thread::exec_prio_is(executor, prio);
+    bool doSymmetricTransfer = tmc::detail::this_thread::exec_prio_is(executor, prio);
 
     using Awaitable = std::conditional_t<
-      IsFunc, tmc::task<Result>,
-      std::remove_cvref_t<std::iter_value_t<IterBegin>>>;
+      IsFunc, tmc::task<Result>, std::remove_cvref_t<std::iter_value_t<IterBegin>>>;
     constexpr auto mode = tmc::detail::get_awaitable_traits<Awaitable>::mode;
     static_assert(
       mode != tmc::detail::UNKNOWN, "This doesn't appear to be an awaitable."
@@ -952,15 +833,15 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
     }
     if constexpr (std::is_convertible_v<IterEnd, size_t>) {
       // "Sentinel" is actually a count
-      return aw_spawn_many_impl<Result, Count, false, IsFunc>(
-        std::move(iter), std::move(sentinel), executor, continuation_executor,
-        prio, doSymmetricTransfer
+      return aw_spawn_many_impl<Result, Count, IsFunc>(
+        std::move(iter), std::move(sentinel), executor, continuation_executor, prio,
+        doSymmetricTransfer
       );
     } else {
       // We have both a sentinel and a MaxCount
-      return aw_spawn_many_impl<Result, Count, false, IsFunc>(
-        std::move(iter), std::move(sentinel), maxCount, executor,
-        continuation_executor, prio, doSymmetricTransfer
+      return aw_spawn_many_impl<Result, Count, IsFunc>(
+        std::move(iter), std::move(sentinel), maxCount, executor, continuation_executor,
+        prio, doSymmetricTransfer
       );
     }
   }
@@ -979,44 +860,14 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
     if constexpr (std::is_convertible_v<IterEnd, size_t>) {
       // "Sentinel" is actually a count
       return aw_spawn_many_fork<Result, Count, IsFunc>(
-        std::move(iter), std::move(sentinel), executor, continuation_executor,
-        prio, false
+        std::move(iter), std::move(sentinel), executor, continuation_executor, prio, false
       );
     } else {
       // We have both a sentinel and a MaxCount
       return aw_spawn_many_fork<Result, Count, IsFunc>(
-        std::move(iter), std::move(sentinel), maxCount, executor,
-        continuation_executor, prio, false
-      );
-    }
-  }
-
-  /// Rather than waiting for all results at once, each result will be made
-  /// available immediately as it becomes ready. Each time this is co_awaited,
-  /// it will return the index of a single ready result. The result indexes
-  /// correspond to the indexes of the originally submitted tasks, and the
-  /// values can be accessed using `operator[]`. Results may become ready in any
-  /// order, but when awaited repeatedly, each index from `[0..task_count)` will
-  /// be returned exactly once. You must await this repeatedly until all tasks
-  /// are complete, at which point the index returned will be equal to the
-  /// value of `end()`.
-  inline aw_spawn_many_each<Result, Count, IsFunc> result_each() && {
-#ifndef NDEBUG
-    assert(!is_empty && "You may only submit or co_await this once.");
-    is_empty = true;
-#endif
-    if constexpr (std::is_convertible_v<IterEnd, size_t>) {
-      // "Sentinel" is actually a count
-      return aw_spawn_many_each<Result, Count, IsFunc>(
-        std::move(iter), std::move(sentinel), executor, continuation_executor,
+        std::move(iter), std::move(sentinel), maxCount, executor, continuation_executor,
         prio, false
       );
-    } else {
-      // We have both a sentinel and a MaxCount
-      return aw_spawn_many_each<Result, Count, IsFunc>(
-        std::move(iter), std::move(sentinel), maxCount, executor,
-        continuation_executor, prio, false
-      );
     }
   }
 
@@ -1031,16 +882,14 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
 #endif
 
     using Awaitable = std::conditional_t<
-      IsFunc, tmc::task<Result>,
-      std::remove_cvref_t<std::iter_value_t<IterBegin>>>;
+      IsFunc, tmc::task<Result>, std::remove_cvref_t<std::iter_value_t<IterBegin>>>;
 
     constexpr auto mode = tmc::detail::get_awaitable_traits<Awaitable>::mode;
     static_assert(
       mode != tmc::detail::UNKNOWN, "This doesn't appear to be an awaitable."
     );
 
-    if constexpr (mode == tmc::detail::TMC_TASK ||
-                  mode == tmc::detail::COROUTINE ||
+    if constexpr (mode == tmc::detail::TMC_TASK || mode == tmc::detail::COROUTINE ||
                   mode == tmc::detail::WRAPPER) {
       using TaskArray = std::conditional_t<
         Count == 0, std::vector<work_item>, std::array<work_item, Count>>;
@@ -1054,16 +903,14 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
         const size_t size = taskArr.size();
         for (size_t i = 0; i < size; ++i) {
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
-          taskArr[i] = tmc::detail::into_initiate(
-            tmc::detail::into_known<IsFunc>(std::move(*iter))
-          );
+          taskArr[i] =
+            tmc::detail::into_initiate(tmc::detail::into_known<IsFunc>(std::move(*iter)));
           TMC_DISABLE_WARNING_PESSIMIZING_MOVE_END
           ++iter;
         }
         tmc::detail::post_bulk_checked(executor, taskArr.data(), size, prio);
       } else {
-        if constexpr (Count == 0 &&
-                      requires(IterEnd a, IterBegin b) { a - b; }) {
+        if constexpr (Count == 0 && requires(IterEnd a, IterBegin b) { a - b; }) {
           // Caller didn't specify capacity to preallocate, but we can
           // calculate
           size_t iterSize = static_cast<size_t>(sentinel - iter);
@@ -1075,8 +922,7 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
         }
 
         size_t taskCount = 0;
-        if constexpr (Count != 0 ||
-                      requires(IterEnd a, IterBegin b) { a - b; }) {
+        if constexpr (Count != 0 || requires(IterEnd a, IterBegin b) { a - b; }) {
           const size_t size = taskArr.size();
           while (iter != sentinel && taskCount < size) {
             TMC_DISABLE_WARNING_PESSIMIZING_MOVE_BEGIN
@@ -1101,9 +947,7 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
             ++taskCount;
           }
         }
-        tmc::detail::post_bulk_checked(
-          executor, taskArr.data(), taskCount, prio
-        );
+        tmc::detail::post_bulk_checked(executor, taskArr.data(), taskCount, prio);
       }
     } else { // mode == ASYNC_INITIATE
       if constexpr (std::is_convertible_v<IterEnd, size_t>) {
@@ -1154,8 +998,7 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
   aw_spawn_many& operator=(const aw_spawn_many&) = delete;
   aw_spawn_many(aw_spawn_many&& Other)
       : iter(std::move(Other.iter)), sentinel(std::move(Other.sentinel)),
-        maxCount(std::move(Other.maxCount)),
-        executor(std::move(Other.executor)),
+        maxCount(std::move(Other.maxCount)), executor(std::move(Other.executor)),
         continuation_executor(std::move(Other.continuation_executor)),
         prio(std::move(Other.prio)) {
 #ifndef NDEBUG
@@ -1181,10 +1024,8 @@ aw_spawn_many : public tmc::detail::run_on_mixin<
 
 namespace detail {
 template <
-  typename Result, size_t Count, typename IterBegin, typename IterEnd,
-  bool IsFunc>
-struct awaitable_traits<
-  aw_spawn_many<Result, Count, IterBegin, IterEnd, IsFunc>> {
+  typename Result, size_t Count, typename IterBegin, typename IterEnd, bool IsFunc>
+struct awaitable_traits<aw_spawn_many<Result, Count, IterBegin, IterEnd, IsFunc>> {
   static constexpr configure_mode mode = WRAPPER;
   using result_type = std::conditional_t<
     std::is_void_v<Result>, void,
@@ -1192,7 +1033,7 @@ struct awaitable_traits<
       Count == 0, std::vector<tmc::detail::result_storage_t<Result>>,
       std::array<tmc::detail::result_storage_t<Result>, Count>>>;
   using self_type = aw_spawn_many<Result, Count, IterBegin, IterEnd, IsFunc>;
-  using awaiter_type = aw_spawn_many_impl<Result, Count, false, IsFunc>;
+  using awaiter_type = aw_spawn_many_impl<Result, Count, IsFunc>;
 
   static awaiter_type get_awaiter(self_type&& Awaitable) noexcept {
     return std::forward<self_type>(Awaitable).operator co_await();
diff --git a/include/tmc/spawn_tuple.hpp b/include/tmc/spawn_tuple.hpp
index 7d8bd120..f5dd5f79 100644
--- a/include/tmc/spawn_tuple.hpp
+++ b/include/tmc/spawn_tuple.hpp
@@ -10,114 +10,38 @@
 #include "tmc/detail/concepts_awaitable.hpp" // IWYU pragma: keep
 #include "tmc/detail/concepts_work_item.hpp"
 #include "tmc/detail/mixins.hpp"
-#include "tmc/detail/result_each.hpp"
 #include "tmc/detail/thread_locals.hpp"
+#include "tmc/detail/tuple_helpers.hpp"
 #include "tmc/ex_any.hpp"
 #include "tmc/work_item.hpp"
 
 #include <array>
+#include <atomic>
 #include <cassert>
 #include <coroutine>
 #include <tuple>
 #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
-
 /// The customizable task wrapper / awaitable type returned by
 /// `tmc::spawn_tuple()`.
 template <typename... Result> class aw_spawn_tuple;
 
-template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
+template <typename... Awaitable> class aw_spawn_tuple_impl {
   static constexpr auto Count = sizeof...(Awaitable);
 
-  // When result_each() is called, tasks are synchronized via an atomic bitmask
-  // with only 63 (or 31, on 32-bit) slots for tasks.
-  static_assert(
-    !IsEach || Count < TMC_PLATFORM_BITS,
-    "result_each() supports at most 63 awaitables (31 on 32-bit platforms)."
-  );
-
   static constexpr size_t WorkItemCount =
     std::tuple_size_v<typename tmc::detail::predicate_partition<
       tmc::detail::treat_as_coroutine, std::tuple, Awaitable...>::true_types>;
 
-  // result_each does not symmetric transfer, so we store the
-  // count of consumed results here instead
-  std::conditional_t<IsEach, ptrdiff_t, std::coroutine_handle<>>
-    remaining_count_or_symmetric_task;
+  // If the last task is to be run immediately via symmetric transfer, it is
+  // stored here instead of being posted to the executor.
+  std::coroutine_handle<> symmetric_task;
   std::coroutine_handle<> continuation;
   tmc::ex_any* continuation_executor;
   // the atomic synchronization variable that coordinates between this and
   // awaitable_customizer (task's final_suspend)
-  std::conditional_t<IsEach, std::atomic<size_t>, std::atomic<ptrdiff_t>>
-    sync_flags_or_done_count;
+  std::atomic<ptrdiff_t> done_count;
 
   template <typename T>
   using ResultStorage = tmc::detail::result_storage_t<tmc::detail::void_to_monostate<
@@ -131,21 +55,13 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
 
   // coroutines are prepared and stored in an array, then submitted in bulk
   template <typename T>
-  TMC_FORCE_INLINE inline void prepare_task(
-    T&& Task, ResultStorage<T>* TaskResult, size_t Idx,
-    [[maybe_unused]] size_t ContinuationPrio, work_item& Task_out
-  ) {
+  TMC_FORCE_INLINE inline void
+  prepare_task(T&& Task, ResultStorage<T>* TaskResult, work_item& Task_out) {
     tmc::detail::get_awaitable_traits<T>::set_continuation(Task, &continuation);
     tmc::detail::get_awaitable_traits<T>::set_continuation_executor(
       Task, &continuation_executor
     );
-    tmc::detail::get_awaitable_traits<T>::set_done_count(Task, &sync_flags_or_done_count);
-    if constexpr (IsEach) {
-      tmc::detail::get_awaitable_traits<T>::set_flags(
-        Task, tmc::detail::task_flags::EACH |
-                (Idx << tmc::detail::task_flags::TASKNUM_LOW_OFF) | ContinuationPrio
-      );
-    }
+    tmc::detail::get_awaitable_traits<T>::set_done_count(Task, &done_count);
     if constexpr (!std::is_void_v<
                     typename tmc::detail::get_awaitable_traits<T>::result_type>) {
       tmc::detail::get_awaitable_traits<T>::set_result_ptr(Task, TaskResult);
@@ -159,21 +75,13 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
 
   // awaitables are submitted individually
   template <typename T>
-  TMC_FORCE_INLINE inline void prepare_awaitable(
-    T&& Task, ResultStorage<T>* TaskResult, size_t Idx,
-    [[maybe_unused]] size_t ContinuationPrio
-  ) {
+  TMC_FORCE_INLINE inline void
+  prepare_awaitable(T&& Task, ResultStorage<T>* TaskResult) {
     tmc::detail::get_awaitable_traits<T>::set_continuation(Task, &continuation);
     tmc::detail::get_awaitable_traits<T>::set_continuation_executor(
       Task, &continuation_executor
     );
-    tmc::detail::get_awaitable_traits<T>::set_done_count(Task, &sync_flags_or_done_count);
-    if constexpr (IsEach) {
-      tmc::detail::get_awaitable_traits<T>::set_flags(
-        Task, tmc::detail::task_flags::EACH |
-                (Idx << tmc::detail::task_flags::TASKNUM_LOW_OFF) | ContinuationPrio
-      );
-    }
+    tmc::detail::get_awaitable_traits<T>::set_done_count(Task, &done_count);
     if constexpr (!std::is_void_v<
                     typename tmc::detail::get_awaitable_traits<T>::result_type>) {
       tmc::detail::get_awaitable_traits<T>::set_result_ptr(Task, TaskResult);
@@ -181,29 +89,14 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
   }
 
   void set_done_count(size_t NumTasks) {
-    if constexpr (IsEach) {
-      remaining_count_or_symmetric_task = static_cast<ptrdiff_t>(NumTasks);
-      sync_flags_or_done_count.store(
-        tmc::detail::task_flags::EACH, std::memory_order_release
-      );
-    } else {
-      sync_flags_or_done_count.store(
-        static_cast<ptrdiff_t>(NumTasks), std::memory_order_release
-      );
-    }
+    done_count.store(static_cast<ptrdiff_t>(NumTasks), std::memory_order_release);
   }
 
   aw_spawn_tuple_impl(
     AwaitableTuple&& Tasks, tmc::ex_any* Executor, tmc::ex_any* ContinuationExecutor,
     size_t Prio, bool DoSymmetricTransfer
   )
-      : continuation_executor{ContinuationExecutor} {
-    if constexpr (!IsEach) {
-      remaining_count_or_symmetric_task = nullptr;
-    }
-
-    size_t continuationPriority = tmc::detail::this_thread::this_task().prio;
-
+      : symmetric_task{nullptr}, continuation_executor{ContinuationExecutor} {
     std::array<work_item, WorkItemCount> taskArr;
 
     // Prepare each task as if I loops from [0..Count),
@@ -219,13 +112,11 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
          if constexpr (mode == tmc::detail::ASYNC_INITIATE) {
            // The std::move is a bit misleading here - this actually forwards the
            // awaitable from within the tuple with its original value category.
-           prepare_awaitable(
-             std::get<I>(std::move(Tasks)), &std::get<I>(result), I, continuationPriority
-           );
+           prepare_awaitable(std::get<I>(std::move(Tasks)), &std::get<I>(result));
          } else {
            prepare_task(
              tmc::detail::into_known<false>(std::get<I>(std::move(Tasks))),
-             &std::get<I>(result), I, continuationPriority, taskArr[taskIdx]
+             &std::get<I>(result), taskArr[taskIdx]
            );
            ++taskIdx;
          }
@@ -237,13 +128,10 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
     if constexpr (WorkItemCount != 0) {
       auto doneCount = Count;
       auto postCount = WorkItemCount;
-      if constexpr (!IsEach) {
-        if (DoSymmetricTransfer) {
-          remaining_count_or_symmetric_task =
-            TMC_WORK_ITEM_AS_STD_CORO(taskArr[WorkItemCount - 1]);
-          --doneCount;
-          --postCount;
-        }
+      if (DoSymmetricTransfer) {
+        symmetric_task = TMC_WORK_ITEM_AS_STD_CORO(taskArr[WorkItemCount - 1]);
+        --doneCount;
+        --postCount;
       }
       set_done_count(doneCount);
       tmc::detail::post_bulk_checked(Executor, taskArr.data(), postCount, Prio);
@@ -267,33 +155,28 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
   }
 
 public:
-  /*** SUPPORTS REGULAR AWAIT ***/
   /// Always suspends.
-  inline bool await_ready() const noexcept
-    requires(!IsEach)
-  {
+  inline bool await_ready() const noexcept {
     // Always suspends, due to the possibility to resume on another executor.
     return false;
   }
 
   /// Suspends the outer coroutine, submits the wrapped task to the
   /// executor, and waits for it to complete.
-  inline std::coroutine_handle<> await_suspend(std::coroutine_handle<> Outer) noexcept
-    requires(!IsEach)
-  {
+  inline std::coroutine_handle<> await_suspend(std::coroutine_handle<> Outer) noexcept {
 #ifndef NDEBUG
-    assert(sync_flags_or_done_count.load() >= 0 && "You may only co_await this once.");
+    assert(done_count.load() >= 0 && "You may only co_await this once.");
 #endif
     continuation = Outer;
     std::coroutine_handle<> next;
-    if (remaining_count_or_symmetric_task != nullptr) {
+    if (symmetric_task != nullptr) {
       // symmetric transfer to the last task IF it should run immediately
-      next = remaining_count_or_symmetric_task;
+      next = symmetric_task;
     } else {
       // 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.
-      auto remaining = sync_flags_or_done_count.fetch_sub(1, std::memory_order_acq_rel);
+      auto remaining = done_count.fetch_sub(1, std::memory_order_acq_rel);
       // No symmetric transfer - all tasks were already posted.
       // Suspend if remaining > 0 (task is still running)
       if (remaining > 0) {
@@ -320,70 +203,14 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
   /// void, its slot is represented by a std::monostate. If the awaitable would
   /// return a non-default-constructible type, that result will be wrapped in a
   /// std::optional.
-  TMC_AWAIT_RESUME inline ResultTuple&& await_resume() noexcept
-    requires(!IsEach)
-  {
+  TMC_AWAIT_RESUME inline ResultTuple&& await_resume() noexcept {
     return std::move(result);
   }
-  /*** END REGULAR AWAIT ***/
-
-  /*** SUPPORTS EACH() ***/
-  /// Always suspends.
-  inline bool await_ready() const noexcept
-    requires(IsEach)
-  {
-    return tmc::detail::result_each_await_ready();
-  }
-
-  /// Suspends the outer coroutine, submits the wrapped task to the
-  /// executor, and waits for it to complete.
-  inline bool await_suspend(std::coroutine_handle<> Outer) noexcept
-    requires(IsEach)
-  {
-    return tmc::detail::result_each_await_suspend(
-      remaining_count_or_symmetric_task, Outer, continuation, continuation_executor,
-      sync_flags_or_done_count
-    );
-  }
-
-  /// Returns the value provided by the wrapped tasks.
-  /// Each task has a slot in the tuple. If the task would return void, its
-  /// slot is represented by a std::monostate.
-  TMC_AWAIT_RESUME inline size_t await_resume() noexcept
-    requires(IsEach)
-  {
-    return tmc::detail::result_each_await_resume(
-      remaining_count_or_symmetric_task, sync_flags_or_done_count
-    );
-  }
-
-  /// Provides a sentinel value that can be compared against the value returned
-  /// from co_await.
-  inline size_t end() noexcept
-    requires(IsEach)
-  {
-    return 64;
-  }
-
-  // Gets the ready result at the given index.
-  template <size_t I>
-  inline std::tuple_element_t<I, ResultTuple>& get() noexcept
-    requires(IsEach)
-  {
-    return std::get<I>(result);
-  }
-  /*** END EACH() ***/
 
   // This must be awaited and all child tasks completed before destruction.
 #ifndef NDEBUG
   ~aw_spawn_tuple_impl() noexcept {
-    if constexpr (IsEach) {
-      assert(
-        remaining_count_or_symmetric_task == 0 && "You must submit or co_await this."
-      );
-    } else {
-      assert(sync_flags_or_done_count.load() < 0 && "You must submit or co_await this.");
-    }
+    assert(done_count.load() < 0 && "You must submit or co_await this.");
   }
 #endif
 
@@ -397,11 +224,7 @@ template <bool IsEach, typename... Awaitable> class aw_spawn_tuple_impl {
 
 template <typename... Result>
 using aw_spawn_tuple_fork =
-  tmc::detail::rvalue_only_awaitable<aw_spawn_tuple_impl<false, Result...>>;
-
-template <typename... Result>
-using aw_spawn_tuple_each =
-  tmc::detail::lvalue_only_awaitable<aw_spawn_tuple_impl<true, Result...>>;
+  tmc::detail::rvalue_only_awaitable<aw_spawn_tuple_impl<Result...>>;
 
 template <typename... Awaitable>
 class [[nodiscard(
@@ -446,7 +269,7 @@ class [[nodiscard(
   }
 
   /// Initiates all of the wrapped tasks and waits for them to complete.
-  aw_spawn_tuple_impl<false, Awaitable...> operator co_await() && noexcept {
+  aw_spawn_tuple_impl<Awaitable...> operator co_await() && noexcept {
     bool doSymmetricTransfer = tmc::detail::this_thread::exec_prio_is(executor, prio);
 #ifndef NDEBUG
     if constexpr (Count != 0) {
@@ -455,7 +278,7 @@ class [[nodiscard(
     }
     is_empty = true;
 #endif
-    return aw_spawn_tuple_impl<false, Awaitable...>(
+    return aw_spawn_tuple_impl<Awaitable...>(
       std::move(wrapped), executor, continuation_executor, prio, doSymmetricTransfer
     );
   }
@@ -552,27 +375,6 @@ class [[nodiscard(
     );
   }
 
-  /// Rather than waiting for all results at once, each result will be made
-  /// available immediately as it becomes ready. Each time this is co_awaited,
-  /// it will return the index of a single ready result. The result indexes
-  /// correspond to the indexes of the originally submitted tasks, and the
-  /// values can be accessed using `.get<index>()`. Results may become ready
-  /// in any order, but when awaited repeatedly, each index from
-  /// `[0..task_count)` will be returned exactly once. You must await this
-  /// repeatedly until all results have been consumed, at which point the index
-  /// returned will be equal to the value of `end()`.
-  inline aw_spawn_tuple_each<Awaitable...> result_each() && {
-#ifndef NDEBUG
-    if constexpr (Count != 0) {
-      // Ensure that this was not previously moved-from
-      assert(!is_empty && "You may only submit or co_await this once.");
-    }
-    is_empty = true;
-#endif
-    return aw_spawn_tuple_each<Awaitable...>(
-      std::move(wrapped), executor, continuation_executor, prio, false
-    );
-  }
 };
 
 /// Spawns multiple awaitables and returns an awaiter that allows you to await
@@ -612,7 +414,7 @@ template <typename... Awaitables> struct awaitable_traits<aw_spawn_tuple<Awaitab
   using result_type = std::tuple<tmc::detail::void_to_monostate<
     typename tmc::detail::get_awaitable_traits<Awaitables>::result_type>...>;
   using self_type = aw_spawn_tuple<Awaitables...>;
-  using awaiter_type = aw_spawn_tuple_impl<false, Awaitables...>;
+  using awaiter_type = aw_spawn_tuple_impl<Awaitables...>;
 
   static awaiter_type get_awaiter(self_type&& Awaitable) noexcept {
     return static_cast<self_type&&>(Awaitable).operator co_await();

From efdeced137e39e1ec3052150c586516d86696c3a Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 19:04:59 -0700
Subject: [PATCH 4/9] add agent documentation

---
 AGENTS.md | 130 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 CLAUDE.md |   1 +
 2 files changed, 131 insertions(+)
 create mode 100644 AGENTS.md
 create mode 100644 CLAUDE.md

diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 00000000..c55eadd5
--- /dev/null
+++ b/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.
diff --git a/CLAUDE.md b/CLAUDE.md
new file mode 100644
index 00000000..43c994c2
--- /dev/null
+++ b/CLAUDE.md
@@ -0,0 +1 @@
+@AGENTS.md

From 4a33bea3f8aacc71ada304c12d0a83fed3543a09 Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 19:05:05 -0700
Subject: [PATCH 5/9] cleanup

---
 include/tmc/detail/tsan.hpp | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/include/tmc/detail/tsan.hpp b/include/tmc/detail/tsan.hpp
index 3adc553d..b7680314 100644
--- a/include/tmc/detail/tsan.hpp
+++ b/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

From 6542397819575be68f946228848494d469f0d63a Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 19:08:50 -0700
Subject: [PATCH 6/9] cleanup

---
 include/tmc/spawn_tuple.hpp | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/include/tmc/spawn_tuple.hpp b/include/tmc/spawn_tuple.hpp
index f5dd5f79..d66230df 100644
--- a/include/tmc/spawn_tuple.hpp
+++ b/include/tmc/spawn_tuple.hpp
@@ -5,7 +5,6 @@
 
 #pragma once
 
-#include "tmc/detail/awaitable_customizer.hpp"
 #include "tmc/detail/compat.hpp"
 #include "tmc/detail/concepts_awaitable.hpp" // IWYU pragma: keep
 #include "tmc/detail/concepts_work_item.hpp"
@@ -75,8 +74,7 @@ template <typename... Awaitable> class aw_spawn_tuple_impl {
 
   // awaitables are submitted individually
   template <typename T>
-  TMC_FORCE_INLINE inline void
-  prepare_awaitable(T&& Task, ResultStorage<T>* TaskResult) {
+  TMC_FORCE_INLINE inline void prepare_awaitable(T&& Task, ResultStorage<T>* TaskResult) {
     tmc::detail::get_awaitable_traits<T>::set_continuation(Task, &continuation);
     tmc::detail::get_awaitable_traits<T>::set_continuation_executor(
       Task, &continuation_executor
@@ -374,7 +372,6 @@ class [[nodiscard(
       std::move(wrapped), executor, continuation_executor, prio, false
     );
   }
-
 };
 
 /// Spawns multiple awaitables and returns an awaiter that allows you to await

From 6e165bbab3c63aaa8d28ca529f0bc18b43787453 Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 19:34:49 -0700
Subject: [PATCH 7/9] final cleanup

---
 include/tmc/mux_many.hpp | 46 +++++++++++++++++++++++++++-------------
 1 file changed, 31 insertions(+), 15 deletions(-)

diff --git a/include/tmc/mux_many.hpp b/include/tmc/mux_many.hpp
index 070d6d39..a7ad3f01 100644
--- a/include/tmc/mux_many.hpp
+++ b/include/tmc/mux_many.hpp
@@ -93,21 +93,24 @@ class mux_many : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
     "mux_many supports up to 63 awaitables (31 on 32-bit platforms)."
   );
 
-  // The count of submitted-but-not-yet-consumed results.
-  ptrdiff_t remaining_count;
   std::coroutine_handle<> continuation;
   tmc::ex_any* continuation_executor;
+
   // Bitmap of slots that have been forked but not yet returned from co_await.
   // A set bit means the slot is active: its result is pending or ready but has
   // not been consumed, so the slot may not be re-forked. Non-atomic; only
   // mutated by the (single-threaded) owner.
   size_t active_slots;
+  // equal to popcount(active_slots)
+  ptrdiff_t remaining_count;
+
   // the atomic synchronization variable that coordinates between this and
   // awaitable_customizer (task's final_suspend)
   std::atomic<size_t> sync_flags = tmc::detail::task_flags::EACH;
 
   // Non-default-constructible results are stored in a std::optional. This is not visible
-  // to the user; when get() is called, a reference to the constructed object is returned.
+  // to the user; when operator[] is called, a reference to the constructed object is
+  // returned.
   using ResultStorage = tmc::detail::result_storage_t<Result>;
   struct empty {};
   using ResultArray =
@@ -337,8 +340,8 @@ class mux_many : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
   {
     assert(Idx < result_arr.size());
     assert(
-      !is_active(Idx) && "You may only call get() on a slot after its result "
-                         "has been returned from co_await."
+      !is_active(Idx) && "You may only call operator[] on a slot after its result has "
+                         "been returned from co_await."
     );
     if constexpr (std::is_default_constructible_v<Result>) {
       return result_arr[Idx];
@@ -378,7 +381,7 @@ class mux_many : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
   /// `Executor` defaults to the current executor.
   /// `Priority` defaults to the current priority.
   ///
-  /// This method is not thread-safe.
+  /// This method is not thread-safe to call concurrently with itself or `co_await`.
   template <typename Aw, typename Exec = tmc::ex_any*>
   inline void fork(
     size_t Idx, Aw&& Awaitable, Exec&& Executor = tmc::current_executor(),
@@ -449,20 +452,33 @@ class mux_many : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
   /// call this function on other compilers, but no HALO-specific optimizations
   /// will be applied.
   ///
-  /// This method is not thread-safe.
+  /// This method is not thread-safe to call concurrently with itself or `co_await`.
   ///
-  /// WARNING: Don't allow coroutines passed into this to cross a loop boundary,
-  /// or Clang will try to reuse the same allocation for multiple active
-  /// coroutines.
+  /// 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:
+  /// immediately.
+  ///
+  /// Proper usage, taking both of the above into account:
   /// ```
-  /// auto mux = tmc::mux_many<int, 2>();
-  /// co_await mux.fork_clang(0, task(0));
-  /// co_await mux.fork_clang(1, task(1));
-  /// for (size_t i = co_await mux; i != mux.end(); i = co_await mux) { ... }
+  /// 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();
+  ///   }
+  /// }
   /// ```
   template <typename Aw, typename Exec = tmc::ex_any*>
   [[nodiscard("You must co_await fork_clang() immediately for HALO to be possible.")]]

From 282fa359c510f0f7b6ee65c5c2569d480326a092 Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Thu, 2 Jul 2026 22:00:49 -0700
Subject: [PATCH 8/9] fix race by replacing fetch_or with CAS release

---
 include/tmc/detail/result_each.ipp | 52 ++++++++++++++----------------
 1 file changed, 25 insertions(+), 27 deletions(-)

diff --git a/include/tmc/detail/result_each.ipp b/include/tmc/detail/result_each.ipp
index 982b5d94..95c17850 100644
--- a/include/tmc/detail/result_each.ipp
+++ b/include/tmc/detail/result_each.ipp
@@ -22,34 +22,32 @@ bool result_each_await_suspend(
   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 by holding the EACH bit (lock bit).
-    //
-    // 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 (remaining_count == 0) {
+    return false;
+  }
+  // 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.
   }
-  // We got a result.
-  return false;
 }
 
 size_t result_each_await_resume(

From f3d54ba83fb4c21cb6a2e91f9941fb7ef6daac16 Mon Sep 17 00:00:00 2001
From: tzcnt <trailingzerocount@gmail.com>
Date: Fri, 3 Jul 2026 07:39:03 -0700
Subject: [PATCH 9/9] noinline get() under TSan

This prevents the compiler from generating a speculative load of incomplete awaitable paths, which triggers a TSan false positive. It's truly a false positive because the unused result is discarded.
---
 include/tmc/detail/tsan.hpp | 13 +++++++++++++
 include/tmc/mux_many.hpp    |  4 +++-
 include/tmc/mux_tuple.hpp   |  4 +++-
 3 files changed, 19 insertions(+), 2 deletions(-)

diff --git a/include/tmc/detail/tsan.hpp b/include/tmc/detail/tsan.hpp
index b7680314..0f6c15cc 100644
--- a/include/tmc/detail/tsan.hpp
+++ b/include/tmc/detail/tsan.hpp
@@ -25,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
diff --git a/include/tmc/mux_many.hpp b/include/tmc/mux_many.hpp
index a7ad3f01..2fd3106b 100644
--- a/include/tmc/mux_many.hpp
+++ b/include/tmc/mux_many.hpp
@@ -12,6 +12,7 @@
 #include "tmc/detail/concepts_work_item.hpp"
 #include "tmc/detail/result_each.hpp"
 #include "tmc/detail/thread_locals.hpp"
+#include "tmc/detail/tsan.hpp"
 #include "tmc/ex_any.hpp"
 #include "tmc/work_item.hpp"
 
@@ -335,7 +336,8 @@ class mux_many : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
   inline size_t end() noexcept { return 64; }
 
   /// Gets the ready result at the given index.
-  inline std::add_lvalue_reference_t<Result> operator[](size_t Idx) noexcept
+  TMC_TSAN_NO_SPECULATE inline std::add_lvalue_reference_t<Result>
+  operator[](size_t Idx) noexcept
     requires(!std::is_void_v<Result>)
   {
     assert(Idx < result_arr.size());
diff --git a/include/tmc/mux_tuple.hpp b/include/tmc/mux_tuple.hpp
index 6f53f125..af003319 100644
--- a/include/tmc/mux_tuple.hpp
+++ b/include/tmc/mux_tuple.hpp
@@ -12,6 +12,7 @@
 #include "tmc/detail/concepts_work_item.hpp"
 #include "tmc/detail/result_each.hpp"
 #include "tmc/detail/thread_locals.hpp"
+#include "tmc/detail/tsan.hpp"
 #include "tmc/detail/tuple_helpers.hpp"
 #include "tmc/ex_any.hpp"
 #include "tmc/work_item.hpp"
@@ -226,7 +227,8 @@ class mux_tuple : private tmc::detail::AwaitTagNoGroupCoAwaitLvalue {
   inline size_t end() noexcept { return 64; }
 
   /// Gets the ready result at the given index.
-  template <size_t I> inline std::tuple_element_t<I, ValueTuple>& get() noexcept {
+  template <size_t I>
+  TMC_TSAN_NO_SPECULATE inline std::tuple_element_t<I, ValueTuple>& get() noexcept {
     assert(
       !is_active<I>() && "You may only call get() on a slot after its result "
                          "has been returned from co_await."