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 { 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 { 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 { 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 { 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 library - but if it is not used elsewhere in the project, you should consult the user first, as is a heavyweight include. If the user rejects the use of , `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()`, 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 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/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& sync_flags + std::coroutine_handle<>& continuation, std::atomic& 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..95c17850 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 #include @@ -19,54 +17,36 @@ namespace tmc { namespace detail { -bool result_each_await_ready() noexcept { - // Always suspends, due to the possibility to resume on another executor. - return false; -} - bool result_each_await_suspend( ptrdiff_t remaining_count, std::coroutine_handle<> Outer, - std::coroutine_handle<>& continuation, tmc::ex_any* continuation_executor, - std::atomic& sync_flags + std::coroutine_handle<>& continuation, std::atomic& sync_flags ) noexcept { continuation = Outer; - if (remaining_count != 0) { - // This logic is necessary because we submitted all child tasks before the - // parent suspended. Allowing parent to be resumed before it suspends - // would be UB. Therefore we need to block the resumption until here. - // WARNING: We can use fetch_sub here because we know this bit is set. - // It generates xadd instruction which is slightly more efficient than - // fetch_and. But not safe to use if the bit might not be set. - size_t resumeState; - do { - resumeState = sync_flags.fetch_sub( - tmc::detail::task_flags::EACH, std::memory_order_acq_rel - ); - assert(0 != (resumeState & tmc::detail::task_flags::EACH)); - if (0 == (resumeState & ~tmc::detail::task_flags::EACH)) { - return true; // we suspended and no tasks were ready - } - // A result became ready, so try to resume immediately. - resumeState = sync_flags.fetch_or( - tmc::detail::task_flags::EACH, std::memory_order_acq_rel - ); - if (0 != (resumeState & tmc::detail::task_flags::EACH)) { - return true; // Another thread already resumed - } - // If we resumed, but another thread already consumed - // all the results, try again to suspend - } while (0 == (resumeState & ~tmc::detail::task_flags::EACH)); - } - if (continuation_executor == nullptr || - tmc::detail::this_thread::exec_is(continuation_executor)) { + if (remaining_count == 0) { return false; - } else { - // Need to resume on a different executor - tmc::detail::post_checked( - continuation_executor, std::move(Outer), - tmc::detail::this_thread::this_task().prio - ); - return true; + } + // Children are submitted before the parent suspends. Allowing the parent to + // be resumed before it suspends would be UB, so children are blocked from + // resuming it by the EACH bit (lock bit), which is held for the entire time + // the parent is running. It may only be released here, atomically with the + // decision to suspend, and only if no results are already ready. + // This release must be the final operation via CAS rather than speculatively, because + // once released, another thread may resume this and destroy the coroutine frame. + size_t state = sync_flags.load(std::memory_order_acquire); + while (true) { + assert(0 != (state & tmc::detail::task_flags::EACH)); + if (0 != (state & ~tmc::detail::task_flags::EACH)) { + // A result is ready. Resume immediately. The EACH bit was never + // released, so no child could have claimed the resumption. + return false; + } + if (sync_flags.compare_exchange_weak( + state, state & ~tmc::detail::task_flags::EACH, std::memory_order_acq_rel, + std::memory_order_acquire + )) { + return true; + } + // CAS failure means a child just set its result bit; retry to consume it. } } diff --git a/include/tmc/detail/tsan.hpp b/include/tmc/detail/tsan.hpp index 3adc553d..0f6c15cc 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 @@ -20,3 +25,16 @@ #else #define TMC_INLINE_OR_TSAN TMC_FORCE_INLINE #endif + +// Some compilers (observed on Clang 18/19 and Apple Clang) may convert +// `Cond ? mux.get<0>() : mux.get<1>()` into a select of both branches. +// The reads happen after the atomic acquire of Cond, so it's not a true race, +// but it triggers a TSan false positive because the discarded value races +// with a write from the completing task. Forcing this to be noinline under +// TSan prevents the speculative load from being issued, while still allowing +// the function to be instrumented by TSan for genuine races. +#ifdef TMC_HAS_TSAN +#define TMC_TSAN_NO_SPECULATE __attribute__((noinline)) +#else +#define TMC_TSAN_NO_SPECULATE +#endif 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 +#include + +namespace tmc { +namespace detail { +// Replace void with std::monostate (void is not a valid tuple element type) +template +using void_to_monostate = std::conditional_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 struct last_type { + using type = typename decltype((std::type_identity{}, ...))::type; +}; + +template <> struct last_type<> { + // workaround for empty tuples - the task object will be void / empty + using type = void; +}; + +template using last_type_t = typename last_type::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