diff --git a/README.md b/README.md index 73fb0fd..30dc371 100644 --- a/README.md +++ b/README.md @@ -57,8 +57,10 @@ header only core ## cth::coro coroutine runtime on asio +- `co::context`: interface of where coroutines run and wait, implemented by `scheduler` and `strand` - `co::scheduler`: `asio::io_context` on its own threads -- `co::executor`: spawns tasks, moves execution back onto the scheduler after foreign awaitables +- `co::strand`: serial queue on a scheduler, its jobs run one at a time and in post order +- `co::executor`: spawns tasks, every `co_await` returns the caller to its context, except `schedule()` - `this_coro::executor`, `this_coro::wait(duration)`, native handle waits ## cth::os @@ -99,7 +101,7 @@ auto task = []() -> cth::co::executor_task { co_return 42; }; -int const result = cth::co::sync(executor.spawn(task())); +int const result = executor.sync_wait(task()); ``` ## logging diff --git a/cth/incl/cth/meta/concepts.hpp b/cth/incl/cth/meta/concepts.hpp index cbceed7..7fe8d51 100644 --- a/cth/incl/cth/meta/concepts.hpp +++ b/cth/incl/cth/meta/concepts.hpp @@ -95,6 +95,12 @@ concept trivial = std::is_trivial_v; template concept trivially_copyable = std::is_trivially_copyable_v; +template +concept nothrow_constructible = std::constructible_from && std::is_nothrow_constructible_v; + +template +concept nothrow_copy_constructible = std::copy_constructible && std::is_nothrow_copy_constructible_v; + template concept reference = std::is_reference_v; diff --git a/cth/incl/cth/meta/traits/trait_packs.hpp b/cth/incl/cth/meta/traits/trait_packs.hpp index 7dac50e..7e70137 100644 --- a/cth/incl/cth/meta/traits/trait_packs.hpp +++ b/cth/incl/cth/meta/traits/trait_packs.hpp @@ -1,58 +1,86 @@ #pragma once -#include "cth/meta/utility.hpp" - +#include #include -namespace cth::mta { +namespace cth::mta::dev { +template class Trait, class... Args> +struct unpacked { + using type = void; +}; +template class Trait, class... Args> +struct unpacked { + using type = Trait; +}; +} +namespace cth::mta { +/** + * a trait with its arguments, instantiated only when unpacked + * @details @ref type is Trait, void if Trait is ill-formed + */ template class Trait, class... Args> struct trait_pack { static constexpr bool can_unpack = requires { typename Trait; }; - static constexpr auto unpack() { - if constexpr(can_unpack) - return std::type_identity>{}; - else - return std::type_identity{}; - } - - using unpacked_type = type_of_t; + using unpacked_type = dev::unpacked::type; using type = unpacked_type; + + static constexpr auto unpack() { return std::type_identity{}; } }; -template -consteval auto first_pack() { - if constexpr(Pack::can_unpack) - return Pack::unpack(); - else if constexpr(!Pack::can_unpack && sizeof...(Rest) > 0) - return first_pack(); - else - static_assert(false, "no unpackable trait found"); +/** + * a pack that always unpacks to T + */ +template +using type_pack = trait_pack; } -template -using first_pack_t = type_of_t()>; - +namespace cth::mta::dev { template -consteval auto unpack_trait() { - if constexpr(Pack::can_unpack) - return Pack::unpack(); - else - static_assert(false, "Unable to unpack trait"); +struct unpack_trait; + +template class Trait, class... Args> +struct unpack_trait> { + static_assert(requires { typename Trait; }, "Unable to unpack trait"); + using type = Trait; +}; + +template +struct first_pack { + static_assert(false, "no unpackable trait found"); + using type = void; +}; + +template class Trait, class... Args, class... Rest> +struct first_pack, Rest...> : first_pack {}; + +template class Trait, class... Args, class... Rest> + requires requires { typename Trait; } +struct first_pack, Rest...> { + using type = Trait; +}; } +namespace cth::mta { +/** + * unpacks the pack + * @pre the pack can unpack + */ template -using unpack_trait_t = decltype(unpack_trait())::type; +using unpack_trait_t = dev::unpack_trait::type; -template -consteval auto conditional_pack() { - if constexpr(Condition) - return unpack_trait(); - else - return unpack_trait(); -} +/** + * unpacks the first pack that can unpack + */ +template +using first_pack_t = dev::first_pack::type; +/** + * unpacks only the chosen pack + * @pre the chosen pack can unpack + */ template -using conditional_pack_t = type_of_t()>; +using conditional_pack_t = + std::conditional_t, dev::unpack_trait>::type; } diff --git a/cth/tests/src/meta/concepts_test.cpp b/cth/tests/src/meta/concepts_test.cpp index df6282a..5119eb8 100644 --- a/cth/tests/src/meta/concepts_test.cpp +++ b/cth/tests/src/meta/concepts_test.cpp @@ -26,6 +26,37 @@ TYPE_TEST(trivially_copyable, main) { ASSERT_TRUE((satisfies) ); } +TYPE_TEST(nothrow_constructible, main) { + struct Point { + float x, y; + }; + struct throwing { + explicit throwing(int) noexcept(false) {} + }; + + ASSERT_TRUE((nothrow_constructible) ); + ASSERT_TRUE((nothrow_constructible) ); + ASSERT_TRUE((nothrow_constructible) ); + ASSERT_FALSE((nothrow_constructible) ); + ASSERT_FALSE((nothrow_constructible) ); + ASSERT_TRUE((satisfies) ); +} + +TYPE_TEST(nothrow_copy_constructible, main) { + struct Point { + float x, y; + }; + struct throwing_copy { + throwing_copy() = default; + throwing_copy(throwing_copy const&) noexcept(false) {} + }; + + ASSERT_TRUE((nothrow_copy_constructible) ); + ASSERT_TRUE((nothrow_copy_constructible) ); + ASSERT_FALSE((nothrow_copy_constructible) ); + ASSERT_TRUE((satisfies) ); +} + TYPE_TEST(any_satisfy, main) { ASSERT_TRUE((any_satisfy) ); ASSERT_TRUE((any_satisfy) ); diff --git a/cth/tests/src/meta/traits/trait_packs_test.cpp b/cth/tests/src/meta/traits/trait_packs_test.cpp index f5064fa..ab7ee99 100644 --- a/cth/tests/src/meta/traits/trait_packs_test.cpp +++ b/cth/tests/src/meta/traits/trait_packs_test.cpp @@ -25,6 +25,13 @@ PACK_TEST(first_trait_pack, main) { static_assert(std::same_as, pack1_t>); } +PACK_TEST(type_pack, main) { + static_assert(type_pack::can_unpack); + static_assert(std::same_as>, int const&>); + static_assert(std::same_as>, void>); + static_assert(std::same_as>, float>); +} + PACK_TEST(conditional_trait_pack, main) { static_assert(std::same_as, pack1_t>); static_assert(std::same_as, pack2_t>); diff --git a/cth_coro/incl/cth/coro/awaiters/base/native_handle_awaiter_base.hpp b/cth_coro/incl/cth/coro/awaiters/base/native_handle_awaiter_base.hpp index 4702f0f..1ed8c24 100644 --- a/cth_coro/incl/cth/coro/awaiters/base/native_handle_awaiter_base.hpp +++ b/cth_coro/incl/cth/coro/awaiters/base/native_handle_awaiter_base.hpp @@ -1,18 +1,32 @@ #pragma once -#include "cth/coro/scheduler.hpp" -#include "cth/coro/tasks/awaiters/dev/this_coro_awaiter_base.hpp" +#include "cth/coro/context.hpp" +#include "cth/coro/awaiters/base/wait_awaiter_base.hpp" +#include "cth/coro/utility/fwd.hpp" #include +#include + namespace cth::co { -template -struct native_handle_awaiter_base : dev::this_coro_awaiter_base { +/** + * base of awaiters that wait on a native handle + * @details resumes the caller on its context, a failed wait throws in the caller + * @attention a derived await_resume() calls @ref wait_awaiter_base::await_resume() first + */ +struct native_handle_awaiter_base : this_coro_awaiter_base, wait_awaiter_base { constexpr explicit native_handle_awaiter_base(os::native_handle_t handle) : _handle{handle} {} [[nodiscard]] bool await_ready() const { return os::check_ready(_handle); } - void await_suspend(std::coroutine_handle<> h) const { - this->payload().scheduler().await(_handle, [h] { h.resume(); }); + + /** + * resumes the caller on its context once the handle is signaled + * @param caller this_coro coroutine that awaits the handle + */ + template + void await_suspend(std::coroutine_handle caller) { + static_assert(this_coro_compatible, "native handle awaiters need a this_coro caller"); + caller.promise().context().await(_handle, resume(caller)); } private: diff --git a/cth_coro/incl/cth/coro/awaiters/base/osfence_awaiter_base.hpp b/cth_coro/incl/cth/coro/awaiters/base/osfence_awaiter_base.hpp index 3bb9ad2..26942e2 100644 --- a/cth_coro/incl/cth/coro/awaiters/base/osfence_awaiter_base.hpp +++ b/cth_coro/incl/cth/coro/awaiters/base/osfence_awaiter_base.hpp @@ -4,9 +4,7 @@ #include "cth/os/fence.hpp" namespace cth::co { -template -struct osfence_awaiter_base : native_handle_awaiter_base { - explicit osfence_awaiter_base(os::fence const& fence) : native_handle_awaiter_base{ - fence.native_handle()} {} +struct osfence_awaiter_base : native_handle_awaiter_base { + explicit osfence_awaiter_base(os::fence const& fence) : native_handle_awaiter_base{fence.native_handle()} {} }; } diff --git a/cth_coro/incl/cth/coro/awaiters/base/wait_awaiter_base.hpp b/cth_coro/incl/cth/coro/awaiters/base/wait_awaiter_base.hpp new file mode 100644 index 0000000..f138082 --- /dev/null +++ b/cth_coro/incl/cth/coro/awaiters/base/wait_awaiter_base.hpp @@ -0,0 +1,39 @@ +#pragma once +#include "cth/coro/exception.hpp" + +#include +#include + + +namespace cth::co { +/** + * base of awaiters that wait with @ref context::await(), a failed wait throws in the caller + * @attention a derived await_resume() calls the one of this base first, else a failed wait goes unnoticed + */ +class wait_awaiter_base { +public: + /** + * throws if the wait failed + * @throws cth::except::coro_exception reason: the wait failed + */ + void await_resume() const { + CTH_CORO_STABLE_THROW(_error, "wait failed, got {}", _error.message()) {} + } + +protected: + /** + * gives the callback for @ref context::await(), it stores the error and resumes the coroutine + * @param h coroutine to resume + * @return callback for @ref context::await() + */ + [[nodiscard]] auto resume(std::coroutine_handle<> h) { + return [this, h](std::error_code ec) { + _error = ec; + h.resume(); + }; + } + +private: + std::error_code _error{}; +}; +} diff --git a/cth_coro/incl/cth/coro/awaiters/data_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/data_awaiter.hpp index a5abe10..acf3bdc 100644 --- a/cth_coro/incl/cth/coro/awaiters/data_awaiter.hpp +++ b/cth_coro/incl/cth/coro/awaiters/data_awaiter.hpp @@ -1,18 +1,27 @@ #pragma once +#include + #include +#include namespace cth::co { template class data_awaiter { public: - constexpr explicit data_awaiter(T type) : _type{type} {} + constexpr explicit data_awaiter(T value) noexcept(mta::nothrow_constructible) : + _type{std::forward(value)} {} [[nodiscard]] constexpr bool await_ready() const noexcept { return true; } constexpr void await_suspend(std::coroutine_handle<>) const noexcept {} - [[nodiscard]] constexpr T await_resume() const noexcept { return _type; } + /** + * @attention moves the value out, resume once + */ + [[nodiscard]] constexpr T await_resume() noexcept(mta::nothrow_constructible) { + return std::forward(_type); + } private: T _type; diff --git a/cth_coro/incl/cth/coro/awaiters/raw_native_handle_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/raw_native_handle_awaiter.hpp index 248ba34..e956ea6 100644 --- a/cth_coro/incl/cth/coro/awaiters/raw_native_handle_awaiter.hpp +++ b/cth_coro/incl/cth/coro/awaiters/raw_native_handle_awaiter.hpp @@ -1,6 +1,6 @@ #pragma once - -#include "cth/coro/scheduler.hpp" +#include "cth/coro/context.hpp" +#include "cth/coro/awaiters/base/wait_awaiter_base.hpp" #include @@ -8,14 +8,17 @@ namespace cth::co { -struct raw_native_handle_awaiter { - co::scheduler const& scheduler; +/** + * waits until the handle is signaled, resumes the caller on the context + * @details a failed wait throws in the caller + */ +struct raw_native_handle_awaiter : wait_awaiter_base { + constexpr raw_native_handle_awaiter(co::context const& ctx, os::native_handle_t h) : context{ctx}, handle{h} {} + + co::context const& context; os::native_handle_t handle; [[nodiscard]] bool await_ready() const { return os::check_ready(handle); } - void await_suspend(std::coroutine_handle<> h) const { - scheduler.await(handle, [h] { h.resume(); }); - } - static constexpr void await_resume() {} + void await_suspend(std::coroutine_handle<> h) { context.await(handle, resume(h)); } }; } diff --git a/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp index b0f09a1..f09a589 100644 --- a/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp +++ b/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp @@ -1,41 +1,46 @@ #pragma once -#include "cth/coro/scheduler.hpp" -#include "cth/coro/this_coro/this_coro_base.hpp" +#include "cth/coro/context.hpp" #include "cth/coro/utility/fwd.hpp" +#include + #include namespace cth::co { /** - * switches execution onto a target @ref scheduler and stays there - * @details - * @tparam Pyld payload of the migrating coroutine, defaulted to @ref this_coro::default_payload + * switches execution onto a target @ref context and stays there + * @details a this_coro caller runs on the target afterwards, its payload stays */ -template -struct schedule_awaiter : this_coro_base { +struct schedule_awaiter : this_coro_awaiter_base { public: - constexpr schedule_awaiter(co::scheduler const& target) : _target{target} {} - - [[nodiscard]] bool await_ready() const noexcept { return _target.owns_thread(); } - - template requires this_coro::scheduler_payload - void await_suspend(std::coroutine_handle h) { - // even though we don't inject it, payload compatibility should be enforced - static_assert(!this_coro_compatible || this_coro::promise_injectable_with); - - if constexpr(this_coro::scheduler_payload && this_coro_compatible) - h.promise().rebind_scheduler(_target); + constexpr schedule_awaiter(co::context const& target) : _target{target} {} + + /** + * @return false, @ref await_suspend() decides + */ + [[nodiscard]] constexpr bool await_ready() const noexcept { return false; } + + /** + * rebinds a this_coro caller to the target and posts the caller onto the target + * @details owns_thread() is not enough: a scheduler also owns the threads that run the jobs of its strands + * @return false if @ref context::current() is the target, the caller continues without a post + */ + template + bool await_suspend(std::coroutine_handle h) { + if constexpr(this_coro_compatible) + h.promise().rebind_context(_target); + + if(context::current() == &_target) + return false; _target.post([h]() { h.resume(); }); + return true; } - template P> - constexpr void inject(P const&) const noexcept {} - constexpr void await_resume() const noexcept {} private: - co::scheduler const& _target; + co::context const& _target; }; } diff --git a/cth_coro/incl/cth/coro/awaiters/time_point_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/time_point_awaiter.hpp new file mode 100644 index 0000000..60a1f40 --- /dev/null +++ b/cth_coro/incl/cth/coro/awaiters/time_point_awaiter.hpp @@ -0,0 +1,27 @@ +#pragma once +#include "cth/coro/context.hpp" +#include "cth/coro/awaiters/base/wait_awaiter_base.hpp" +#include "cth/coro/utility/fwd.hpp" + +#include + +#include + + +namespace cth::co { +/** + * waits until the time point, resumes the caller on the context + * @details a failed wait throws in the caller + */ +struct time_point_awaiter : wait_awaiter_base { + constexpr time_point_awaiter(chrono::time_point_t tp, co::context const& ctx) : timePoint{tp}, context{ctx} {} + + chrono::time_point_t timePoint; + co::context const& context; + + [[nodiscard]] bool await_ready() const noexcept { return chrono::clock_t::now() >= timePoint; } + + void await_suspend(std::coroutine_handle<> h) { context.await(timePoint, resume(h)); } +}; + +} diff --git a/cth_coro/incl/cth/coro/awaiters/wait_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/wait_awaiter.hpp deleted file mode 100644 index 607cc2a..0000000 --- a/cth_coro/incl/cth/coro/awaiters/wait_awaiter.hpp +++ /dev/null @@ -1,27 +0,0 @@ -#pragma once -#include "cth/coro/scheduler.hpp" -#include "cth/coro/this_coro/payload.hpp" -#include "cth/coro/utility/fwd.hpp" - -#include -#include - - -namespace cth::co { -struct wait_awaiter { - - std::chrono::steady_clock::time_point timePoint; - co::scheduler const& scheduler; - - [[nodiscard]] bool await_ready() const noexcept { - return std::chrono::steady_clock::now() >= timePoint; - } - - void await_suspend(std::coroutine_handle<> h) const { - scheduler.await(timePoint, [h] { h.resume(); }); - } - - constexpr void await_resume() const noexcept {} -}; - -} diff --git a/cth_coro/incl/cth/coro/context.hpp b/cth_coro/incl/cth/coro/context.hpp new file mode 100644 index 0000000..d90bd1b --- /dev/null +++ b/cth_coro/incl/cth/coro/context.hpp @@ -0,0 +1,92 @@ +#pragma once +#include +#include + +#include +#include +#include + + +namespace cth::co { +/** + * where coroutines run and wait + * @details implemented by @ref scheduler and @ref strand, @ref executor and the awaiters only use this interface + */ +class context { +public: + using void_func = std::move_only_function; + using await_func = std::move_only_function; + using native_handle = os::native_handle_t; + + /** + * runs the work on this context + */ + virtual void post(void_func work) const = 0; + + /** + * calls the callback on this context once the handle is signaled + * @details the callback gets the result of the wait, an empty error_code -> signaled + * @param cb to call + */ + virtual void await(native_handle, await_func cb) const = 0; + /** + * calls the callback on this context once the time point is reached + * @details the callback gets the result of the wait, an empty error_code -> reached + * @param time_point to wait for + * @param cb to call + */ + virtual void await(chrono::time_point_t time_point, await_func cb) const = 0; + + /** + * true if the calling thread runs a job of this context + * @details a scheduler also owns the threads that run the jobs of its strands + */ + [[nodiscard]] virtual bool owns_thread() const = 0; + + /** + * gets the context whose job runs on the calling thread + * @return context of the job, nullptr -> no job of a context + */ + [[nodiscard]] static context const* current() noexcept { return _current; } + +protected: + context() = default; + ~context() = default; + + context(context const& other) = default; + context& operator=(context const& other) = default; + context(context&& other) noexcept = default; + context& operator=(context&& other) noexcept = default; + + /** + * wraps the work: @ref context::current() is this context while the work runs + * @details restores the previous context after the work, also if the work throws + * @details forwards the arguments of the call to the work + * @param work to wrap + * @return wrapped work + */ + template + [[nodiscard]] auto scoped(Work work) const { + return [self = this, work = std::move(work)](Args&&... args) mutable { + current_guard const guard{self}; + work(std::forward(args)...); + }; + } + + /** + * context of the job that runs on this thread, see @ref context::current() + */ + static inline thread_local context const* _current = nullptr; + +private: + struct current_guard { + explicit current_guard(context const* ctx) noexcept : previous{std::exchange(_current, ctx)} {} + ~current_guard() { _current = previous; } + + context const* previous; + + current_guard(current_guard const&) = delete; + current_guard& operator=(current_guard const&) = delete; + }; +}; +} diff --git a/cth_coro/incl/cth/coro/exception.hpp b/cth_coro/incl/cth/coro/exception.hpp index c1ddd90..a4860b6 100644 --- a/cth_coro/incl/cth/coro/exception.hpp +++ b/cth_coro/incl/cth/coro/exception.hpp @@ -1,5 +1,6 @@ #pragma once #include +#include namespace cth::except { class coro_exception : public default_exception { @@ -12,3 +13,13 @@ class coro_exception : public default_exception { ) : default_exception{std::format("cth_coro: {}", msg), severity, loc, std::move(trace)} {} }; } + + +#define CTH_CORO_WARN(expression, fmt_message, ...) \ + CTH_WARN_T(cth::except::coro_exception, expression, fmt_message, __VA_ARGS__) + +#define CTH_CORO_STABLE_ERR(expression, fmt_message, ...) \ + CTH_STABLE_ERR_T(cth::except::coro_exception, expression, fmt_message, __VA_ARGS__) + +#define CTH_CORO_STABLE_THROW(expression, fmt_message, ...) \ + CTH_STABLE_THROW_T(cth::except::coro_exception, expression, fmt_message, __VA_ARGS__) diff --git a/cth_coro/incl/cth/coro/executor.hpp b/cth_coro/incl/cth/coro/executor.hpp index 85c7f7f..0784b79 100644 --- a/cth_coro/incl/cth/coro/executor.hpp +++ b/cth_coro/incl/cth/coro/executor.hpp @@ -1,106 +1,82 @@ #pragma once +#include "cth/coro/sync.hpp" #include "cth/coro/awaiters/data_awaiter.hpp" #include "cth/coro/awaiters/schedule_awaiter.hpp" -#include "cth/coro/func/steal.hpp" #include "cth/coro/tasks/dev/spawn_task.hpp" +#include "cth/coro/this_coro/payload.hpp" #include "cth/coro/utility/concepts.hpp" #include +#include + namespace cth::co { /** - * wraps the scheduler with coroutines functionality + * wraps the context with coroutines functionality */ class executor { public: - constexpr executor(co::scheduler const& sched) noexcept : _sched(&sched) {} + constexpr executor(co::context const& context) noexcept : _context(&context) {} constexpr ~executor() = default; /** - * calling co_await on this switches the execution context to this scheduler and stays there - * @tparam Pyld payload the caller must carry, defaulted to @ref this_coro::default_payload - */ - template - [[nodiscard]] constexpr auto schedule() const { return schedule_awaiter{scheduler()}; } - - - /** - * steals the execution context to the scheduler after the coroutine finished, keeps it as awaitable - * @tparam Pyld payload type; the scheduler is injected automatically when it is a scheduler_payload - * @tparam Awaitable must be compatible with the this_coro framework - * @param awaitable to steal context from - * @param args extra arguments to construct the payload with (besides the auto-injected scheduler) - * @return awaitable with context stolen - * @attention the awaiter does not own the task, co_await it in the same expression, do not store it + * calling co_await on this switches the execution onto this context and stays there + * @details the payload of the caller stays */ - template - [[nodiscard]] constexpr auto steal(Awaitable&& awaitable, Args&&... args) const -> awaiter_t { - if constexpr(this_coro::scheduler_payload) - return co::steal(std::forward(awaitable), Pyld{scheduler(), std::forward(args)...}); - else - return co::steal(std::forward(awaitable), Pyld{std::forward(args)...}); - } + [[nodiscard]] constexpr auto schedule() const { return schedule_awaiter{context()}; } /** - * steals the execution context to the scheduler after the coroutine finished - * @note converts the awaitable to a task - * @tparam Pyld payload type; the scheduler is injected automatically when it is a scheduler_payload - * @tparam Awaitable must not be compatible with the this_coro framework - * @param awaitable to steal context from - * @param args extra arguments to construct the payload with (besides the auto-injected scheduler) - * @return task with context stolen + * runs the task on this context once awaited, with a fresh payload + * @details the caller continues on its context, a caller without a context continues on this context + * @tparam Pyld payload of the task, constructed from the arguments + * @param task to run on this context + * @param args to construct the payload with + * @return cold spawn task, starts when awaited */ - template - [[nodiscard]] auto steal(Awaitable awaitable, Args&&... args) const -> capture_task> { - if constexpr(this_coro::scheduler_payload) - return co::steal(std::move(awaitable), Pyld{scheduler(), std::forward(args)...}); - else - return co::steal(std::move(awaitable), Pyld{std::forward(args)...}); + template + [[nodiscard]] auto spawn(Awaitable task, Args&&... args) const -> spawn_task, Pyld> { + return spawn_impl(Pyld{std::forward(args)...}, *this, std::move(task)); } /** - * runs the coroutine on the scheduler once started - * @details the caller continues on its own scheduler afterwards, with one post if the coroutine ended elsewhere, - * a caller without this_coro support continues where the coroutine ended - * @tparam Pyld payload type; the scheduler is injected automatically when it is a scheduler_payload - * @param task coroutine to run on the scheduler - * @param args extra arguments to construct the payload with (besides the auto-injected scheduler) - * @return cold spawn task; starts on the scheduler when awaited + * spawns the task and blocks until it ends + * @tparam Pyld payload of the task, constructed from the arguments + * @param task to run on this context + * @param args to construct the payload with + * @return result of the task + * @attention blocks the calling thread + * @attention waiting from a job of the own context deadlocks (a strand job, or the only worker of a scheduler) */ template - [[nodiscard]] auto spawn(Awaitable task, Args&&... args) const -> spawn_task> { - if constexpr(this_coro::scheduler_payload) - return spawn_impl(Pyld{scheduler(), std::forward(args)...}, *this, std::move(task)); - else - return spawn_impl(Pyld{std::forward(args)...}, *this, std::move(task)); + auto sync_wait(Awaitable task, Args&&... args) const -> awaited_t { + return co::sync(spawn(std::move(task), std::forward(args)...)); } private: + /** + * @param payload the promise constructor reads it, the task inherits it + * @param self the promise constructor reads it, the task runs on its context + */ template static auto spawn_impl( - Pyld payload, + [[maybe_unused]] Pyld payload, executor self, Awaitable task ) - -> spawn_task> { + -> spawn_task, Pyld> { co_await self.schedule(); - - // the final awaiter of the spawn_task resumes the caller on its own scheduler - if constexpr(this_coro_awaitable) - co_return co_await co::steal(std::move(task), payload); // injects the payload, no hop - else - co_return co_await task; + co_return co_await std::move(task); } - co::scheduler const* _sched; + co::context const* _context; public: /** - * gets underlying scheduler + * gets underlying context */ - constexpr co::scheduler const& scheduler() const { return *_sched; } + constexpr co::context const& context() const { return *_context; } constexpr bool operator==(executor const&) const = default; @@ -113,11 +89,18 @@ class executor { } +namespace cth::co::dev { +template +template +constexpr spawn_promise::spawn_promise(Pyld const& payload, executor const& self, Args&&...) : + this_coro_promise_base{self.context(), payload} {} +} + namespace cth::co::this_coro { struct [[nodiscard]] executor_tag : tag_base { - [[nodiscard]] static constexpr auto operator()(default_payload const& p) { - return data_awaiter{executor{p.scheduler()}}; + [[nodiscard]] static constexpr auto operator()(co::context const& c, auto const&) { + return data_awaiter{executor{c}}; } }; diff --git a/cth_coro/incl/cth/coro/func/dev/steal.hpp b/cth_coro/incl/cth/coro/func/dev/steal.hpp new file mode 100644 index 0000000..f94c4be --- /dev/null +++ b/cth_coro/incl/cth/coro/func/dev/steal.hpp @@ -0,0 +1,21 @@ +#pragma once +#include "cth/coro/tasks/dev/capture_task.hpp" +#include "cth/coro/utility/concepts.hpp" + +#include + +#include + +namespace cth::co::dev { +/** + * captures a foreign awaitable + * @details the capture continues the caller on its context, also after an exception + * @param awaitable to capture + * @return capture task, awaitable by this_coro coroutines + */ +template +[[nodiscard]] auto steal(Awaitable awaitable) -> capture_task> { + co_return co_await std::move(awaitable); // void: co_await, then co_return +} + +} diff --git a/cth_coro/incl/cth/coro/func/steal.hpp b/cth_coro/incl/cth/coro/func/steal.hpp deleted file mode 100644 index 24a96a5..0000000 --- a/cth_coro/incl/cth/coro/func/steal.hpp +++ /dev/null @@ -1,84 +0,0 @@ -#pragma once -#include "cth/coro/utility.hpp" -#include "cth/coro/awaiters/schedule_awaiter.hpp" -#include "cth/coro/tasks/dev/capture_task.hpp" -#include "cth/coro/this_coro/payload.hpp" -#include "cth/coro/utility/fwd.hpp" - -#include "cth/meta/coro.hpp" - -namespace cth::co { - - - -template -[[nodiscard]] auto steal(Awaitable awaitable, Pyld payload) -> capture_task> { - static_assert(has_scheduler, "foreign awaitables can only be stolen with a scheduler in the payload"); - - using result_t = awaited_t; - - if constexpr(std::same_as) { - co_await std::move(awaitable); - co_await schedule_awaiter{payload.scheduler()}; - co_return; - } else { - decltype(auto) result = co_await std::move(awaitable); - co_await schedule_awaiter{payload.scheduler()}; - co_return result; - } -} - - -/** - * injects the payload into the awaiter of the awaitable - * @attention the awaiter does not own the task, co_await it in the same expression, do not store it - */ -template -constexpr auto steal(Awaitable&& awaitable, Pyld const& payload) -> awaiter_t { - static_assert( - this_coro::awaitable_injectable_with, - "this coro awaitable not compatible with used payload type" - ); - - auto awaiter = co::extract_awaiter(std::forward(awaitable)); - awaiter.inject(payload); - - return awaiter; -} - - -/** - * injects a payload constructed from the arguments into the awaiter of the awaitable - * @attention the awaiter does not own the task, co_await it in the same expression, do not store it - */ -template -constexpr auto steal(Awaitable&& awaitable, Args&&... args) -> awaiter_t { - return co::steal(std::forward(awaitable), Pyld{std::forward(args)...}); -} - -template -[[nodiscard]] auto steal(Awaitable awaitable, Args&&... args) -> capture_task> { - return co::steal(std::move(awaitable), Pyld{std::forward(args)...}); -} - - - -template -[[nodiscard]] auto steal( - Awaitable awaitable, - scheduler const& scheduler -) -> capture_task> { - return co::steal(std::move(awaitable), this_coro::default_payload{scheduler}); -} - -/** - * injects the default payload of the scheduler into the awaiter of the awaitable - * @attention the awaiter does not own the task, co_await it in the same expression, do not store it - */ -template -constexpr auto steal(Awaitable&& awaitable, scheduler const& scheduler) -> awaiter_t { - return co::steal(std::forward(awaitable), this_coro::default_payload{scheduler}); -} - - -} diff --git a/cth_coro/incl/cth/coro/scheduler.hpp b/cth_coro/incl/cth/coro/scheduler.hpp index e2f7245..cf17386 100644 --- a/cth_coro/incl/cth/coro/scheduler.hpp +++ b/cth_coro/incl/cth/coro/scheduler.hpp @@ -3,12 +3,10 @@ #include #include +#include "context.hpp" #include "utility/fwd.hpp" -#include - -#include #include #include @@ -19,15 +17,17 @@ using autostart_t = decltype(autostart); } namespace cth::co { -class scheduler { +/** + * runs its jobs on its own worker threads + * @details @ref context on an asio::io_context + */ +class scheduler final : public context { struct Impl; + friend class strand; static inline thread_local Impl* _threadScheduler = nullptr; public: - using native_handle = cth::os::native_handle_t; - using void_func = std::move_only_function; - // ReSharper disable once CppNonExplicitConvertingConstructor /** * constructs the scheduler with workers @@ -48,10 +48,10 @@ class scheduler { */ ~scheduler(); - void post(void_func work) const; + void post(void_func work) const override; - void await(native_handle, void_func cb) const; - void await(chrono::time_point_t time_point, void_func cb) const; + void await(native_handle, await_func cb) const override; + void await(chrono::time_point_t time_point, await_func cb) const override; void start(); @@ -73,7 +73,7 @@ class scheduler { std::vector _workers{}; public: - [[nodiscard]] bool owns_thread() const { return _threadScheduler == _impl.get(); } + [[nodiscard]] bool owns_thread() const override { return _threadScheduler == _impl.get(); } [[nodiscard]] size_t workers() const { return _workers.size(); } [[nodiscard]] size_t active_workers() const { diff --git a/cth_coro/incl/cth/coro/strand.hpp b/cth_coro/incl/cth/coro/strand.hpp new file mode 100644 index 0000000..11c35ee --- /dev/null +++ b/cth_coro/incl/cth/coro/strand.hpp @@ -0,0 +1,44 @@ +#pragma once +#include "cth/coro/context.hpp" +#include "cth/coro/utility/fwd.hpp" + +#include + + +namespace cth::co { +/** + * serial queue on a @ref scheduler + * @details its jobs run one at a time and in post order on the workers of the scheduler, + * a coroutine holds the strand between two suspension points, not across a co_await + * @attention the scheduler must outlive the strand + */ +class strand final : public context { + struct Impl; + +public: + /** + * constructs the strand on the scheduler + * @param scheduler to run the jobs on + */ + explicit strand(scheduler const& scheduler); + ~strand(); + + void post(void_func work) const override; + + void await(native_handle, await_func cb) const override; + void await(chrono::time_point_t time_point, await_func cb) const override; + +private: + [[nodiscard]] Impl& impl() const { return *_impl; } + + std::unique_ptr _impl; + +public: + [[nodiscard]] bool owns_thread() const override; + + strand(strand const& other) = delete; + strand& operator=(strand const& other) = delete; + strand(strand&& other) noexcept; + strand& operator=(strand&& other) noexcept; +}; +} diff --git a/cth_coro/incl/cth/coro/tasks/awaiters/dev/affine_final_promise_awaiter.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/dev/affine_final_promise_awaiter.hpp new file mode 100644 index 0000000..7bcf709 --- /dev/null +++ b/cth_coro/incl/cth/coro/tasks/awaiters/dev/affine_final_promise_awaiter.hpp @@ -0,0 +1,33 @@ +#pragma once +#include "cth/coro/context.hpp" + +#include + + +namespace cth::co::dev { +/** + * final awaiter, continues the parent on its context + * @details transfers to the parent if @ref context::current() is the context of the parent, else posts the resume + * onto it, runs after co_return and after an exception + * @attention the post can throw inside noexcept: terminates + */ +struct affine_final_promise_awaiter { + [[nodiscard]] constexpr bool await_ready() noexcept { return false; } + + /** + * @param h of the ending task, its promise derives @ref affine_promise + * @return parent if @ref context::current() is the context of the parent, else std::noop_coroutine() + */ + template + std::coroutine_handle<> await_suspend(std::coroutine_handle h) noexcept { + auto const continuation = h.promise().continuation; + auto const* target = h.promise().callerContext; + if(context::current() == target) + return continuation; + target->post([continuation] { continuation.resume(); }); // nothing touches the frame after the post + return std::noop_coroutine(); + } + + constexpr void await_resume() noexcept {} +}; +} diff --git a/cth_coro/incl/cth/coro/tasks/awaiters/dev/basic_final_promise_awaiter.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/dev/basic_final_promise_awaiter.hpp deleted file mode 100644 index 81c95b0..0000000 --- a/cth_coro/incl/cth/coro/tasks/awaiters/dev/basic_final_promise_awaiter.hpp +++ /dev/null @@ -1,18 +0,0 @@ -#pragma once -#include - -namespace cth::co::dev { -struct basic_final_promise_awaiter { - constexpr bool await_ready() noexcept { return false; } - - template - [[nodiscard]] auto await_suspend(std::coroutine_handle h) noexcept - -> std::coroutine_handle<> { - if(auto continuation = h.promise().continuation) - return continuation; - return std::noop_coroutine(); - } - - constexpr void await_resume() noexcept {} -}; -} diff --git a/cth_coro/incl/cth/coro/tasks/awaiters/dev/capture_promise_awaiter.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/dev/capture_promise_awaiter.hpp deleted file mode 100644 index 82782d1..0000000 --- a/cth_coro/incl/cth/coro/tasks/awaiters/dev/capture_promise_awaiter.hpp +++ /dev/null @@ -1,14 +0,0 @@ -#pragma once -#include "basic_promise_awaiter.hpp" - -#include "cth/coro/utility/fwd.hpp" - -namespace cth::co::dev { - -template -struct capture_promise_awaiter - : basic_promise_awaiter - , capture_awaiter_base { - using dev::basic_promise_awaiter::basic_promise_awaiter; -}; -} diff --git a/cth_coro/incl/cth/coro/tasks/awaiters/dev/spawn_final_awaiter.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/dev/spawn_final_awaiter.hpp deleted file mode 100644 index 9e91889..0000000 --- a/cth_coro/incl/cth/coro/tasks/awaiters/dev/spawn_final_awaiter.hpp +++ /dev/null @@ -1,34 +0,0 @@ -#pragma once -#include "cth/coro/scheduler.hpp" - -#include - - -namespace cth::co::dev { -/** - * final awaiter of @ref spawn_promise, resumes the awaiting coroutine on its own scheduler - * @details transfers directly if the scheduler of the caller owns the thread, else posts the resume onto it, - * a caller without a known scheduler continues where the spawned coroutine ended - * @attention terminates if the post throws, the final awaiter is noexcept - */ -struct spawn_final_awaiter { - [[nodiscard]] constexpr bool await_ready() const noexcept { return false; } - - template - [[nodiscard]] auto await_suspend(std::coroutine_handle h) const noexcept -> std::coroutine_handle<> { - auto const continuation = h.promise().continuation; - if(!continuation) - return std::noop_coroutine(); - - auto const* caller = h.promise().caller; - if(caller == nullptr || caller->owns_thread()) - return continuation; - - // the caller may destroy this coroutine as soon as it resumes, nothing touches the frame after the post - caller->post([continuation] { continuation.resume(); }); - return std::noop_coroutine(); - } - - constexpr void await_resume() const noexcept {} -}; -} diff --git a/cth_coro/incl/cth/coro/tasks/awaiters/dev/spawn_promise_awaiter.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/dev/spawn_promise_awaiter.hpp deleted file mode 100644 index a35f816..0000000 --- a/cth_coro/incl/cth/coro/tasks/awaiters/dev/spawn_promise_awaiter.hpp +++ /dev/null @@ -1,47 +0,0 @@ -#pragma once -#include "basic_promise_awaiter.hpp" - -#include "cth/coro/this_coro/payload.hpp" -#include "cth/coro/this_coro/this_coro_base.hpp" -#include "cth/coro/utility/fwd.hpp" - -#include - -#include - - -namespace cth::co::dev { -/** - * awaiter of @ref spawn_task, takes the scheduler of the caller from the injected payload - * @details the payload of the caller is not stored and not rebound, - * a caller without this_coro support injects nothing - * @tparam Promise of the spawned coroutine, needs a `caller` member - */ -template -struct spawn_promise_awaiter : this_coro_base, private basic_promise_awaiter { -private: - using base_t = basic_promise_awaiter; - -public: - using promise_type = base_t::promise_type; - - using base_t::await_ready; - using base_t::await_resume; - using base_t::base_t; - - /** - * stores the scheduler of the caller - * @param payload of the caller - */ - template - constexpr void inject(Pyld const& payload) noexcept { _caller = &payload.scheduler(); } - - [[nodiscard]] auto await_suspend(std::coroutine_handle<> caller) noexcept -> std::coroutine_handle { - base_t::handle.promise().caller = _caller; - return base_t::await_suspend(caller); - } - -private: - co::scheduler const* _caller = nullptr; -}; -} diff --git a/cth_coro/incl/cth/coro/tasks/awaiters/dev/this_coro_awaiter_base.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/dev/this_coro_awaiter_base.hpp deleted file mode 100644 index c87bb4b..0000000 --- a/cth_coro/incl/cth/coro/tasks/awaiters/dev/this_coro_awaiter_base.hpp +++ /dev/null @@ -1,53 +0,0 @@ -#pragma once -#include "cth/coro/this_coro/payload.hpp" -#include "cth/coro/this_coro/this_coro_base.hpp" - -#include - -#include - -namespace cth::co::dev { -/** - * Base for awaitables which support payload injection - * @tparam Pyld payload to work with defaulted to @ref this_coro::default_payload - */ -template -struct this_coro_awaiter_base : this_coro_base { - constexpr this_coro_awaiter_base() = default; - - /** - * Injects the payload into the promise - * - * @tparam Promise type, must be @ref executor_injectable and compatible with @ref Pyld - * @param h handle with promise to inject executor into - */ - template - void inject_payload_into(std::coroutine_handle h) { - constexpr bool injectable = this_coro::promise_injectable_with; - - static_assert( - injectable, - "this coro supporting awaitables may not be awaited by awaitables without a compatible payload" - ); - - if constexpr(injectable) - h.promise().inject(*_payload); - } - - /** - * injects the this_coro payload into this awaiter unless it already has one - * @param payload to inject - */ - constexpr void inject(Pyld const& payload) { - if(!_payload) - _payload.emplace(payload); - } - -protected: - constexpr Pyld& payload() { return *_payload; } - constexpr Pyld const& payload() const { return *_payload; } - -private: - cth::dt::optional _payload{}; -}; -} diff --git a/cth_coro/incl/cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp index beebc98..4c314a6 100644 --- a/cth_coro/incl/cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp +++ b/cth_coro/incl/cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp @@ -1,17 +1,23 @@ #pragma once #include "cth/coro/tasks/awaiters/dev/basic_promise_awaiter.hpp" -#include "cth/coro/tasks/awaiters/dev/this_coro_awaiter_base.hpp" +#include "cth/coro/utility/concepts.hpp" +#include "cth/coro/utility/fwd.hpp" #include +#include + namespace cth::co { +/** + * awaiter of the tasks of this library, the promise of the task attaches to the caller + * @details the task continues the caller on its context + */ template struct this_coro_promise_awaiter - : dev::this_coro_awaiter_base>, + : this_coro_awaiter_base, private dev::basic_promise_awaiter { private: using base_t = dev::basic_promise_awaiter; - using payload_t = Promise::payload_type; public: this_coro_promise_awaiter() = delete; @@ -24,9 +30,16 @@ struct this_coro_promise_awaiter using base_t::base_t; - template - decltype(auto) await_suspend(std::coroutine_handle caller) noexcept { - dev::this_coro_awaiter_base::inject_payload_into(base_t::handle); + /** + * attaches the task to the caller and starts it + * @details a throwing attach (a payload copy) resumes the caller with the exception, the task does not start + * @param caller coroutine that awaits the task + * @return handle of the task + */ + template + auto await_suspend(std::coroutine_handle caller) noexcept(dev::nothrow_attachable) + -> std::coroutine_handle { + base_t::handle.promise().attach(caller.promise()); return base_t::await_suspend(caller); } }; diff --git a/cth_coro/incl/cth/coro/tasks/dev/capture_task.hpp b/cth_coro/incl/cth/coro/tasks/dev/capture_task.hpp index 07ff674..1b73520 100644 --- a/cth_coro/incl/cth/coro/tasks/dev/capture_task.hpp +++ b/cth_coro/incl/cth/coro/tasks/dev/capture_task.hpp @@ -1,24 +1,27 @@ #pragma once #include "cth/coro/tasks/task_base.hpp" -#include "cth/coro/tasks/awaiters/dev/capture_promise_awaiter.hpp" -#include "cth/coro/tasks/promises/basic_promise.hpp" +#include "cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp" +#include "cth/coro/tasks/promises/affine_promise.hpp" #include namespace cth::co { namespace dev { + /** + * promise of @ref capture_task, not a this_coro promise: its body awaits the foreign awaitable as it is + */ template - struct capture_promise : basic_promise { + struct capture_promise : dev::affine_promise { capture_task get_return_object() noexcept; }; } template -class [[nodiscard]] capture_task : public task_base, dev::capture_promise_awaiter> { - using base_t = task_base, dev::capture_promise_awaiter>; +class [[nodiscard]] capture_task : public task_base, this_coro_promise_awaiter> { + using base_t = task_base, this_coro_promise_awaiter>; using base_t::base_t; friend base_t::promise_type; diff --git a/cth_coro/incl/cth/coro/tasks/dev/spawn_task.hpp b/cth_coro/incl/cth/coro/tasks/dev/spawn_task.hpp index d11de8a..6c94e66 100644 --- a/cth_coro/incl/cth/coro/tasks/dev/spawn_task.hpp +++ b/cth_coro/incl/cth/coro/tasks/dev/spawn_task.hpp @@ -1,9 +1,9 @@ #pragma once #include "cth/coro/tasks/task_base.hpp" -#include "cth/coro/tasks/awaiters/dev/spawn_final_awaiter.hpp" -#include "cth/coro/tasks/awaiters/dev/spawn_promise_awaiter.hpp" -#include "cth/coro/tasks/promises/basic_promise.hpp" +#include "cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp" +#include "cth/coro/tasks/promises/affine_promise.hpp" +#include "cth/coro/tasks/promises/this_coro_promise_base.hpp" #include "cth/coro/utility/fwd.hpp" #include @@ -11,14 +11,32 @@ namespace cth::co::dev { -template -struct spawn_promise : basic_promise { +template +struct spawn_promise : dev::affine_promise, this_coro_promise_base { /** - * scheduler of the awaiting coroutine, nullptr -> unknown + * constructs from the coroutine parameters, runs on the context of the executor + * @details defined in executor.hpp, the executor is incomplete here + * @param payload of the spawned task + * @param self executor that spawns the task */ - co::scheduler const* caller = nullptr; + template + constexpr explicit spawn_promise(Pyld const& payload, executor const& self, Args&&...); - spawn_task get_return_object() noexcept; + + spawn_task get_return_object() noexcept; + + /** + * stores the context the caller continues on, copies no payload + * @details a caller without a context continues on the context of the executor + * @param caller promise of the awaiting coroutine + */ + template + constexpr void attach(CallerPromise const& caller) noexcept { + if constexpr(this_coro_compatible) + this->callerContext = &caller.context(); + else + this->callerContext = &this->context(); + } }; } @@ -27,13 +45,15 @@ namespace cth::co { /** * task of @ref executor::spawn() - * @details this_coro awaitable: the caller continues on its own scheduler after the spawned coroutine ended, - * a caller without this_coro support continues where the spawned coroutine ended + * @details runs on the context of the executor with its own payload, + * the caller continues on its context, a caller without a context continues on the context of the executor * @tparam T result type + * @tparam Pyld payload of the spawned task */ -template -class [[nodiscard]] spawn_task : public task_base, dev::spawn_promise_awaiter> { - using base_t = task_base, dev::spawn_promise_awaiter>; +template +class [[nodiscard]] spawn_task + : public task_base, this_coro_promise_awaiter> { + using base_t = task_base, this_coro_promise_awaiter>; using base_t::base_t; friend base_t::promise_type; @@ -42,13 +62,13 @@ class [[nodiscard]] spawn_task : public task_base, dev::sp spawn_task(spawn_task const& other) = delete; spawn_task& operator=(spawn_task const& other) = delete; - spawn_task(spawn_task&& other) noexcept = default; - spawn_task& operator=(spawn_task&& other) noexcept = default; + constexpr spawn_task(spawn_task&& other) noexcept = default; + constexpr spawn_task& operator=(spawn_task&& other) noexcept = default; }; -template -spawn_task dev::spawn_promise::get_return_object() noexcept { - return spawn_task{std::coroutine_handle::from_promise(*this)}; +template +spawn_task dev::spawn_promise::get_return_object() noexcept { + return spawn_task{std::coroutine_handle::from_promise(*this)}; } } diff --git a/cth_coro/incl/cth/coro/tasks/executor_task.hpp b/cth_coro/incl/cth/coro/tasks/executor_task.hpp index 4383c41..88803f2 100644 --- a/cth_coro/incl/cth/coro/tasks/executor_task.hpp +++ b/cth_coro/incl/cth/coro/tasks/executor_task.hpp @@ -1,18 +1,48 @@ #pragma once #include "cth/coro/tasks/task_base.hpp" #include "cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp" -#include "cth/coro/tasks/promises/basic_promise.hpp" +#include "cth/coro/tasks/promises/affine_promise.hpp" #include "cth/coro/tasks/promises/this_coro_promise_base.hpp" +#include "cth/coro/this_coro/this_coro_base.hpp" #include "cth/coro/utility/fwd.hpp" +#include +#include + #include namespace cth::co::dev { template -struct executor_promise : basic_promise, this_coro_promise_base { +struct executor_promise : dev::affine_promise, this_coro_promise_base { executor_task get_return_object() noexcept; + + /** + * copies the context and the payload of the caller, the task continues the caller on its context + * @param caller promise of the awaiting coroutine + */ + template + constexpr void attach(CallerPromise const& caller) + noexcept(mta::nothrow_constructible const&>) { + static_assert( + this_coro_compatible, + "this_coro tasks need a this_coro caller, use spawn() or sync_wait()" + ); + + this->inherit(caller); + dev::affine_promise::attach(caller); + } + +private: + /** + * payload of the caller, the own payload for a caller without one (the static_assert in attach reports it) + */ + template + using caller_payload_t = mta::conditional_pack_t< + this_coro_compatible, + mta::trait_pack, + mta::type_pack>; }; } @@ -20,7 +50,9 @@ struct executor_promise : basic_promise, this_coro_promise_base { namespace cth::co { /** * Task to be spawned on the @ref executor + * @details runs on the context of the caller, the caller continues on its context afterwards * @tparam T task return type + * @tparam Pyld payload of the task, the payload of the caller must convert to it */ template class [[nodiscard]] executor_task : public task_base, this_coro_promise_awaiter> { diff --git a/cth_coro/incl/cth/coro/tasks/promises/affine_promise.hpp b/cth_coro/incl/cth/coro/tasks/promises/affine_promise.hpp new file mode 100644 index 0000000..767a7bb --- /dev/null +++ b/cth_coro/incl/cth/coro/tasks/promises/affine_promise.hpp @@ -0,0 +1,32 @@ +#pragma once +#include "basic_promise.hpp" + +#include "cth/coro/tasks/awaiters/dev/affine_final_promise_awaiter.hpp" +#include "cth/coro/utility/fwd.hpp" + +#include + + +namespace cth::co::dev { +/** + * promise of tasks that continue the parent on its context + * @details its final awaiter is @ref affine_final_promise_awaiter, it reads @ref callerContext + */ +template +struct affine_promise : basic_promise { + /** + * stores the context of the caller, the task continues the caller on it + * @param caller promise of the awaiting coroutine + */ + template + constexpr void attach(CallerPromise const& caller) noexcept { + static_assert(this_coro_compatible, "the caller must be a this_coro coroutine"); + callerContext = &caller.context(); + } + + /** + * context the parent continues on, @ref attach() sets it before the task runs + */ + co::context const* callerContext = nullptr; +}; +} diff --git a/cth_coro/incl/cth/coro/tasks/promises/basic_promise.hpp b/cth_coro/incl/cth/coro/tasks/promises/basic_promise.hpp index 65bdf51..79a0521 100644 --- a/cth_coro/incl/cth/coro/tasks/promises/basic_promise.hpp +++ b/cth_coro/incl/cth/coro/tasks/promises/basic_promise.hpp @@ -1,11 +1,10 @@ #pragma once #include "raw_promise.hpp" -#include "cth/coro/tasks/awaiters/dev/basic_final_promise_awaiter.hpp" #include namespace cth::co { -template +template struct basic_promise : raw_promise { using initial_suspend_type = decltype(InitS); using final_suspend_type = decltype(FinalS); diff --git a/cth_coro/incl/cth/coro/tasks/promises/raw_promise.hpp b/cth_coro/incl/cth/coro/tasks/promises/raw_promise.hpp index 4eda45d..59b6dd7 100644 --- a/cth_coro/incl/cth/coro/tasks/promises/raw_promise.hpp +++ b/cth_coro/incl/cth/coro/tasks/promises/raw_promise.hpp @@ -2,23 +2,34 @@ #include "cth/coro/utility/fwd.hpp" #include +#include #include +#include +#include +namespace cth::co::dev { +/** + * members of every @ref raw_promise: the exception and the continuation + */ +struct raw_promise_base { + std::exception_ptr exception; + std::coroutine_handle<> continuation = nullptr; + + constexpr void unhandled_exception() { exception = std::current_exception(); } +}; +} + namespace cth::co { template struct raw_promise : dev::raw_promise_base { using value_type = T; dt::optional result; - std::exception_ptr exception; - std::coroutine_handle<> continuation = nullptr; - - constexpr void unhandled_exception() { exception = std::current_exception(); } template - constexpr void return_value(U&& value) noexcept { + constexpr void return_value(U&& value) noexcept(mta::nothrow_constructible) { result.emplace(std::forward(value)); } }; @@ -28,10 +39,6 @@ struct raw_promise : dev::raw_promise_base { using value_type = void; dt::optional result; - std::exception_ptr exception; - std::coroutine_handle<> continuation = nullptr; - - constexpr void unhandled_exception() { exception = std::current_exception(); } constexpr void return_void() noexcept { result.emplace(); } }; diff --git a/cth_coro/incl/cth/coro/tasks/promises/this_coro_promise_base.hpp b/cth_coro/incl/cth/coro/tasks/promises/this_coro_promise_base.hpp index 5ca2b35..16e6eb2 100644 --- a/cth_coro/incl/cth/coro/tasks/promises/this_coro_promise_base.hpp +++ b/cth_coro/incl/cth/coro/tasks/promises/this_coro_promise_base.hpp @@ -1,20 +1,20 @@ #pragma once #include "cth/coro/utility.hpp" #include "cth/coro/awaiters/raw_native_handle_awaiter.hpp" -#include "cth/coro/func/steal.hpp" +#include "cth/coro/func/dev/steal.hpp" #include "cth/coro/tasks/dev/capture_task.hpp" #include "cth/coro/this_coro/payload.hpp" +#include "cth/coro/this_coro/this_coro_base.hpp" #include "cth/coro/utility/concepts.hpp" #include +#include #include #include namespace cth::co { -class executor; - template concept native_handle_waitable = requires(T const& value) { { value.native_handle() } -> std::same_as; @@ -22,74 +22,90 @@ concept native_handle_waitable = requires(T const& value) { /** - * promises supporting the libs this_coro namespace must publicly inherit from this + * base of this_coro promises, holds the context the coroutine runs on and the payload * @details - * - children must forward the payload on construction + * - promises supporting the libs this_coro namespace must publicly inherit from this + * - the awaiter of a task calls @ref inherit() before the task runs, a spawned task gets both on construction */ template struct this_coro_promise_base : this_coro_base { + /** + * @param context the coroutine runs on + * @param payload of the coroutine + */ + constexpr this_coro_promise_base(co::context const& context, Pyld const& payload) + noexcept(mta::nothrow_copy_constructible) : + _context{&context}, + _payload{payload} {} + constexpr this_coro_promise_base() noexcept = default; + + /** + * calls the tag with the context and the payload + * @return awaitable of the tag + */ template [[nodiscard]] constexpr decltype(auto) await_transform(Tag&& t) { static_assert( - requires(Tag t) { { t.operator()(std::declval()) } -> awaitable; }, - "Pyld must satisfy the tags call operator" + requires(Tag tag, co::context const& c, Pyld& p) { { std::forward(tag)(c, p) } -> awaitable; }, + "the tag must be callable with the context and the payload" ); - return std::forward(t).operator()(*_payload); + return std::forward(t)(*_context, *_payload); } - + /** + * awaits the awaitable as it is, its awaiter resumes this coroutine on its context + */ template [[nodiscard]] constexpr auto await_transform(Awaitable&& awaitable) -> awaiter_t { - static_assert( - this_coro::awaitable_injectable_with, - "awaited task carries a payload incompatible with this coroutine's payload" - ); - - auto awaiter = co::extract_awaiter(std::forward(awaitable)); - - awaiter.inject(*_payload); - - return awaiter; + return co::extract_awaiter(std::forward(awaitable)); } - template - [[nodiscard]] constexpr auto await_transform(Awaitable&& captured) { return cth::co::extract_awaiter(captured); } - - template requires (this_coro::scheduler_payload) + /** + * waits on the native handle of the waitable, resumes this coroutine on its context + */ + template [[nodiscard]] constexpr auto await_transform(Waitable const& waitable) { - return raw_native_handle_awaiter{_payload->scheduler(), waitable.native_handle()}; + return raw_native_handle_awaiter{*_context, waitable.native_handle()}; } - template - requires (this_coro::scheduler_payload && !native_handle_waitable) + /** + * captures the foreign awaitable, the capture resumes this coroutine on its context + */ + template requires (!native_handle_waitable) [[nodiscard]] auto await_transform(Awaitable awaitable) -> capture_task> { - return co::steal(std::move(awaitable), *_payload); + return dev::steal(std::move(awaitable)); } /** - * injects the this_coro default_payload into the promise - * @param payload to inject + * copies the context and the payload of the caller + * @param caller promise of the awaiting coroutine */ - constexpr void inject(Pyld const& payload) { _payload = payload; } + template + constexpr void inherit(CallerPromise const& caller) + noexcept(mta::nothrow_constructible const&>) { + static_assert( + this_coro::promise_injectable_with>, + "the payload of the caller must convert to the payload of the task" + ); - /** - * repoints the promise's payload at a new scheduler, preserving every other payload field - * @details only available for scheduler payloads; the payload is engaged on every path that reaches - * this (a this_coro co_await goes through await_transform, which derefs the payload first) - * @param scheduler the coroutine has migrated onto - */ - constexpr void rebind_scheduler(co::scheduler const& scheduler) - requires this_coro::scheduler_payload - { - _payload->rebind_scheduler(scheduler); + _context = &caller.context(); + _payload.emplace(caller.payload()); } - constexpr this_coro_promise_base(Pyld const& payload) noexcept : _payload{payload} {} - constexpr this_coro_promise_base() noexcept = default; + /** + * sets the context the coroutine runs on, keeps the payload + * @param context the coroutine moves to + */ + constexpr void rebind_context(co::context const& context) noexcept { _context = &context; } private: + co::context const* _context = nullptr; cth::dt::optional _payload{}; + +public: + [[nodiscard]] constexpr co::context const& context() const noexcept { return *_context; } + [[nodiscard]] constexpr Pyld const& payload() const noexcept { return *_payload; } }; } diff --git a/cth_coro/incl/cth/coro/tasks/sync_task.hpp b/cth_coro/incl/cth/coro/tasks/sync_task.hpp index f242db1..6ee0ba3 100644 --- a/cth_coro/incl/cth/coro/tasks/sync_task.hpp +++ b/cth_coro/incl/cth/coro/tasks/sync_task.hpp @@ -1,9 +1,7 @@ #pragma once #include "cth/coro/tasks/task_base.hpp" -#include "cth/coro/tasks/awaiters/this_coro_promise_awaiter.hpp" #include "cth/coro/tasks/awaiters/dev/basic_promise_awaiter.hpp" #include "cth/coro/tasks/promises/sync_promise_base.hpp" -#include "cth/coro/tasks/promises/this_coro_promise_base.hpp" #include "cth/coro/utility/fwd.hpp" @@ -49,12 +47,6 @@ template struct sync_promise : dev::sync_promise_template { sync_task get_return_object() noexcept; }; - - -template -struct sync_executor_promise : this_coro_promise_base, sync_promise_template { - sync_executor_task get_return_object() noexcept; -}; } namespace cth::co { @@ -66,14 +58,6 @@ namespace cth::co { template class sync_task : public dev::sync_task_template, dev::basic_promise_awaiter> {}; using sync_void_task = sync_task; - -template -class sync_executor_task : public dev::sync_task_template< - dev::sync_executor_promise, - this_coro_promise_awaiter - > {}; - -using sync_executor_void_task = sync_executor_task; } @@ -82,9 +66,4 @@ template auto sync_promise::get_return_object() noexcept -> sync_task { return {std::coroutine_handle>::from_promise(*this)}; } - -template -auto sync_executor_promise::get_return_object() noexcept -> sync_executor_task { - return {std::coroutine_handle>::from_promise(*this)}; -} } diff --git a/cth_coro/incl/cth/coro/this_coro.hpp b/cth_coro/incl/cth/coro/this_coro.hpp index 5019ee1..a5fd5d3 100644 --- a/cth_coro/incl/cth/coro/this_coro.hpp +++ b/cth_coro/incl/cth/coro/this_coro.hpp @@ -1,6 +1,6 @@ #pragma once +#include "cth/coro/this_coro/context_tag.hpp" #include "cth/coro/this_coro/native_handle_tag.hpp" #include "cth/coro/this_coro/payload.hpp" -#include "cth/coro/this_coro/scheduler_tag.hpp" #include "cth/coro/this_coro/wait_tag.hpp" diff --git a/cth_coro/incl/cth/coro/this_coro/context_tag.hpp b/cth_coro/incl/cth/coro/this_coro/context_tag.hpp new file mode 100644 index 0000000..55cee52 --- /dev/null +++ b/cth_coro/incl/cth/coro/this_coro/context_tag.hpp @@ -0,0 +1,12 @@ +#pragma once +#include "cth/coro/awaiters/data_awaiter.hpp" +#include "cth/coro/utility/fwd.hpp" + + +namespace cth::co::this_coro { +struct [[nodiscard]] context_tag : tag_base { + static constexpr auto operator()(co::context const& c, auto const&) { return data_awaiter{c}; } +}; + +inline constexpr context_tag context{}; +} diff --git a/cth_coro/incl/cth/coro/this_coro/native_handle_tag.hpp b/cth_coro/incl/cth/coro/this_coro/native_handle_tag.hpp index fdced7a..74f9aad 100644 --- a/cth_coro/incl/cth/coro/this_coro/native_handle_tag.hpp +++ b/cth_coro/incl/cth/coro/this_coro/native_handle_tag.hpp @@ -1,16 +1,12 @@ #pragma once #include "cth/coro/awaiters/raw_native_handle_awaiter.hpp" -#include "cth/coro/this_coro/payload.hpp" #include "cth/coro/utility/fwd.hpp" -#include - - namespace cth::co::this_coro { struct [[nodiscard]] native_handle_tag : tag_base { constexpr explicit native_handle_tag(os::native_handle_t h) : handle{h} {} - constexpr auto operator()(scheduler_payload auto const& p) const { - return raw_native_handle_awaiter{p.scheduler(), handle}; + constexpr auto operator()(co::context const& c, auto const&) const { + return raw_native_handle_awaiter{c, handle}; } os::native_handle_t handle; }; diff --git a/cth_coro/incl/cth/coro/this_coro/payload.hpp b/cth_coro/incl/cth/coro/this_coro/payload.hpp index 73fdfd8..5c3ccea 100644 --- a/cth_coro/incl/cth/coro/this_coro/payload.hpp +++ b/cth_coro/incl/cth/coro/this_coro/payload.hpp @@ -4,23 +4,9 @@ namespace cth::co::this_coro { /** - * class base for payloads carrying a scheduler - * @details this coro machinery will auto-inject the scheduler for types inheriting this base + * base of custom payloads, carries no data + * @details @ref default_payload, a custom payload slices to it when its task awaits a task with the default payload */ -struct scheduler_payload_base { - constexpr scheduler_payload_base(co::scheduler const& scheduler) : _scheduler{&scheduler} {} - - constexpr co::scheduler const& scheduler() const { return *_scheduler; } - - /** - * framework use: repoint the payload at the scheduler the coroutine has migrated onto - * @details keeps every derived field intact, only the scheduler is swapped - * @param scheduler the coroutine is now running on - */ - constexpr void rebind_scheduler(co::scheduler const& scheduler) noexcept { _scheduler = &scheduler; } - -private: - co::scheduler const* _scheduler; -}; +struct payload_base {}; } diff --git a/cth_coro/incl/cth/coro/this_coro/scheduler_tag.hpp b/cth_coro/incl/cth/coro/this_coro/scheduler_tag.hpp deleted file mode 100644 index 0bf7b07..0000000 --- a/cth_coro/incl/cth/coro/this_coro/scheduler_tag.hpp +++ /dev/null @@ -1,15 +0,0 @@ -#pragma once -#include "cth/coro/awaiters/data_awaiter.hpp" -#include "cth/coro/utility/fwd.hpp" -#include "payload.hpp" - - -namespace cth::co::this_coro { -struct [[nodiscard]] scheduler_tag : tag_base { - static constexpr auto operator()(scheduler_payload auto const& payload) { - return data_awaiter{payload.scheduler()}; - } -}; - -inline constexpr scheduler_tag scheduler{}; -} diff --git a/cth_coro/incl/cth/coro/this_coro/this_coro_base.hpp b/cth_coro/incl/cth/coro/this_coro/this_coro_base.hpp index 3b8d228..908b05c 100644 --- a/cth_coro/incl/cth/coro/this_coro/this_coro_base.hpp +++ b/cth_coro/incl/cth/coro/this_coro/this_coro_base.hpp @@ -1,13 +1,12 @@ #pragma once #include "cth/coro/utility/fwd.hpp" -#include +#include namespace cth::co { /** - * base for this_coro participants specifying the payload - * @details inherit from this base to mark your type this coro compatible. - * most likely has to implement an inject(payload auto const& p) + * base of this_coro promises, specifies the payload + * @details @ref this_coro_compatible detects it * @tparam Pyld payload carried through the coroutine */ template @@ -22,10 +21,6 @@ using payload_t = std::remove_cvref_t::payload_type; } namespace cth::co::this_coro { -template -concept awaitable_injectable_with = this_coro_compatible> - && pyld_compatible_with>, Pyld>; - template concept promise_injectable_with = this_coro_compatible diff --git a/cth_coro/incl/cth/coro/this_coro/wait_tag.hpp b/cth_coro/incl/cth/coro/this_coro/wait_tag.hpp index 1f213f3..bcf9985 100644 --- a/cth_coro/incl/cth/coro/this_coro/wait_tag.hpp +++ b/cth_coro/incl/cth/coro/this_coro/wait_tag.hpp @@ -1,6 +1,5 @@ #pragma once -#include "cth/coro/awaiters/wait_awaiter.hpp" -#include "cth/coro/this_coro/payload.hpp" +#include "cth/coro/awaiters/time_point_awaiter.hpp" #include "cth/coro/utility/fwd.hpp" #include @@ -9,7 +8,7 @@ namespace cth::co::this_coro { struct [[nodiscard]] wait_tag : tag_base { constexpr explicit wait_tag(chrono::time_point_t t) : timePoint{t} {} - constexpr auto operator()(scheduler_payload auto const& p) const { return wait_awaiter{timePoint, p.scheduler()}; } + constexpr auto operator()(co::context const& c, auto const&) const { return time_point_awaiter{timePoint, c}; } chrono::time_point_t timePoint; }; diff --git a/cth_coro/incl/cth/coro/utility/concepts.hpp b/cth_coro/incl/cth/coro/utility/concepts.hpp index 544a32d..69f60fb 100644 --- a/cth_coro/incl/cth/coro/utility/concepts.hpp +++ b/cth_coro/incl/cth/coro/utility/concepts.hpp @@ -6,34 +6,31 @@ #include -namespace cth::co::this_coro {} - - namespace cth::co { - -template -concept has_scheduler = requires(T const& t) { - { t.scheduler() } -> std::convertible_to; -}; - - -template -concept this_coro_awaitable = this_coro_compatible>; - +/** + * true if the awaiter of @ref T derives @ref this_coro_awaiter_base + * @details the awaiter resumes the caller on its context itself + */ template -concept captured_awaitable = co::awaitable && std::is_base_of_v>; - +concept this_coro_awaitable = + co::awaitable && std::is_base_of_v>>; +/** + * true if @ref T is awaitable, not a @ref this_coro_awaitable and not a tag + */ template -concept foreign_awaitable = - co::awaitable && !captured_awaitable && !this_coro_awaitable && !co::this_coro::tag; - -template -concept non_this_coro_awaitable = awaitable && !this_coro_awaitable; - +concept foreign_awaitable = co::awaitable && !this_coro_awaitable && !co::this_coro::tag; template concept sync_promise_type = std::is_base_of_v; +namespace dev { + /** + * true if @ref Promise attaches to @ref CallerPromise without throwing + */ + template + concept nothrow_attachable = requires(Promise& p, CallerPromise const& c) { { p.attach(c) } noexcept; }; +} + } diff --git a/cth_coro/incl/cth/coro/utility/fwd.hpp b/cth_coro/incl/cth/coro/utility/fwd.hpp index cf4d4a7..bdac442 100644 --- a/cth_coro/incl/cth/coro/utility/fwd.hpp +++ b/cth_coro/incl/cth/coro/utility/fwd.hpp @@ -3,7 +3,9 @@ #include namespace cth::co { +class context; class scheduler; +class strand; class executor; } @@ -17,20 +19,16 @@ template concept tag = std::is_base_of_v>; struct executor_tag; -struct scheduler_tag; +struct context_tag; struct wait_tag; } namespace cth::co::this_coro { +struct payload_base; - -struct scheduler_payload_base; -template -concept scheduler_payload = std::is_base_of_v>; - -using default_payload = scheduler_payload_base; +using default_payload = payload_base; } @@ -51,16 +49,17 @@ namespace dev { /** * dev only, use @ref raw_promise - * @details token marking the raw_promise + * @details base of every raw_promise, defined in raw_promise.hpp */ - struct raw_promise_base {}; + struct raw_promise_base; } template struct this_coro_base; /** - * true if @ref T opts into the this_coro payload machinery by deriving @ref this_coro_base + * true if @ref T is a this_coro promise + * @details this_coro promises derive @ref this_coro_base */ template concept this_coro_compatible = std::is_base_of_v>; @@ -100,11 +99,8 @@ class sync_task; template class executor_task; -template -class spawn_task; - template -class sync_executor_task; +class spawn_task; } @@ -114,20 +110,17 @@ class coro_exception; } namespace cth::co { -template struct schedule_awaiter; -struct wait_awaiter; +struct time_point_awaiter; -template -struct this_coro_promise_awaiter_base; template struct this_coro_promise_base; /** - * opt-in tag for awaiters that return to the right context themselves - * @details @ref this_coro_promise_base awaits them as they are, without a steal back to its payload context + * base of awaiters that resume the caller on its context themselves + * @details @ref this_coro_promise_base awaits them as they are */ -struct capture_awaiter_base {}; +struct this_coro_awaiter_base {}; struct sync_promise_base; diff --git a/cth_coro/src/scheduler.cpp b/cth_coro/src/scheduler.cpp index 5e75c03..0a3b451 100644 --- a/cth_coro/src/scheduler.cpp +++ b/cth_coro/src/scheduler.cpp @@ -1,67 +1,11 @@ #include "cth/coro/scheduler.hpp" -#include "utility/asio.hpp" -#include "utility/native_handle_helpers.hpp" -#include "utility/timer_pool.hpp" +#include "scheduler_impl.hpp" #include -#include -#include - - -namespace cth::co { - -struct scheduler::Impl { - using guard_t = asio::executor_work_guard; - - // ReSharper disable once CppNonExplicitConvertingConstructor - Impl(size_t workers) : ctx{static_cast(workers)}, timerPool{ctx} { ctx.stop(); } - - [[nodiscard]] bool active() const { return workGuard != std::nullopt || !ctx.stopped(); } - - void start() { - ctx.restart(); - workGuard.emplace(asio::make_work_guard(ctx)); - } - void request_stop() { - workGuard = std::nullopt; - ctx.stop(); - } - - void post(void_func work) { bas::post(ctx, std::move(work)); } - - void await(native_handle handle, void_func callback) { - auto handler = wrap_unique(handle, ctx); - auto& waiter = *handler; // before the move below, argument evaluation order is unspecified - async_wait_signaled( - waiter, - [ - h = std::move(handler), - cb = std::move(callback) - ](asio::error_code const& ec) mutable { - ASIO_EC_STABLE_THROW(ec, "async wait for handle [{}] failed", h->native_handle()) - - cb(); - } - ); - } - - void await(chrono::time_point_t time_point, void_func callback) { - timerPool.set(time_point, std::move(callback)); - } - - - bas::io_context ctx; - std::optional workGuard; - - timer_pool timerPool; -}; -} - - namespace cth::co { scheduler::scheduler(size_t workers) : _impl{std::make_unique(workers)}, @@ -73,20 +17,25 @@ scheduler::~scheduler() { if(_impl != nullptr) await_stop(); } -void scheduler::post(void_func work) const { impl().post(std::move(work)); } +void scheduler::post(void_func work) const { dev::post_on(impl().ctx.get_executor(), std::move(work)); } -void scheduler::await(native_handle handle, void_func cb) const { impl().await(handle, std::move(cb)); } -void scheduler::await(std::chrono::steady_clock::time_point time_point, void_func cb) const { - impl().await(time_point, std::move(cb)); +void scheduler::await(native_handle handle, await_func cb) const { + dev::await_on(impl().ctx.get_executor(), impl().ctx, handle, std::move(cb)); +} +void scheduler::await(chrono::time_point_t time_point, await_func cb) const { + dev::await_on(impl().ctx.get_executor(), impl().timerPool, time_point, std::move(cb)); } void scheduler::start() { + CTH_CRITICAL(active(), "start() on an active scheduler, the workers are still running") {} + impl().start(); for(auto& worker : _workers) worker = std::jthread( - [&ctx = impl().ctx, impl = _impl.get(), &count = *_activeWorkers] { + [&ctx = impl().ctx, impl = _impl.get(), &count = *_activeWorkers, self = this] { ++count; _threadScheduler = impl; + _current = self; ctx.run(); --count; diff --git a/cth_coro/src/scheduler_impl.hpp b/cth_coro/src/scheduler_impl.hpp new file mode 100644 index 0000000..d8f49ba --- /dev/null +++ b/cth_coro/src/scheduler_impl.hpp @@ -0,0 +1,39 @@ +#pragma once +#include "cth/coro/scheduler.hpp" + +#include "utility/asio/ops.hpp" +#include "utility/timer_pool.hpp" + + +#include +#include + +#include + + +namespace cth::co { + +struct scheduler::Impl { + using guard_t = asio::executor_work_guard; + + // ReSharper disable once CppNonExplicitConvertingConstructor + Impl(size_t workers) : ctx{static_cast(workers)}, timerPool{ctx} { ctx.stop(); } + + [[nodiscard]] bool active() const { return workGuard != std::nullopt || !ctx.stopped(); } + + void start() { + ctx.restart(); + workGuard.emplace(asio::make_work_guard(ctx)); + } + void request_stop() { + workGuard = std::nullopt; + ctx.stop(); + } + + + bas::io_context ctx; + std::optional workGuard; + + timer_pool timerPool; +}; +} diff --git a/cth_coro/src/strand.cpp b/cth_coro/src/strand.cpp new file mode 100644 index 0000000..75d6624 --- /dev/null +++ b/cth_coro/src/strand.cpp @@ -0,0 +1,42 @@ +#include "cth/coro/strand.hpp" + +#include "scheduler_impl.hpp" + + +#include + + +namespace cth::co { + +struct strand::Impl { + scheduler::Impl& parent; + bas::strand executor; +}; +} + + +namespace cth::co { + +strand::strand(scheduler const& scheduler) : + _impl{std::make_unique(scheduler.impl(), bas::make_strand(scheduler.impl().ctx))} {} +strand::~strand() = default; + +void strand::post(void_func work) const { dev::post_on(impl().executor, scoped(std::move(work))); } + +void strand::await(native_handle handle, await_func cb) const { + dev::await_on(impl().executor, impl().parent.ctx, handle, scoped(std::move(cb))); +} +void strand::await(chrono::time_point_t time_point, await_func cb) const { + dev::await_on(impl().executor, impl().parent.timerPool, time_point, scoped(std::move(cb))); +} + +bool strand::owns_thread() const { return impl().executor.running_in_this_thread(); } + +} + + +namespace cth::co { +strand::strand(strand&& other) noexcept = default; +strand& strand::operator=(strand&& other) noexcept = default; + +} diff --git a/cth_coro/src/utility/asio.hpp b/cth_coro/src/utility/asio.hpp deleted file mode 100644 index 5831046..0000000 --- a/cth_coro/src/utility/asio.hpp +++ /dev/null @@ -1,12 +0,0 @@ -#pragma once -#include - -#include "cth/coro/exception.hpp" - -namespace bas = asio; - -#define ASIO_EC_STABLE_THROW(ec, msg, ...) \ - CTH_STABLE_THROW_T(cth::except::coro_exception, ec, msg, __VA_ARGS__) { \ - details->add("message: {}", ec.message()); \ - details->add("category: {}", ec.category().name()); \ - } diff --git a/cth_coro/src/utility/asio/ops.hpp b/cth_coro/src/utility/asio/ops.hpp new file mode 100644 index 0000000..d79f7f8 --- /dev/null +++ b/cth_coro/src/utility/asio/ops.hpp @@ -0,0 +1,68 @@ +#pragma once +#include "../native_handle_helpers.hpp" +#include "../timer_pool.hpp" + +#include "cth/coro/context.hpp" + +#include +#include +#include +#include + +#include +#include + + +namespace cth::co::dev { +/** + * runs the work on the executor + * @param executor to run on + * @param work to run + */ +template +void post_on(Executor const& executor, Work&& work) { + bas::post(executor, std::forward(work)); +} + +/** + * calls the callback on the executor once the handle is signaled + * @details the callback gets the error code of the wait + * @param executor to call the callback on + * @param ctx io_context of the executor, runs the wait + * @param cb to call with the error code + * @throws cth::except::os_exception if the handle duplication fails + */ +template +void await_on(Executor const& executor, bas::io_context& ctx, context::native_handle handle, Callback&& cb) { + auto handler = wrap_unique(handle, ctx); + auto& waiter = *handler; // before the move below, argument evaluation order is unspecified + async_wait_signaled( + waiter, + bas::bind_executor( + executor, + [ + h = std::move(handler), // keeps the descriptor alive until the completion + cb = std::forward(cb) + ](std::error_code const& ec) mutable { cb(ec); } + ) + ); +} + +/** + * calls the callback on the executor once the time point is reached + * @details the timer callback dispatches the callback onto the executor, the callback gets the error code of the wait + * @param executor to call the callback on + * @param pool timer pool of the io_context of the executor + * @param time_point to wait for + * @param cb to call with the error code + */ +template +void await_on(Executor const& executor, timer_pool& pool, chrono::time_point_t time_point, Callback&& cb) { + pool.set( + time_point, + [executor, cb = std::forward(cb)](std::error_code const& ec) mutable { + bas::dispatch(executor, [cb = std::move(cb), ec]() mutable { cb(ec); }); + } + ); +} +} diff --git a/cth_coro/src/utility/timer_pool.hpp b/cth_coro/src/utility/timer_pool.hpp index c6136c8..7d86581 100644 --- a/cth_coro/src/utility/timer_pool.hpp +++ b/cth_coro/src/utility/timer_pool.hpp @@ -1,5 +1,4 @@ #pragma once -#include "asio.hpp" #include "native_handle_helpers.hpp" #include "cth/coro/utility/fwd.hpp" @@ -9,6 +8,10 @@ #include +#include +#include +#include + namespace cth::co { namespace dev { @@ -38,23 +41,26 @@ namespace dev { class timer_pool { public: - using void_func = std::move_only_function; + using await_func = std::move_only_function; using timer_t = dev::adapted_timer; timer_pool(bas::io_context& ctx) : _ctx{ctx} {} - void set(chrono::time_point_t time_point, void_func callback) { + /** + * calls the callback once the time point is reached + * @details releases the timer before the callback, the callback gets the error code of the wait + * @param time_point to wait for + * @param callback to call with the error code + */ + void set(chrono::time_point_t time_point, await_func callback) { auto& timer = new_timer(); auto& handler = timer.set(time_point); async_wait_signaled( handler, - [&handler, &timer, this, cb = std::move(callback)](asio::error_code const& ec) mutable { - ASIO_EC_STABLE_THROW(ec, "async wait for handle [{}] failed", handler.native_handle()) - + [&timer, this, cb = std::move(callback)](std::error_code const& ec) mutable { release(timer); - - cb(); + cb(ec); } ); } diff --git a/cth_coro/tests/src/awaiters/awaiters_test.cpp b/cth_coro/tests/src/awaiters/awaiters_test.cpp index 4d29144..d70f7d4 100644 --- a/cth_coro/tests/src/awaiters/awaiters_test.cpp +++ b/cth_coro/tests/src/awaiters/awaiters_test.cpp @@ -1,9 +1,13 @@ -#include "cth/coro/awaiters/wait_awaiter.hpp" +#include "cth/coro/awaiters/time_point_awaiter.hpp" #include "../test.hpp" +#include "cth/coro/awaiters/data_awaiter.hpp" +#include "cth/coro/scheduler.hpp" #include "cth/meta/coro.hpp" +#include + #define AWAITER_TEST(suite, name) CORO_EX_TEST(_awaiters, suite, name) @@ -13,10 +17,10 @@ auto const now() { return std::chrono::steady_clock::now(); } scheduler scheduler{}; -AWAITER_TEST(wait_awaiter, main) { - wait_awaiter awaiter{now(), scheduler}; +AWAITER_TEST(time_point_awaiter, main) { + time_point_awaiter awaiter{now(), scheduler}; - static_assert(awaitable, "must be awaitable"); + static_assert(awaitable, "must be awaitable"); EXPECT_TRUE(awaiter.await_ready()); @@ -27,5 +31,22 @@ AWAITER_TEST(wait_awaiter, main) { awaiter.await_resume(); } +AWAITER_TEST(data_awaiter, moves_the_value_out) { + data_awaiter awaiter{std::make_unique(3)}; + + EXPECT_TRUE(awaiter.await_ready()); + + auto const value = awaiter.await_resume(); + ASSERT_NE(value, nullptr); + EXPECT_EQ(*value, 3); +} + +AWAITER_TEST(data_awaiter, keeps_a_reference) { + int value = 3; + data_awaiter awaiter{value}; + + EXPECT_EQ(&awaiter.await_resume(), &value); +} + } diff --git a/cth_coro/tests/src/awaiters/base/native_handle_awaiter_base_test.cpp b/cth_coro/tests/src/awaiters/base/native_handle_awaiter_base_test.cpp index 745bbcb..ef50233 100644 --- a/cth_coro/tests/src/awaiters/base/native_handle_awaiter_base_test.cpp +++ b/cth_coro/tests/src/awaiters/base/native_handle_awaiter_base_test.cpp @@ -3,6 +3,7 @@ #include "../../test.hpp" #include +#include #include #include #include @@ -13,10 +14,10 @@ namespace cth::co { namespace { -struct test_native_handle_awaiter : native_handle_awaiter_base<> { - using native_handle_awaiter_base<>::native_handle_awaiter_base; +struct test_native_handle_awaiter : native_handle_awaiter_base { + using native_handle_awaiter_base::native_handle_awaiter_base; - static constexpr void await_resume() noexcept {} + void await_resume() const { native_handle_awaiter_base::await_resume(); } }; auto await_handle(os::native_handle_t handle, std::atomic_bool const& signaled) -> diff --git a/cth_coro/tests/src/awaiters/base/osfence_awaiter_base_test.cpp b/cth_coro/tests/src/awaiters/base/osfence_awaiter_base_test.cpp index cbb6765..1ab075a 100644 --- a/cth_coro/tests/src/awaiters/base/osfence_awaiter_base_test.cpp +++ b/cth_coro/tests/src/awaiters/base/osfence_awaiter_base_test.cpp @@ -3,6 +3,7 @@ #include "../../test.hpp" #include +#include #include #include @@ -12,10 +13,10 @@ namespace cth::co { namespace { -struct test_osfence_awaiter : osfence_awaiter_base<> { - using osfence_awaiter_base<>::osfence_awaiter_base; +struct test_osfence_awaiter : osfence_awaiter_base { + using osfence_awaiter_base::osfence_awaiter_base; - static constexpr void await_resume() noexcept {} + void await_resume() const { osfence_awaiter_base::await_resume(); } }; auto await_fence(os::fence const& fence, std::atomic_bool const& signaled) -> diff --git a/cth_coro/tests/src/awaiters/base/wait_awaiter_base_test.cpp b/cth_coro/tests/src/awaiters/base/wait_awaiter_base_test.cpp new file mode 100644 index 0000000..ab8490c --- /dev/null +++ b/cth_coro/tests/src/awaiters/base/wait_awaiter_base_test.cpp @@ -0,0 +1,57 @@ +#include "cth/coro/awaiters/base/wait_awaiter_base.hpp" + +#include "../../test.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace cth::co { +namespace { + // skips the ready check, the wait of the context always runs + struct forced_wait_awaiter : native_handle_awaiter_base { + using native_handle_awaiter_base::native_handle_awaiter_base; + + [[nodiscard]] static constexpr bool await_ready() noexcept { return false; } + }; + + struct file_closer { + void operator()(std::FILE* file) const noexcept { std::fclose(file); } + }; + + auto await_failing(os::native_handle_t handle) -> executor_task { + bool caught = false; + try { + co_await forced_wait_awaiter{handle}; + } catch(except::coro_exception const&) { + caught = true; + } + co_return caught; + } +} + +// epoll does not support regular files: the wait on one fails with operation_not_supported +CORO_TEST(wait_awaiter_base, failed_wait_throws_in_caller) { +#ifdef CTH_FS_POSIX + std::unique_ptr const file{std::tmpfile()}; + ASSERT_NE(file, nullptr); + os::native_handle_t const handle = fileno(file.get()); + + scheduler sched{autostart, 1}; + strand serial{sched}; + + EXPECT_TRUE(executor{sched}.sync_wait(await_failing(handle))); + EXPECT_TRUE(executor{serial}.sync_wait(await_failing(handle))); +#else + GTEST_SKIP() << "needs a handle that the wait rejects"; +#endif +} + +} diff --git a/cth_coro/tests/src/awaiters/raw_native_handle_awaiter_test.cpp b/cth_coro/tests/src/awaiters/raw_native_handle_awaiter_test.cpp index 4a4a11c..1edf865 100644 --- a/cth_coro/tests/src/awaiters/raw_native_handle_awaiter_test.cpp +++ b/cth_coro/tests/src/awaiters/raw_native_handle_awaiter_test.cpp @@ -3,6 +3,7 @@ #include "../test.hpp" #include +#include #include #include #include diff --git a/cth_coro/tests/src/executor_test.cpp b/cth_coro/tests/src/executor_test.cpp index 1fc7eec..f49a0c1 100644 --- a/cth_coro/tests/src/executor_test.cpp +++ b/cth_coro/tests/src/executor_test.cpp @@ -7,36 +7,17 @@ #include #include -#include +#include #include #include #include #include +#include #include namespace cth::co { -template -auto sync_wait(executor& exec, Task task) { - using value_type = Task::promise_type::value_type; - - auto wrapper = [](executor& exec, Task t) -> sync_task { - if constexpr(mta::is_void) - co_await exec.spawn(std::move(t)); - else - co_return co_await exec.spawn(std::move(t)); - }; - - auto wrapped = wrapper(exec, std::move(task)); - wrapped.handle().resume(); - - wrapped.wait(); - - if constexpr(!mta::is_void) - return std::move(*wrapped.handle().promise().result); -} - CORO_TEST(executor, spawn_void_task) { scheduler sched{autostart, 1}; executor exec{sched}; @@ -47,7 +28,7 @@ CORO_TEST(executor, spawn_void_task) { co_return; }; - sync_wait(exec, task(ran)); + exec.sync_wait(task(ran)); EXPECT_TRUE(ran); } @@ -57,7 +38,7 @@ CORO_TEST(executor, spawn_value_task) { auto task = []() -> executor_task { co_return 42; }; - int const result = sync_wait(exec, task()); + int const result = exec.sync_wait(task()); EXPECT_EQ(result, 42); } @@ -71,7 +52,7 @@ CORO_TEST(executor, spawn_move_only_type) { co_return std::make_unique(123); }; - auto const result = sync_wait(exec, task()); + auto const result = exec.sync_wait(task()); ASSERT_TRUE(result != nullptr); EXPECT_EQ(*result, 123); } @@ -84,18 +65,63 @@ CORO_TEST(executor, spawn_execution_context) { auto task = []() -> executor_task { co_return std::this_thread::get_id(); }; - auto const workerThread = sync_wait(exec, task()); + auto const workerThread = exec.sync_wait(task()); EXPECT_NE(mainThread, workerThread); } +CORO_TEST(executor, sync_wait_value) { + scheduler sched{autostart, 1}; + executor exec{sched}; + + auto task = [](scheduler const& s) -> executor_task> { + co_return std::pair{7, s.owns_thread()}; + }; + + auto const [value, onSched] = exec.sync_wait(task(sched)); + EXPECT_EQ(value, 7); + EXPECT_TRUE(onSched); +} + +// the flag is set after a suspension: sync_wait returns only after the end of the task +CORO_TEST(executor, sync_wait_void) { + scheduler sched{autostart, 1}; + executor exec{sched}; + std::atomic endedOnSched{false}; + + auto task = [](scheduler const& s, std::atomic& flag) -> executor_task { + co_await this_coro::wait(std::chrono::microseconds{200}); + flag = s.owns_thread(); + }; + + exec.sync_wait(task(sched, endedOnSched)); + EXPECT_TRUE(endedOnSched); +} + +CORO_TEST(executor, sync_wait_exception) { + scheduler sched{autostart, 1}; + executor exec{sched}; + + auto voidTask = []() -> executor_task { + throw std::runtime_error("void boom"); + co_return; + }; + auto valueTask = []() -> executor_task { + throw std::logic_error("value boom"); + co_return 0; + }; + + EXPECT_THROW(exec.sync_wait(voidTask()), std::runtime_error); + EXPECT_THROW(exec.sync_wait(valueTask()), std::logic_error); +} + CORO_TEST(executor, context_this_coro_executor) { scheduler sched{autostart, 1}; executor expected{sched}; auto task = []() -> executor_task { co_return co_await this_coro::executor; }; - auto const actual = sync_wait(expected, task()); + auto const actual = expected.sync_wait(task()); EXPECT_EQ(actual, expected); } @@ -107,7 +133,7 @@ CORO_TEST(executor, context_nested_task_propagation) { auto root = [](auto c) -> executor_task { co_return co_await c(); }; - auto const actual = sync_wait(expected, root(child)); + auto const actual = expected.sync_wait(root(child)); EXPECT_EQ(actual, expected); } @@ -124,7 +150,7 @@ CORO_TEST(executor, context_deep_recursion) { } }; - auto const actual = sync_wait(expected, Recursive{}(10)); + auto const actual = expected.sync_wait(Recursive{}(10)); EXPECT_EQ(actual, expected); } @@ -152,6 +178,22 @@ struct foreign_value_task { awaiter operator co_await() { return {val}; } }; +// completes on the target scheduler, throws on resume +struct foreign_throwing_awaiter { + scheduler const& target; + + bool await_ready() const noexcept { return false; } + void await_suspend(std::coroutine_handle<> h) const { + target.post([h] { h.resume(); }); + } + [[noreturn]] void await_resume() const { throw std::runtime_error("foreign boom"); } +}; + +// the construction of the co_return value throws +struct throwing_value { + explicit throwing_value(int) { throw std::runtime_error("value construction"); } +}; + CORO_TEST(executor, interop_steal_foreign_void) { scheduler sched{autostart, 1}; executor exec{sched}; @@ -169,7 +211,7 @@ CORO_TEST(executor, interop_steal_foreign_void) { throw std::runtime_error("Did not return to executor thread"); }; - sync_wait(exec, task(mainID)); + exec.sync_wait(task(mainID)); } CORO_TEST(executor, interop_steal_foreign_value) { @@ -178,47 +220,12 @@ CORO_TEST(executor, interop_steal_foreign_value) { auto task = []() -> executor_task { co_return co_await foreign_value_task{99}; }; - int const res = sync_wait(exec, task()); + int const res = exec.sync_wait(task()); EXPECT_EQ(res, 99); } -CORO_TEST(executor, interop_explicit_steal_injection) { - scheduler sched{autostart, 1}; - executor expected{sched}; - - auto task = []() -> executor_task { co_return co_await this_coro::executor; }; - - auto launcher = [](executor& e, auto t) -> executor_task { co_return co_await e.steal(t()); }; - - executor const actual = sync_wait(expected, launcher(expected, task)); - - EXPECT_EQ(actual, expected); -} - -CORO_TEST(executor, interop_cross_executor_migration) { - scheduler sched{autostart, 2}; - executor intermediate{sched}; - executor expected{sched}; - - auto sub_task = []() -> executor_task { co_return co_await this_coro::executor; }; - - auto root_task = [](executor& e1, executor& e2, auto st) -> executor_task> { - auto e = co_await this_coro::executor; - if(e != e1) - co_return std::nullopt; - - co_return co_await e2.steal(st()); - }; - - auto const actual = sync_wait(intermediate, root_task(intermediate, expected, sub_task)); - EXPECT_EQ(actual, expected); -} - -// Regression for the stale-payload bug: hopping onto a *different* executor via schedule() moves the -// physical thread but never rewrites the promise's payload, so the this_coro tags keep resolving -// against the original (home) scheduler. Two distinct schedulers are required -- executor equality -// compares the scheduler pointer, so a single shared scheduler could not tell the two apart. -// Expected behaviour: after the hop, this_coro::executor reports the hop target. Fails until fixed. +// after schedule() onto another scheduler, this_coro::executor reports the target. Two schedulers: executor +// equality compares the scheduler pointer, executors on one scheduler compare equal and prove nothing. CORO_TEST(executor, context_reflects_schedule_hop) { scheduler home{autostart, 1}; scheduler other{autostart, 1}; @@ -230,28 +237,67 @@ CORO_TEST(executor, context_reflects_schedule_hop) { co_return co_await this_coro::executor; // ask "which executor am I on?" }; - auto const actual = sync_wait(homeExec, task(otherExec)); + auto const actual = homeExec.sync_wait(task(otherExec)); EXPECT_EQ(actual, otherExec); } -// Same defect observed through the scheduler tag: after the hop this_coro::scheduler must name the -// scheduler the coroutine is now running on, not the one baked into the payload at spawn. -CORO_TEST(executor, context_scheduler_reflects_schedule_hop) { +// the same through the context tag: after schedule() this_coro::context names the target, not the +// context of the spawn. +CORO_TEST(executor, context_tag_reflects_schedule_hop) { scheduler home{autostart, 1}; scheduler other{autostart, 1}; executor homeExec{home}; executor otherExec{other}; - auto task = [](executor target) -> executor_task { + auto task = [](executor target) -> executor_task { co_await target.schedule(); - scheduler const& current = co_await this_coro::scheduler; + context const& current = co_await this_coro::context; co_return ¤t; }; - auto const* actual = sync_wait(homeExec, task(otherExec)); + auto const* actual = homeExec.sync_wait(task(otherExec)); EXPECT_EQ(actual, &other); } +// schedule() onto the context of the caller continues without a post: the job posted before runs later +CORO_TEST(executor, context_schedule_onto_own_context_continues_without_post) { + std::atomic posted{false}; + scheduler sched{autostart, 1}; + executor exec{sched}; + + auto task = [](executor e, std::atomic& flag) -> executor_task { + e.context().post([&flag] { flag = true; }); + co_await e.schedule(); + co_return !flag; + }; + + EXPECT_TRUE(exec.sync_wait(task(exec, posted))); +} + +// a child that moves to another scheduler continues the caller on the scheduler of the caller +CORO_TEST(executor, context_caller_continues_on_own_scheduler_after_child_schedule) { + scheduler home{autostart, 1}; + scheduler other{autostart, 1}; + executor homeExec{home}; + + auto child = [](executor target) -> executor_task { + co_await target.schedule(); + co_return target.context().owns_thread(); + }; + + auto task = [](scheduler const& h, executor o, auto c) + -> executor_task> { + bool const childOnOther = co_await c(o); + context const& current = co_await this_coro::context; + co_return std::tuple{childOnOther, h.owns_thread(), ¤t}; + }; + + auto const [childOnOther, onHome, current] = homeExec.sync_wait(task(home, executor{other}, child)); + EXPECT_TRUE(childOnOther); + EXPECT_TRUE(onHome); + EXPECT_EQ(current, &home); +} + // spawn() runs the child on its scheduler, the caller continues on its own scheduler afterwards CORO_TEST(executor, interop_spawn_caller_continues_on_own_scheduler) { scheduler home{autostart, 1}; @@ -261,13 +307,13 @@ CORO_TEST(executor, interop_spawn_caller_continues_on_own_scheduler) { auto child = [](scheduler const& target) -> executor_task { co_return target.owns_thread(); }; auto task = [](scheduler const& h, scheduler const& o, auto c) - -> executor_task> { + -> executor_task> { bool const childOnOther = co_await executor{o}.spawn(c(o)); - scheduler const& current = co_await this_coro::scheduler; + context const& current = co_await this_coro::context; co_return std::tuple{childOnOther, h.owns_thread(), ¤t}; }; - auto const [childOnOther, onHome, current] = sync_wait(homeExec, task(home, other, child)); + auto const [childOnOther, onHome, current] = homeExec.sync_wait(task(home, other, child)); EXPECT_TRUE(childOnOther); EXPECT_TRUE(onHome); EXPECT_EQ(current, &home); @@ -284,7 +330,7 @@ CORO_TEST(executor, interop_spawn_on_own_scheduler) { co_return std::pair{childOnHome, h.owns_thread()}; }; - auto const [childOnHome, onHome] = sync_wait(homeExec, task(homeExec, home, child)); + auto const [childOnHome, onHome] = homeExec.sync_wait(task(homeExec, home, child)); EXPECT_TRUE(childOnHome); EXPECT_TRUE(onHome); } @@ -305,7 +351,7 @@ CORO_TEST(executor, interop_spawn_void_caller_continues_on_own_scheduler) { co_return h.owns_thread(); }; - EXPECT_TRUE(sync_wait(homeExec, task(home, other, childOnOther, child))); + EXPECT_TRUE(homeExec.sync_wait(task(home, other, childOnOther, child))); EXPECT_TRUE(childOnOther); } @@ -326,7 +372,7 @@ CORO_TEST(executor, interop_spawn_child_ends_elsewhere) { co_return std::pair{afterThird, h.owns_thread()}; }; - auto const [afterThird, afterHome] = sync_wait(homeExec, task(home, executor{other}, executor{third}, child)); + auto const [afterThird, afterHome] = homeExec.sync_wait(task(home, executor{other}, executor{third}, child)); EXPECT_TRUE(afterThird); EXPECT_TRUE(afterHome); } @@ -351,7 +397,7 @@ CORO_TEST(executor, interop_spawn_exception_reaches_caller_on_own_scheduler) { co_return std::pair{caught, h.owns_thread()}; }; - auto const [caught, onHome] = sync_wait(homeExec, task(home, executor{other}, child)); + auto const [caught, onHome] = homeExec.sync_wait(task(home, executor{other}, child)); EXPECT_TRUE(caught); EXPECT_TRUE(onHome); } @@ -376,46 +422,32 @@ CORO_TEST(executor, interop_spawn_nested_each_level_on_own_scheduler) { }; auto const [grandchildOnInner, childOnMiddle, onHome] = - sync_wait(homeExec, task(home, middle, inner, child, grandchild)); + homeExec.sync_wait(task(home, middle, inner, child, grandchild)); EXPECT_TRUE(grandchildOnInner); EXPECT_TRUE(childOnMiddle); EXPECT_TRUE(onHome); } -// a caller without this_coro support has no scheduler to return to: it continues where the child ended -CORO_TEST(executor, interop_spawn_caller_without_payload_continues_where_child_ended) { +// a caller without a scheduler continues on the scheduler of the executor, also if the child ends elsewhere +CORO_TEST(executor, interop_spawn_caller_without_scheduler_continues_on_executor_scheduler) { scheduler other{autostart, 1}; + scheduler third{autostart, 1}; - auto child = []() -> executor_task { co_return; }; - - auto task = [](scheduler const& o, auto c) -> sync_task { - co_await executor{o}.spawn(c()); - co_return o.owns_thread(); + auto child = [](executor end) -> executor_task { + co_await end.schedule(); + co_return end.context().owns_thread(); }; - auto waiter = task(other, child); - waiter.start(); - EXPECT_TRUE(waiter.await()); -} - -// the capture_task of steal() owns its payload: the payload of steal(foreign, scheduler) is a temporary -CORO_TEST(executor, interop_captured_steal_continues_on_steal_context) { - scheduler home{autostart, 1}; - scheduler other{autostart, 1}; - executor homeExec{home}; - - auto task = [](scheduler const& o) -> executor_task> { - co_await co::steal(foreign_task{}, o); - bool const voidOnOther = o.owns_thread(); - - int const value = co_await co::steal(foreign_value_task{7}, o); - co_return std::tuple{voidOnOther, value, o.owns_thread()}; + auto task = [](scheduler const& o, executor t, auto c) -> sync_task> { + bool const childOnThird = co_await executor{o}.spawn(c(t)); + co_return std::pair{childOnThird, o.owns_thread()}; }; - auto const [voidOnOther, value, valueOnOther] = sync_wait(homeExec, task(other)); - EXPECT_TRUE(voidOnOther); - EXPECT_EQ(value, 7); - EXPECT_TRUE(valueOnOther); + auto waiter = task(other, executor{third}, child); + waiter.start(); + auto const [childOnThird, onOther] = waiter.await(); + EXPECT_TRUE(childOnThird); + EXPECT_TRUE(onOther); } CORO_TEST(executor, errors_exception_in_spawned_task) { @@ -437,7 +469,7 @@ CORO_TEST(executor, errors_exception_in_spawned_task) { co_return caught; }; - bool const caught = sync_wait(exec, wrapper(task)); + bool const caught = exec.sync_wait(wrapper(task)); EXPECT_TRUE(caught); } @@ -462,41 +494,102 @@ CORO_TEST(executor, errors_exception_in_nested_task) { co_return caught; }; - bool const caught = sync_wait(exec, wrapper(root, child)); + bool const caught = exec.sync_wait(wrapper(root, child)); EXPECT_TRUE(caught); } +// a child that throws after schedule() onto another scheduler: the parent catches on its own scheduler +CORO_TEST(executor, errors_parent_catches_on_own_scheduler_after_child_schedule) { + scheduler home{autostart, 1}; + scheduler other{autostart, 1}; + executor homeExec{home}; + + auto child = [](executor target) -> executor_task { + co_await target.schedule(); + throw std::runtime_error("child boom"); + }; + + auto task = [](scheduler const& h, executor o, auto c) + -> executor_task> { + bool caughtOnHome = false; + try { + co_await c(o); + } catch(std::runtime_error const&) { + caughtOnHome = h.owns_thread(); + } + context const& current = co_await this_coro::context; + co_return std::tuple{caughtOnHome, h.owns_thread(), ¤t}; + }; + + auto const [caughtOnHome, onHome, current] = homeExec.sync_wait(task(home, executor{other}, child)); + EXPECT_TRUE(caughtOnHome); + EXPECT_TRUE(onHome); + EXPECT_EQ(current, &home); +} + +// a foreign awaitable that completes on another scheduler and throws: the caller catches on its own scheduler +CORO_TEST(executor, errors_foreign_exception_reaches_caller_on_own_scheduler) { + scheduler home{autostart, 1}; + scheduler other{autostart, 1}; + executor homeExec{home}; + + auto task = [](scheduler const& h, scheduler const& o) -> executor_task { + bool caughtOnHome = false; + try { + co_await foreign_throwing_awaiter{o}; + } catch(std::runtime_error const&) { + caughtOnHome = h.owns_thread(); + } + co_return caughtOnHome; + }; + + EXPECT_TRUE(homeExec.sync_wait(task(home, other))); +} + +// a co_return value whose construction throws: the caller catches the exception +CORO_TEST(executor, errors_throwing_return_value_reaches_caller) { + scheduler sched{autostart, 1}; + executor exec{sched}; + + auto child = []() -> executor_task { co_return 1; }; + + auto task = [](auto c) -> executor_task { + bool caught = false; + try { + (void) co_await c(); + } catch(std::runtime_error const&) { + caught = true; + } + co_return caught; + }; + + EXPECT_TRUE(exec.sync_wait(task(child))); +} + CORO_TEST(executor, stress_concurrent_spawns) { scheduler sched{autostart, 4}; executor exec{sched}; constexpr int count = 100; - std::vector> tasks; - tasks.reserve(count); - std::atomic completed{0}; - std::latch allDone{count}; - auto increment_task = [&](std::atomic& ctr) -> executor_task { - // ref capture is fine bc no co_await + auto increment_task = [](std::atomic& ctr) -> executor_task { ctr.fetch_add(1, std::memory_order_relaxed); - allDone.count_down(); co_return; }; - auto spawner = [&](std::vector>& t) -> executor_task { - // ref capture fine bc no co_await - for(int i = 0; i < count; ++i) { - auto h = exec.spawn(increment_task(completed)); - h.handle().resume(); - t.push_back(std::move(h)); - } + // starts the spawn now, waits later + auto start = [](spawn_task t) -> sync_task { co_await t; }; - allDone.wait(); - co_return; - }; + std::vector> tasks; + tasks.reserve(count); + for(int i = 0; i < count; ++i) { + tasks.push_back(start(exec.spawn(increment_task(completed)))); + tasks.back().start(); + } - sync_wait(exec, spawner(tasks)); + for(auto const& task : tasks) + task.wait(); EXPECT_EQ(completed.load(), count); } diff --git a/cth_coro/tests/src/scheduler_test.cpp b/cth_coro/tests/src/scheduler_test.cpp index 0d93f44..14fffd0 100644 --- a/cth_coro/tests/src/scheduler_test.cpp +++ b/cth_coro/tests/src/scheduler_test.cpp @@ -158,6 +158,29 @@ CORO_TEST(scheduler, owns_thread_in_task) { EXPECT_EQ(successCount.load(), taskCount); } +CORO_TEST(scheduler, current_is_the_scheduler_of_the_job) { + scheduler sched{autostart, 2}; + scheduler other{autostart, 1}; + + std::atomic seenOnSched{nullptr}; + std::atomic seenOnOther{nullptr}; + std::latch allDone{2}; + + sched.post([&] { + seenOnSched = context::current(); + allDone.count_down(); + }); + other.post([&] { + seenOnOther = context::current(); + allDone.count_down(); + }); + + allDone.wait(); + EXPECT_EQ(seenOnSched.load(), &sched); + EXPECT_EQ(seenOnOther.load(), &other); + EXPECT_EQ(context::current(), nullptr); +} + CORO_TEST(scheduler, task_ordering_single_worker) { scheduler sched(1); int val = 0; diff --git a/cth_coro/tests/src/strand_test.cpp b/cth_coro/tests/src/strand_test.cpp new file mode 100644 index 0000000..448a047 --- /dev/null +++ b/cth_coro/tests/src/strand_test.cpp @@ -0,0 +1,281 @@ +#include "test.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace cth::co { + +namespace { + constexpr auto TIMEOUT = std::chrono::seconds{5}; +} + +CORO_TEST(strand, jobs_run_one_at_a_time_in_post_order) { + constexpr size_t workers = 4; + constexpr size_t jobs = 1000; + + std::atomic running{0}; + std::atomic maxRunning{0}; + std::vector order{}; + order.reserve(jobs); + std::latch done{jobs}; + + scheduler sched{autostart, workers}; + strand serial{sched}; + + for(size_t i = 0; i < jobs; ++i) + serial.post([&, i] { + auto const now = running.fetch_add(1) + 1; + auto max = maxRunning.load(); + while(max < now && !maxRunning.compare_exchange_weak(max, now)) {} + + std::this_thread::yield(); // widens the window for an overlap + order.push_back(i); // the strand serializes the access + + running.fetch_sub(1); + done.count_down(); + }); + + done.wait(); + + EXPECT_FALSE(serial.owns_thread()); + EXPECT_EQ(maxRunning.load(), 1); + EXPECT_TRUE(std::ranges::equal(order, std::views::iota(size_t{0}, jobs))); +} + +CORO_TEST(strand, await_handle_calls_inside_strand) { + os::fence fence{}; + std::atomic inStrand{false}; + std::atomic failed{true}; + atomic_fence done{}; + + scheduler sched{autostart, 4}; + strand serial{sched}; + + serial.await( + fence.native_handle(), + [&](std::error_code ec) { + inStrand = serial.owns_thread(); + failed = static_cast(ec); + done.signal(); + } + ); + fence.signal(); + + done.wait(); + EXPECT_TRUE(inStrand); + EXPECT_FALSE(failed); +} + +CORO_TEST(strand, await_time_point_calls_inside_strand) { + std::atomic inStrand{false}; + std::atomic failed{true}; + atomic_fence done{}; + + scheduler sched{autostart, 4}; + strand serial{sched}; + + serial.await( + chrono::clock_t::now() + std::chrono::milliseconds{1}, + [&](std::error_code ec) { + inStrand = serial.owns_thread(); + failed = static_cast(ec); + done.signal(); + } + ); + + done.wait(); + EXPECT_TRUE(inStrand); + EXPECT_FALSE(failed); +} + +CORO_TEST(strand, schedule_continues_inside_strand) { + scheduler sched{autostart, 4}; + strand serial{sched}; + + auto task = [](strand const& target) -> executor_task> { + bool const before = target.owns_thread(); + co_await executor{target}.schedule(); + co_return std::pair{before, target.owns_thread()}; + }; + + auto const [before, after] = executor{sched}.sync_wait(task(serial)); + EXPECT_FALSE(before); + EXPECT_TRUE(after); +} + +CORO_TEST(strand, spawned_task_waits_inside_strand) { + scheduler sched{autostart, 4}; + strand serial{sched}; + + auto task = []() -> executor_task> { + co_await this_coro::wait(std::chrono::microseconds{200}); + + context const& current = co_await this_coro::context; + co_return std::pair{¤t, current.owns_thread()}; + }; + + auto const [current, inStrand] = executor{serial}.sync_wait(task()); + EXPECT_EQ(current, &serial); + EXPECT_TRUE(inStrand); +} + +CORO_TEST(strand, schedule_onto_scheduler_leaves_strand) { + std::binary_semaphore nextJobRan{0}; + + scheduler sched{autostart, 2}; + strand serial{sched}; + + auto task = [](strand const& s, executor target, std::binary_semaphore& jobRan) + -> executor_task> { + s.post([&jobRan] { jobRan.release(); }); // queued behind the current job + + co_await target.schedule(); + + bool const ranBeside = jobRan.try_acquire_for(TIMEOUT); + context const& current = co_await this_coro::context; + co_return std::tuple{ranBeside, s.owns_thread(), ¤t}; + }; + + auto const [ranBeside, inStrand, current] = executor{serial}.sync_wait(task(serial, sched, nextJobRan)); + EXPECT_TRUE(ranBeside); + EXPECT_FALSE(inStrand); + EXPECT_EQ(current, &sched); +} + +CORO_TEST(strand, spawn_onto_scheduler_leaves_strand) { + std::binary_semaphore nextJobRan{0}; + + scheduler sched{autostart, 2}; + strand serial{sched}; + + auto child = [](strand const& s, std::binary_semaphore& jobRan) -> executor_task> { + bool const ranBeside = jobRan.try_acquire_for(TIMEOUT); + co_return std::pair{ranBeside, s.owns_thread()}; + }; + + auto task = [](strand const& s, executor target, std::binary_semaphore& jobRan, auto c) + -> executor_task> { + s.post([&jobRan] { jobRan.release(); }); // queued behind the current job + + co_return co_await target.spawn(c(s, jobRan)); + }; + + auto const [ranBeside, inStrand] = executor{serial}.sync_wait(task(serial, sched, nextJobRan, child)); + EXPECT_TRUE(ranBeside); + EXPECT_FALSE(inStrand); +} + +CORO_TEST(strand, current_is_the_context_of_the_job) { + std::atomic inStrand{nullptr}; + std::atomic inScheduler{nullptr}; + std::latch done{2}; + + scheduler sched{autostart, 2}; + strand serial{sched}; + + serial.post([&] { + inStrand = context::current(); + done.count_down(); + }); + sched.post([&] { + inScheduler = context::current(); + done.count_down(); + }); + done.wait(); + + EXPECT_EQ(context::current(), nullptr); + EXPECT_EQ(inStrand.load(), &serial); + EXPECT_EQ(inScheduler.load(), &sched); +} + +CORO_TEST(strand, caller_continues_outside_the_strand_its_child_ended_in) { + scheduler sched{autostart, 2}; + strand serial{sched}; + + auto child = [](executor end) -> executor_task { co_await end.schedule(); }; + + auto task = [](scheduler const& b, strand const& s, auto c) + -> executor_task> { + co_await executor{b}.spawn(c(executor{s})); + context const& current = co_await this_coro::context; + co_return std::tuple{s.owns_thread(), b.owns_thread(), ¤t}; + }; + + auto const [inStrand, onScheduler, current] = executor{sched}.sync_wait(task(sched, serial, child)); + EXPECT_FALSE(inStrand); + EXPECT_TRUE(onScheduler); + EXPECT_EQ(current, &sched); +} + +CORO_TEST(strand, caller_in_strand_continues_in_strand_after_spawn) { + scheduler sched{autostart, 2}; + strand serial{sched}; + + auto child = [](scheduler const& b, strand const& s) -> executor_task> { + co_return std::pair{b.owns_thread(), s.owns_thread()}; + }; + + auto task = [](scheduler const& b, strand const& s, auto c) + -> executor_task> { + auto const [childOnScheduler, childInStrand] = co_await executor{b}.spawn(c(b, s)); + context const& current = co_await this_coro::context; + co_return std::tuple{childOnScheduler, childInStrand, s.owns_thread(), ¤t}; + }; + + auto const [childOnScheduler, childInStrand, inStrand, current] = + executor{serial}.sync_wait(task(sched, serial, child)); + EXPECT_TRUE(childOnScheduler); + EXPECT_FALSE(childInStrand); + EXPECT_TRUE(inStrand); + EXPECT_EQ(current, &serial); +} + +CORO_TEST(strand, scheduler_jobs_run_beside_strand_job) { + std::binary_semaphore strandRunning{0}; + std::binary_semaphore schedulerRan{0}; + std::binary_semaphore strandDone{0}; + std::atomic overlapped{false}; + + scheduler sched{autostart, 2}; + strand serial{sched}; + + serial.post([&] { + strandRunning.release(); + overlapped = schedulerRan.try_acquire_for(TIMEOUT); + strandDone.release(); + }); + + ASSERT_TRUE(strandRunning.try_acquire_for(TIMEOUT)); + sched.post([&] { schedulerRan.release(); }); + + ASSERT_TRUE(strandDone.try_acquire_for(2 * TIMEOUT)); + EXPECT_TRUE(overlapped); +} + +CORO_TEST(strand, move_only) { + EXPECT_FALSE(std::is_copy_constructible_v); + EXPECT_FALSE(std::is_copy_assignable_v); + EXPECT_TRUE(std::is_nothrow_move_constructible_v); + EXPECT_TRUE(std::is_nothrow_move_assignable_v); +} + +} diff --git a/cth_coro/tests/src/tasks/sync_task_test.cpp b/cth_coro/tests/src/tasks/sync_task_test.cpp index 0b8ffc5..0b1397f 100644 --- a/cth_coro/tests/src/tasks/sync_task_test.cpp +++ b/cth_coro/tests/src/tasks/sync_task_test.cpp @@ -2,6 +2,8 @@ #include "test.hpp" +#include + namespace cth::co { CORO_TEST(sync_task, blocks_until_finished) { atomic_fence coroStarted{}; diff --git a/cth_coro/tests/src/this_coro/payload_test.cpp b/cth_coro/tests/src/this_coro/payload_test.cpp index 94bf4d9..ca2e9fa 100644 --- a/cth_coro/tests/src/this_coro/payload_test.cpp +++ b/cth_coro/tests/src/this_coro/payload_test.cpp @@ -6,16 +6,21 @@ #include "cth/coro/tasks/executor_task.hpp" #include "cth/coro/this_coro.hpp" +#include + +#include #include +#include +#include +#include namespace cth::co { namespace { - // a custom payload: the scheduler context (so the built-in tags keep working) plus a marker value - struct tagged_payload : this_coro::scheduler_payload_base { - constexpr tagged_payload(co::scheduler const& s, int marker) - : this_coro::scheduler_payload_base{s}, _marker{marker} {} + // a custom payload: a marker value, the built-in tags take the context from the promise + struct tagged_payload : this_coro::payload_base { + constexpr explicit tagged_payload(int marker) : _marker{marker} {} [[nodiscard]] constexpr int marker() const { return _marker; } @@ -30,14 +35,25 @@ namespace { // a custom this_coro tag that reads the marker off whatever payload the coroutine carries struct marker_tag : this_coro::tag_base { - static constexpr auto operator()(marked_payload auto const& p) { return data_awaiter{p.marker()}; } + static constexpr auto operator()(co::context const&, marked_payload auto const& p) { + return data_awaiter{p.marker()}; + } }; inline constexpr marker_tag marker{}; - // invariant guard for the exact regression that caused the empty-payload bug: the awaiter must - // expose payload_type, else this_coro tasks silently demote to the foreign/steal path and their - // payload optional is never injected. a compile error here beats a runtime deref of an empty optional. + struct parent_payload : this_coro::payload_base {}; + + // the conversion from the payload of the parent throws + struct throwing_payload : this_coro::payload_base { + throwing_payload(parent_payload const&) { throw std::runtime_error("payload conversion"); } + }; + + static_assert(!mta::nothrow_constructible); + + // invariant guard: the awaiter of a this_coro task must derive this_coro_awaiter_base, else the caller + // captures the task as a foreign awaitable and the task never gets the context and the payload. + // a compile error here beats a runtime deref of an empty payload. static_assert(this_coro_awaitable>, "executor_task lost its this_coro classification"); static_assert(this_coro_awaitable>); @@ -49,7 +65,7 @@ THIS_CORO_TEST(custom_payload, marker_injected) { auto task = []() -> executor_task { co_return co_await marker; }; - auto const actual = sync(exec.spawn(task(), 42)); + auto const actual = exec.sync_wait(task(), 42); EXPECT_EQ(actual, 42); } @@ -60,7 +76,7 @@ THIS_CORO_TEST(custom_payload, propagates_through_nested) { auto child = []() -> executor_task { co_return co_await marker; }; auto root = [](auto c) -> executor_task { co_return co_await c(); }; - auto const actual = sync(exec.spawn(root(child), 7)); + auto const actual = exec.sync_wait(root(child), 7); EXPECT_EQ(actual, 7); } @@ -76,7 +92,7 @@ THIS_CORO_TEST(custom_payload, deep_recursion_propagation) { } }; - auto const actual = sync(exec.spawn(Recursive{}(8), 99)); + auto const actual = exec.sync_wait(Recursive{}(8), 99); EXPECT_EQ(actual, 99); } @@ -84,10 +100,10 @@ THIS_CORO_TEST(custom_payload, builtin_executor_tag_coexists) { scheduler sched{autostart, 1}; executor exec{sched}; - // tagged_payload derives scheduler_payload_base, so the built-in tags still resolve against it + // the built-in tags take the context from the promise, they work with every payload auto task = []() -> executor_task { co_return co_await this_coro::executor; }; - auto const actual = sync(exec.spawn(task(), 1)); + auto const actual = exec.sync_wait(task(), 1); EXPECT_EQ(actual, exec); } @@ -97,11 +113,71 @@ THIS_CORO_TEST(custom_payload, custom_and_builtin_tags_in_one_task) { auto task = []() -> executor_task { (void) co_await this_coro::executor; // built-in tag resolves under the custom payload - co_return co_await marker; // custom tag reads the injected marker + co_return co_await marker; // custom tag reads the marker of the payload }; - auto const actual = sync(exec.spawn(task(), 5)); + auto const actual = exec.sync_wait(task(), 5); EXPECT_EQ(actual, 5); } +// the custom payload slices to the default payload of the child +THIS_CORO_TEST(custom_payload, awaits_default_payload_task) { + scheduler sched{autostart, 1}; + executor exec{sched}; + + auto child = []() -> executor_task { co_return co_await this_coro::executor; }; + auto root = [](auto c) -> executor_task, tagged_payload> { + auto const e = co_await c(); + co_return std::pair{e, co_await marker}; + }; + + auto const [actual, m] = exec.sync_wait(root(child), 4); + EXPECT_EQ(actual, exec); + EXPECT_EQ(m, 4); +} + +// a throwing payload copy reaches the caller, the child does not start +THIS_CORO_TEST(custom_payload, throwing_copy_reaches_caller) { + scheduler sched{autostart, 1}; + executor exec{sched}; + std::atomic childStarted{false}; + + auto child = [](std::atomic& started) -> executor_task { + started = true; + co_return; + }; + auto root = [](scheduler const& s, std::atomic& started, auto c) + -> executor_task { + bool caughtOnSched = false; + try { + co_await c(started); + } catch(std::runtime_error const&) { + caughtOnSched = s.owns_thread(); + } + co_return caughtOnSched; + }; + + EXPECT_TRUE(exec.sync_wait(root(sched, childStarted, child))); + EXPECT_FALSE(childStarted); +} + +// schedule() changes the scheduler, the payload stays +THIS_CORO_TEST(custom_payload, schedule_keeps_payload) { + scheduler home{autostart, 1}; + scheduler other{autostart, 1}; + executor homeExec{home}; + + auto task = [](executor target) -> executor_task, tagged_payload> { + co_await target.schedule(); + int const m = co_await marker; + context const& current = co_await this_coro::context; + co_return std::tuple{m, ¤t, target.context().owns_thread()}; + }; + + auto const [m, current, onOther] = homeExec.sync_wait(task(executor{other}), 3); + EXPECT_EQ(m, 3); + EXPECT_EQ(current, &other); + EXPECT_TRUE(onOther); +} + } diff --git a/cth_coro/tests/src/this_coro/tags/scheduler_tag_test.cpp b/cth_coro/tests/src/this_coro/tags/context_tag_test.cpp similarity index 61% rename from cth_coro/tests/src/this_coro/tags/scheduler_tag_test.cpp rename to cth_coro/tests/src/this_coro/tags/context_tag_test.cpp index c3b8d3b..e77e1c4 100644 --- a/cth_coro/tests/src/this_coro/tags/scheduler_tag_test.cpp +++ b/cth_coro/tests/src/this_coro/tags/context_tag_test.cpp @@ -8,18 +8,18 @@ namespace cth::co::this_coro { // AI -TAG_TEST(scheduler_tag, returns_correct_scheduler) { +TAG_TEST(context_tag, returns_correct_context) { co::scheduler sched{autostart, 1}; co::executor exec{sched}; - using r_t = co::scheduler const&; + using r_t = co::context const&; - auto task = []() -> executor_task { co_return co_await this_coro::scheduler; }; + auto task = []() -> executor_task { co_return co_await this_coro::context; }; - static_assert(std::same_as); + static_assert(std::same_as); - auto& actual = sync_wait(exec, task()); + auto& actual = exec.sync_wait(task()); EXPECT_EQ(&actual, &sched); } diff --git a/cth_coro/tests/src/this_coro/tags/executor_tag_test.cpp b/cth_coro/tests/src/this_coro/tags/executor_tag_test.cpp index b1acd80..d04bbe0 100644 --- a/cth_coro/tests/src/this_coro/tags/executor_tag_test.cpp +++ b/cth_coro/tests/src/this_coro/tags/executor_tag_test.cpp @@ -14,7 +14,7 @@ TAG_TEST(executor_tag, returns_correct_executor) { auto task = []() -> executor_task { co_return co_await this_coro::executor; }; - auto const actual = sync_wait(exec, task()); + auto const actual = exec.sync_wait(task()); EXPECT_EQ(actual, exec); } diff --git a/cth_coro/tests/src/this_coro/tags/native_handle_tag_test.cpp b/cth_coro/tests/src/this_coro/tags/native_handle_tag_test.cpp index fc5b7e0..c534ed7 100644 --- a/cth_coro/tests/src/this_coro/tags/native_handle_tag_test.cpp +++ b/cth_coro/tests/src/this_coro/tags/native_handle_tag_test.cpp @@ -62,7 +62,7 @@ TAG_TEST(native_handle_tag, ready_when_already_signaled) { std::atomic const unused{true}; auto const start = chrono::clock_t::now(); - EXPECT_TRUE(sync_wait(exec, await_tag_task(event.handle, unused))); + EXPECT_TRUE(exec.sync_wait(await_tag_task(event.handle, unused))); auto const end = chrono::clock_t::now(); EXPECT_LT(end, start + std::chrono::milliseconds{100}); @@ -86,7 +86,7 @@ TAG_TEST(native_handle_tag, suspends_until_signaled) { } }; - EXPECT_TRUE(sync_wait(exec, await_tag_task(event.handle, signaled))); + EXPECT_TRUE(exec.sync_wait(await_tag_task(event.handle, signaled))); } // same contract through the free function spelling, which is how callers actually use it @@ -105,7 +105,7 @@ TAG_TEST(await_handle, resumes_after_signal) { } }; - EXPECT_TRUE(sync_wait(exec, await_handle_task(event.handle, signaled))); + EXPECT_TRUE(exec.sync_wait(await_handle_task(event.handle, signaled))); } // the executor contract must survive the suspension: resumption is driven by the scheduler's completion @@ -132,7 +132,7 @@ TAG_TEST(await_handle, resumes_on_scheduler_thread) { } }; - auto const resumed = sync_wait(exec, task(event.handle)); + auto const resumed = exec.sync_wait(task(event.handle)); EXPECT_NE(resumed, caller); EXPECT_NE(resumed, signaller.load()); diff --git a/cth_coro/tests/src/this_coro/tags/wait_tag_test.cpp b/cth_coro/tests/src/this_coro/tags/wait_tag_test.cpp index e273035..3208a3a 100644 --- a/cth_coro/tests/src/this_coro/tags/wait_tag_test.cpp +++ b/cth_coro/tests/src/this_coro/tags/wait_tag_test.cpp @@ -20,14 +20,14 @@ TAG_TEST(wait_tag, sub_1ms) { auto start = chrono::clock_t::now(); - sync_wait(exec, task()); + exec.sync_wait(task()); auto end = chrono::clock_t::now(); EXPECT_LT(end, start + std::chrono::milliseconds{3}); /* for(size_t i = 0; i < 10; i++) { auto start2 = clock_t::now(); - sync_wait(exec, task()); + exec.sync_wait(task()); auto end2 = clock_t::now(); std::println("waited for: {}", std::chrono::duration_cast(end2 - start2)); @@ -40,7 +40,7 @@ TAG_TEST(wait, main) { co_return; }; - sync_wait(exec, task()); + exec.sync_wait(task()); } TAG_TEST(wait_until, main) { @@ -49,7 +49,7 @@ TAG_TEST(wait_until, main) { co_return; }; - sync_wait(exec, task()); + exec.sync_wait(task()); } diff --git a/cth_coro/tests/src/this_coro/test.hpp b/cth_coro/tests/src/this_coro/test.hpp index c72dc79..b0a3133 100644 --- a/cth_coro/tests/src/this_coro/test.hpp +++ b/cth_coro/tests/src/this_coro/test.hpp @@ -2,19 +2,7 @@ #include "../test.hpp" #include "cth/coro/executor.hpp" -#include "cth/coro/sync.hpp" -#include "cth/coro/tasks/sync_task.hpp" #define THIS_CORO_EX_TEST(ex, suite, name) CORO_EX_TEST(_this_coro##ex, suite, name) #define THIS_CORO_TEST(suite, name) THIS_CORO_EX_TEST(, suite, name) - -namespace cth::co { - - -template -awaited_t sync_wait(executor& exec, Task task) { - return co::sync(exec.spawn(std::move(task))); -} - -} diff --git a/cth_coro/tests/src/this_coro/this_coro_promise_base_test.cpp b/cth_coro/tests/src/this_coro/this_coro_promise_base_test.cpp index 15f9156..079b775 100644 --- a/cth_coro/tests/src/this_coro/this_coro_promise_base_test.cpp +++ b/cth_coro/tests/src/this_coro/this_coro_promise_base_test.cpp @@ -13,7 +13,7 @@ THIS_CORO_TEST(promise_base, tag_injection) { auto task = []() -> executor_task { co_return co_await this_coro::executor; }; - auto const actual = sync_wait(exec, task()); + auto const actual = exec.sync_wait(task()); EXPECT_EQ(actual, exec); } @@ -25,7 +25,7 @@ THIS_CORO_TEST(promise_base, nested_task_propagation) { auto root = [](auto c) -> executor_task { co_return co_await c(); }; - auto const actual = sync_wait(exec, root(child)); + auto const actual = exec.sync_wait(root(child)); EXPECT_EQ(actual, exec); } @@ -42,7 +42,7 @@ THIS_CORO_TEST(promise_base, deep_recursion_propagation) { } }; - auto const actual = sync_wait(exec, Recursive{}(10)); + auto const actual = exec.sync_wait(Recursive{}(10)); EXPECT_EQ(actual, exec); } diff --git a/cth_coro/tests/src/utility/concepts_test.cpp b/cth_coro/tests/src/utility/concepts_test.cpp index 6cb229e..0fa0595 100644 --- a/cth_coro/tests/src/utility/concepts_test.cpp +++ b/cth_coro/tests/src/utility/concepts_test.cpp @@ -1,33 +1,56 @@ #include "cth/coro/utility/concepts.hpp" +#include "cth/coro/awaiters/schedule_awaiter.hpp" #include "cth/coro/tasks/dev/capture_task.hpp" #include "cth/coro/tasks/dev/spawn_task.hpp" #include "cth/coro/tasks/executor_task.hpp" #include "cth/coro/tasks/sync_task.hpp" +#include "cth/coro/this_coro.hpp" namespace cth::co { -// the result of steal() returns to the payload context of the steal itself -static_assert(captured_awaitable>); -static_assert(captured_awaitable>); +// the capture of a foreign awaitable continues the caller on its context +static_assert(this_coro_awaitable>); +static_assert(this_coro_awaitable>); static_assert(!foreign_awaitable>); -static_assert(!this_coro_awaitable>); -// the result of spawn() takes the scheduler of the caller from its payload +// the result of spawn() continues the caller on its context static_assert(this_coro_awaitable>); static_assert(this_coro_awaitable>); -static_assert(!captured_awaitable>); static_assert(!foreign_awaitable>); static_assert(this_coro_awaitable>); -static_assert(!captured_awaitable>); +static_assert(this_coro_awaitable&>); static_assert(!foreign_awaitable>); -static_assert(this_coro_awaitable>); -static_assert(!captured_awaitable>); -static_assert(!foreign_awaitable>); +static_assert(this_coro_awaitable); +static_assert(this_coro_awaitable); +static_assert(!foreign_awaitable); static_assert(foreign_awaitable>); -static_assert(!captured_awaitable>); static_assert(!this_coro_awaitable>); + +namespace { + struct caller_promise {}; + + struct nothrow_attach_promise { + template + void attach(Caller const&) noexcept {} + }; + + struct throwing_attach_promise { + template + void attach(Caller const&) {} + }; +} + +// a promise attaches to its caller, nothrow if the attach is noexcept +static_assert(dev::nothrow_attachable); +static_assert(!dev::nothrow_attachable); +static_assert(!dev::nothrow_attachable); + +// tags are neither +static_assert(!this_coro_awaitable); +static_assert(!foreign_awaitable); +static_assert(!foreign_awaitable); } diff --git a/cth_net/incl/cth/net/coro/websocket.hpp b/cth_net/incl/cth/net/coro/websocket.hpp index 3a34330..f287d37 100644 --- a/cth_net/incl/cth/net/coro/websocket.hpp +++ b/cth_net/incl/cth/net/coro/websocket.hpp @@ -19,7 +19,7 @@ class websocket_connection; -class accept_awaiter : public cth::co::osfence_awaiter_base<> { +class accept_awaiter : public cth::co::osfence_awaiter_base { public: explicit accept_awaiter(net::websocket_server& server) noexcept : osfence_awaiter_base{server.fence()}, @@ -32,7 +32,7 @@ class accept_awaiter : public cth::co::osfence_awaiter_base<> { }; -class message_awaiter_base : public cth::co::osfence_awaiter_base<> { +class message_awaiter_base : public cth::co::osfence_awaiter_base { public: explicit message_awaiter_base(net::websocket_connection& connection) noexcept : osfence_awaiter_base{connection.fence()}, @@ -46,7 +46,10 @@ class message_awaiter : public message_awaiter_base { public: using message_awaiter_base::message_awaiter_base; - [[nodiscard]] bool await_resume() { return _connection.wait(); } + [[nodiscard]] bool await_resume() { + osfence_awaiter_base::await_resume(); + return _connection.wait(); + } }; class text_awaiter : public message_awaiter_base { @@ -54,6 +57,7 @@ class text_awaiter : public message_awaiter_base { using message_awaiter_base::message_awaiter_base; [[nodiscard]] std::string await_resume() { + osfence_awaiter_base::await_resume(); auto const connected = _connection.wait(); CTH_NET_STABLE_THROW(!connected, "WebSocket closed while awaiting text") {} auto const text = _connection.got_text(); @@ -67,6 +71,7 @@ class bytes_awaiter : public message_awaiter_base { using message_awaiter_base::message_awaiter_base; [[nodiscard]] std::vector await_resume() { + osfence_awaiter_base::await_resume(); auto const connected = _connection.wait(); CTH_NET_STABLE_THROW(!connected, "WebSocket closed while awaiting bytes") {} auto const text = _connection.got_text(); @@ -118,6 +123,7 @@ class websocket_connection { }; inline std::optional accept_awaiter::await_resume() const { + osfence_awaiter_base::await_resume(); auto connection = _server.accept(); if(!connection) return std::nullopt; diff --git a/cth_net/tests/src/websocket_test.cpp b/cth_net/tests/src/websocket_test.cpp index e24e2e7..4c622ea 100644 --- a/cth_net/tests/src/websocket_test.cpp +++ b/cth_net/tests/src/websocket_test.cpp @@ -4,6 +4,7 @@ #include "cth/net/websocket_server.hpp" #include +#include #include #include diff --git a/cth_win/incl/cth/win/coro/keyboard.hpp b/cth_win/incl/cth/win/coro/keyboard.hpp index d9cbf9d..d7b1f19 100644 --- a/cth_win/incl/cth/win/coro/keyboard.hpp +++ b/cth_win/incl/cth/win/coro/keyboard.hpp @@ -11,7 +11,7 @@ class keybd_event_queue; * suspends until the queue holds an event, resumes with it * @attention one awaiter per queue at a time */ -class keybd_event_awaiter : public cth::co::osfence_awaiter_base<> { +class keybd_event_awaiter : public cth::co::osfence_awaiter_base { public: using queue_t = win::keybd_event_queue; using event_t = ::cth::io::key_update; diff --git a/cth_win/src/coro/keyboard.cpp b/cth_win/src/coro/keyboard.cpp index fec5702..1c6c9d6 100644 --- a/cth_win/src/coro/keyboard.cpp +++ b/cth_win/src/coro/keyboard.cpp @@ -8,6 +8,7 @@ keybd_event_awaiter::keybd_event_awaiter(queue_t& queue) noexcept : osfence_awaiter_base{queue.fence()}, _queue{queue} {} auto keybd_event_awaiter::await_resume() -> event_t { + osfence_awaiter_base::await_resume(); CTH_CRITICAL(_queue.empty(), "keyboard queue resumed without an event") {} return _queue.pop(); } diff --git a/cth_win/tests/src/coro/keyboard_test.cpp b/cth_win/tests/src/coro/keyboard_test.cpp index b557513..b1dd124 100644 --- a/cth_win/tests/src/coro/keyboard_test.cpp +++ b/cth_win/tests/src/coro/keyboard_test.cpp @@ -3,6 +3,7 @@ #include "cth/win/coro/keyboard.hpp" #include +#include #include #include #include