From e909aadc6cb39124e5868448d932a8b3c46731d6 Mon Sep 17 00:00:00 2001 From: lukas <89145039+codethinki@users.noreply.github.com> Date: Sat, 24 Oct 2026 16:11:54 +0200 Subject: [PATCH] feat(coro): context interface, strand, exact current context - context: the interface of scheduler and strand (post, await, owns_thread), executor, payloads and awaiters take context const& - strand: serial queue on the workers of a scheduler, jobs run one at a time in post order - context::current(): the context of the job on this thread, schedule_awaiter and spawn decide with it - scheduler -> context renames in the payload, the executor, the awaiters and the this_coro tag - asio helpers once in src/utility/asio/ops.hpp, the macro in asio/macro.hpp --- README.md | 4 +- .../base/native_handle_awaiter_base.hpp | 6 +- .../awaiters/base/osfence_awaiter_base.hpp | 2 +- .../awaiters/raw_native_handle_awaiter.hpp | 6 +- .../cth/coro/awaiters/schedule_awaiter.hpp | 39 ++- .../incl/cth/coro/awaiters/wait_awaiter.hpp | 6 +- cth_coro/incl/cth/coro/context.hpp | 84 ++++++ cth_coro/incl/cth/coro/executor.hpp | 54 ++-- cth_coro/incl/cth/coro/func/steal.hpp | 18 +- cth_coro/incl/cth/coro/scheduler.hpp | 19 +- cth_coro/incl/cth/coro/strand.hpp | 44 +++ .../awaiters/dev/spawn_final_awaiter.hpp | 10 +- .../awaiters/dev/spawn_promise_awaiter.hpp | 10 +- .../incl/cth/coro/tasks/dev/spawn_task.hpp | 6 +- .../tasks/promises/this_coro_promise_base.hpp | 28 +- cth_coro/incl/cth/coro/this_coro.hpp | 2 +- .../incl/cth/coro/this_coro/context_tag.hpp | 15 + .../cth/coro/this_coro/native_handle_tag.hpp | 4 +- cth_coro/incl/cth/coro/this_coro/payload.hpp | 20 +- .../incl/cth/coro/this_coro/scheduler_tag.hpp | 15 - cth_coro/incl/cth/coro/this_coro/wait_tag.hpp | 2 +- cth_coro/incl/cth/coro/utility/concepts.hpp | 4 +- cth_coro/incl/cth/coro/utility/fwd.hpp | 12 +- cth_coro/src/scheduler.cpp | 71 +---- cth_coro/src/scheduler_impl.hpp | 39 +++ cth_coro/src/strand.cpp | 42 +++ .../src/utility/{asio.hpp => asio/macro.hpp} | 0 cth_coro/src/utility/asio/ops.hpp | 70 +++++ cth_coro/src/utility/timer_pool.hpp | 2 +- cth_coro/tests/src/awaiters/awaiters_test.cpp | 1 + .../base/native_handle_awaiter_base_test.cpp | 1 + .../base/osfence_awaiter_base_test.cpp | 1 + .../raw_native_handle_awaiter_test.cpp | 1 + cth_coro/tests/src/executor_test.cpp | 16 +- cth_coro/tests/src/strand_test.cpp | 275 ++++++++++++++++++ cth_coro/tests/src/tasks/sync_task_test.cpp | 2 + cth_coro/tests/src/this_coro/payload_test.cpp | 10 +- ...uler_tag_test.cpp => context_tag_test.cpp} | 6 +- cth_net/tests/src/websocket_test.cpp | 1 + cth_win/tests/src/coro/keyboard_test.cpp | 1 + 40 files changed, 744 insertions(+), 205 deletions(-) create mode 100644 cth_coro/incl/cth/coro/context.hpp create mode 100644 cth_coro/incl/cth/coro/strand.hpp create mode 100644 cth_coro/incl/cth/coro/this_coro/context_tag.hpp delete mode 100644 cth_coro/incl/cth/coro/this_coro/scheduler_tag.hpp create mode 100644 cth_coro/src/scheduler_impl.hpp create mode 100644 cth_coro/src/strand.cpp rename cth_coro/src/utility/{asio.hpp => asio/macro.hpp} (100%) create mode 100644 cth_coro/src/utility/asio/ops.hpp create mode 100644 cth_coro/tests/src/strand_test.cpp rename cth_coro/tests/src/this_coro/tags/{scheduler_tag_test.cpp => context_tag_test.cpp} (81%) diff --git a/README.md b/README.md index 73fb0fd..777ebc6 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, moves execution back onto the context after foreign awaitables - `this_coro::executor`, `this_coro::wait(duration)`, native handle waits ## cth::os 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..0eddd48 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,18 @@ #pragma once -#include "cth/coro/scheduler.hpp" +#include "cth/coro/context.hpp" #include "cth/coro/tasks/awaiters/dev/this_coro_awaiter_base.hpp" #include namespace cth::co { -template +template struct native_handle_awaiter_base : dev::this_coro_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(); }); + this->payload().context().await(_handle, [h] { h.resume(); }); } 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..a322ff9 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,7 +4,7 @@ #include "cth/os/fence.hpp" namespace cth::co { -template +template 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/raw_native_handle_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/raw_native_handle_awaiter.hpp index 248ba34..3594980 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 @@ -9,12 +9,12 @@ namespace cth::co { struct raw_native_handle_awaiter { - co::scheduler const& scheduler; + 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(); }); + context.await(handle, [h] { h.resume(); }); } static constexpr void await_resume() {} }; diff --git a/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp index b0f09a1..ba3eb76 100644 --- a/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp +++ b/cth_coro/incl/cth/coro/awaiters/schedule_awaiter.hpp @@ -1,6 +1,7 @@ #pragma once -#include "cth/coro/scheduler.hpp" +#include "cth/coro/context.hpp" #include "cth/coro/this_coro/this_coro_base.hpp" +#include "cth/coro/utility/concepts.hpp" #include "cth/coro/utility/fwd.hpp" #include @@ -8,26 +9,38 @@ namespace cth::co { /** - * switches execution onto a target @ref scheduler and stays there - * @details + * switches execution onto a target @ref context and stays there + * @details noop if the coroutine already runs on the target: @ref context::current() is the target * @tparam Pyld payload of the migrating coroutine, defaulted to @ref this_coro::default_payload */ -template +template struct schedule_awaiter : this_coro_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) { + 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 the payload to the target and posts the coroutine onto the target + * @details owns_thread() is not enough: a scheduler also owns the threads that run its strands + * @return false if the coroutine already runs on the target + */ + template requires this_coro::context_payload + bool 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); + if constexpr(has_context) + h.promise().rebind_context(_target); + + if(context::current() == &_target) + return false; _target.post([h]() { h.resume(); }); + return true; } template P> @@ -36,6 +49,6 @@ struct schedule_awaiter : this_coro_base { constexpr void await_resume() const noexcept {} private: - co::scheduler const& _target; + co::context const& _target; }; } diff --git a/cth_coro/incl/cth/coro/awaiters/wait_awaiter.hpp b/cth_coro/incl/cth/coro/awaiters/wait_awaiter.hpp index 607cc2a..bff52c6 100644 --- a/cth_coro/incl/cth/coro/awaiters/wait_awaiter.hpp +++ b/cth_coro/incl/cth/coro/awaiters/wait_awaiter.hpp @@ -1,5 +1,5 @@ #pragma once -#include "cth/coro/scheduler.hpp" +#include "cth/coro/context.hpp" #include "cth/coro/this_coro/payload.hpp" #include "cth/coro/utility/fwd.hpp" @@ -11,14 +11,14 @@ namespace cth::co { struct wait_awaiter { std::chrono::steady_clock::time_point timePoint; - co::scheduler const& scheduler; + co::context const& context; [[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(); }); + context.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..7c78e63 --- /dev/null +++ b/cth_coro/incl/cth/coro/context.hpp @@ -0,0 +1,84 @@ +#pragma once +#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 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 + * @param cb to call + */ + virtual void await(native_handle, void_func cb) const = 0; + /** + * calls the callback on this context once the time point is reached + * @param time_point to wait for + * @param cb to call + */ + virtual void await(chrono::time_point_t time_point, void_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; + + /** + * @return context of the job that runs on the calling thread, nullptr -> no job of a context + * @details a strand job -> the strand, an other job of a scheduler -> the scheduler + */ + [[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 current() is this context while the work runs + * @details restores the previous context after the work, also if the work throws + */ + [[nodiscard]] auto scoped(void_func work) const { + return [self = this, work = std::move(work)]() mutable { + current_guard const guard{self}; + work(); + }; + } + + /** + * context of the job that runs on this thread, see @ref 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/executor.hpp b/cth_coro/incl/cth/coro/executor.hpp index 85c7f7f..9be040d 100644 --- a/cth_coro/incl/cth/coro/executor.hpp +++ b/cth_coro/incl/cth/coro/executor.hpp @@ -11,68 +11,68 @@ 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 + * calling co_await on this switches the execution onto this context 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()}; } + [[nodiscard]] constexpr auto schedule() const { return schedule_awaiter{context()}; } /** - * 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 + * steals the execution to the context after the coroutine finished, keeps it as awaitable + * @tparam Pyld payload type; the context is injected automatically when it is a context_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) + * @param args extra arguments to construct the payload with (besides the auto-injected context) * @return awaitable with context stolen * @attention the awaiter does not own the task, co_await it in the same expression, do not store it */ 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)...}); + if constexpr(this_coro::context_payload) + return co::steal(std::forward(awaitable), Pyld{context(), std::forward(args)...}); else return co::steal(std::forward(awaitable), Pyld{std::forward(args)...}); } /** - * steals the execution context to the scheduler after the coroutine finished + * steals the execution to the context 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 Pyld payload type; the context is injected automatically when it is a context_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) + * @param args extra arguments to construct the payload with (besides the auto-injected context) * @return task with context stolen */ 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)...}); + if constexpr(this_coro::context_payload) + return co::steal(std::move(awaitable), Pyld{context(), std::forward(args)...}); else return co::steal(std::move(awaitable), Pyld{std::forward(args)...}); } /** - * 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, + * runs the coroutine on the context once started + * @details the caller continues on its own context 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 + * @tparam Pyld payload type; the context is injected automatically when it is a context_payload + * @param task coroutine to run on the context + * @param args extra arguments to construct the payload with (besides the auto-injected context) + * @return cold spawn task; starts on the context when awaited */ 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)); + if constexpr(this_coro::context_payload) + return spawn_impl(Pyld{context(), std::forward(args)...}, *this, std::move(task)); else return spawn_impl(Pyld{std::forward(args)...}, *this, std::move(task)); } @@ -87,20 +87,20 @@ class executor { -> spawn_task> { co_await self.schedule(); - // the final awaiter of the spawn_task resumes the caller on its own scheduler + // the final awaiter of the spawn_task resumes the caller on its own context 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::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; @@ -117,7 +117,7 @@ class executor { 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()}}; + return data_awaiter{executor{p.context()}}; } }; diff --git a/cth_coro/incl/cth/coro/func/steal.hpp b/cth_coro/incl/cth/coro/func/steal.hpp index 24a96a5..86a0c46 100644 --- a/cth_coro/incl/cth/coro/func/steal.hpp +++ b/cth_coro/incl/cth/coro/func/steal.hpp @@ -11,19 +11,19 @@ namespace cth::co { -template +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"); + static_assert(has_context, "foreign awaitables can only be stolen with a context 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_await schedule_awaiter{payload.context()}; co_return; } else { decltype(auto) result = co_await std::move(awaitable); - co_await schedule_awaiter{payload.scheduler()}; + co_await schedule_awaiter{payload.context()}; co_return result; } } @@ -66,18 +66,18 @@ template template [[nodiscard]] auto steal( Awaitable awaitable, - scheduler const& scheduler + context const& context ) -> capture_task> { - return co::steal(std::move(awaitable), this_coro::default_payload{scheduler}); + return co::steal(std::move(awaitable), this_coro::default_payload{context}); } /** - * injects the default payload of the scheduler into the awaiter of the awaitable + * injects the default payload of the context 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}); +constexpr auto steal(Awaitable&& awaitable, context const& context) -> awaiter_t { + return co::steal(std::forward(awaitable), this_coro::default_payload{context}); } diff --git a/cth_coro/incl/cth/coro/scheduler.hpp b/cth_coro/incl/cth/coro/scheduler.hpp index e2f7245..7540c0b 100644 --- a/cth_coro/incl/cth/coro/scheduler.hpp +++ b/cth_coro/incl/cth/coro/scheduler.hpp @@ -3,6 +3,7 @@ #include #include +#include "context.hpp" #include "utility/fwd.hpp" #include @@ -19,15 +20,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 +51,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, void_func cb) const override; + void await(chrono::time_point_t time_point, void_func cb) const override; void start(); @@ -73,7 +76,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..9fa17cd --- /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, void_func cb) const override; + void await(chrono::time_point_t time_point, void_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/spawn_final_awaiter.hpp b/cth_coro/incl/cth/coro/tasks/awaiters/dev/spawn_final_awaiter.hpp index 9e91889..6887ac7 100644 --- 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 @@ -1,14 +1,14 @@ #pragma once -#include "cth/coro/scheduler.hpp" +#include "cth/coro/context.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 + * final awaiter of @ref spawn_promise, resumes the awaiting coroutine on its own context + * @details transfers directly if @ref context::current() is the context of the caller, else posts the resume onto it, + * a caller without a known context continues where the spawned coroutine ended * @attention terminates if the post throws, the final awaiter is noexcept */ struct spawn_final_awaiter { @@ -21,7 +21,7 @@ struct spawn_final_awaiter { return std::noop_coroutine(); auto const* caller = h.promise().caller; - if(caller == nullptr || caller->owns_thread()) + if(caller == nullptr || context::current() == caller) return continuation; // the caller may destroy this coroutine as soon as it resumes, nothing touches the frame after the post 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 index a35f816..460fc75 100644 --- 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 @@ -12,7 +12,7 @@ namespace cth::co::dev { /** - * awaiter of @ref spawn_task, takes the scheduler of the caller from the injected payload + * awaiter of @ref spawn_task, takes the context 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 @@ -30,11 +30,11 @@ struct spawn_promise_awaiter : this_coro_base, priva using base_t::base_t; /** - * stores the scheduler of the caller + * stores the context of the caller * @param payload of the caller */ - template - constexpr void inject(Pyld const& payload) noexcept { _caller = &payload.scheduler(); } + template + constexpr void inject(Pyld const& payload) noexcept { _caller = &payload.context(); } [[nodiscard]] auto await_suspend(std::coroutine_handle<> caller) noexcept -> std::coroutine_handle { base_t::handle.promise().caller = _caller; @@ -42,6 +42,6 @@ struct spawn_promise_awaiter : this_coro_base, priva } private: - co::scheduler const* _caller = nullptr; + co::context const* _caller = nullptr; }; } 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..ba9abab 100644 --- a/cth_coro/incl/cth/coro/tasks/dev/spawn_task.hpp +++ b/cth_coro/incl/cth/coro/tasks/dev/spawn_task.hpp @@ -14,9 +14,9 @@ namespace cth::co::dev { template struct spawn_promise : basic_promise { /** - * scheduler of the awaiting coroutine, nullptr -> unknown + * context of the awaiting coroutine, nullptr -> unknown */ - co::scheduler const* caller = nullptr; + co::context const* caller = nullptr; spawn_task get_return_object() noexcept; }; @@ -27,7 +27,7 @@ namespace cth::co { /** * task of @ref executor::spawn() - * @details this_coro awaitable: the caller continues on its own scheduler after the spawned coroutine ended, + * @details this_coro awaitable: the caller continues on its own context after the spawned coroutine ended, * a caller without this_coro support continues where the spawned coroutine ended * @tparam T result type */ 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..2d9a61b 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 @@ -56,13 +56,13 @@ struct this_coro_promise_base : this_coro_base { template [[nodiscard]] constexpr auto await_transform(Awaitable&& captured) { return cth::co::extract_awaiter(captured); } - template requires (this_coro::scheduler_payload) + template requires (this_coro::context_payload) [[nodiscard]] constexpr auto await_transform(Waitable const& waitable) { - return raw_native_handle_awaiter{_payload->scheduler(), waitable.native_handle()}; + return raw_native_handle_awaiter{_payload->context(), waitable.native_handle()}; } template - requires (this_coro::scheduler_payload && !native_handle_waitable) + requires (this_coro::context_payload && !native_handle_waitable) [[nodiscard]] auto await_transform(Awaitable awaitable) -> capture_task> { return co::steal(std::move(awaitable), *_payload); } @@ -74,15 +74,25 @@ struct this_coro_promise_base : this_coro_base { constexpr void inject(Pyld const& payload) { _payload = payload; } /** - * 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 + * gets the context of the payload + * @details only available for context payloads, the payload must be engaged + */ + [[nodiscard]] constexpr co::context const& context() const + requires this_coro::context_payload + { + return _payload->context(); + } + + /** + * repoints the promise's payload at a new context, preserving every other payload field + * @details only available for context 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 + * @param context the coroutine has migrated onto */ - constexpr void rebind_scheduler(co::scheduler const& scheduler) - requires this_coro::scheduler_payload + constexpr void rebind_context(co::context const& context) + requires this_coro::context_payload { - _payload->rebind_scheduler(scheduler); + _payload->rebind_context(context); } constexpr this_coro_promise_base(Pyld const& payload) noexcept : _payload{payload} {} 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..330057f --- /dev/null +++ b/cth_coro/incl/cth/coro/this_coro/context_tag.hpp @@ -0,0 +1,15 @@ +#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]] context_tag : tag_base { + static constexpr auto operator()(context_payload auto const& payload) { + return data_awaiter{payload.context()}; + } +}; + +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..8e03a33 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 @@ -9,8 +9,8 @@ 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()(context_payload auto const& p) const { + return raw_native_handle_awaiter{p.context(), 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..f5cc1b4 100644 --- a/cth_coro/incl/cth/coro/this_coro/payload.hpp +++ b/cth_coro/incl/cth/coro/this_coro/payload.hpp @@ -4,23 +4,23 @@ 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 + * class base for payloads carrying a context + * @details this coro machinery will auto-inject the context for types inheriting this base */ -struct scheduler_payload_base { - constexpr scheduler_payload_base(co::scheduler const& scheduler) : _scheduler{&scheduler} {} +struct context_payload_base { + constexpr context_payload_base(co::context const& context) : _context{&context} {} - constexpr co::scheduler const& scheduler() const { return *_scheduler; } + constexpr co::context const& context() const { return *_context; } /** - * 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 + * framework use: repoint the payload at the context the coroutine has migrated onto + * @details keeps every derived field intact, only the context is swapped + * @param context the coroutine is now running on */ - constexpr void rebind_scheduler(co::scheduler const& scheduler) noexcept { _scheduler = &scheduler; } + constexpr void rebind_context(co::context const& context) noexcept { _context = &context; } private: - co::scheduler const* _scheduler; + co::context const* _context; }; } 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/wait_tag.hpp b/cth_coro/incl/cth/coro/this_coro/wait_tag.hpp index 1f213f3..575a05e 100644 --- a/cth_coro/incl/cth/coro/this_coro/wait_tag.hpp +++ b/cth_coro/incl/cth/coro/this_coro/wait_tag.hpp @@ -9,7 +9,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()(context_payload auto const& p) const { return wait_awaiter{timePoint, p.context()}; } 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..3b9435a 100644 --- a/cth_coro/incl/cth/coro/utility/concepts.hpp +++ b/cth_coro/incl/cth/coro/utility/concepts.hpp @@ -13,8 +13,8 @@ namespace cth::co { template -concept has_scheduler = requires(T const& t) { - { t.scheduler() } -> std::convertible_to; +concept has_context = requires(T const& t) { + { t.context() } -> std::convertible_to; }; diff --git a/cth_coro/incl/cth/coro/utility/fwd.hpp b/cth_coro/incl/cth/coro/utility/fwd.hpp index cf4d4a7..fe977f4 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,7 +19,7 @@ template concept tag = std::is_base_of_v>; struct executor_tag; -struct scheduler_tag; +struct context_tag; struct wait_tag; @@ -26,11 +28,11 @@ struct wait_tag; namespace cth::co::this_coro { -struct scheduler_payload_base; +struct context_payload_base; template -concept scheduler_payload = std::is_base_of_v>; +concept context_payload = std::is_base_of_v>; -using default_payload = scheduler_payload_base; +using default_payload = context_payload_base; } @@ -114,7 +116,7 @@ class coro_exception; } namespace cth::co { -template +template struct schedule_awaiter; struct wait_awaiter; diff --git a/cth_coro/src/scheduler.cpp b/cth_coro/src/scheduler.cpp index 5e75c03..708729f 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,23 @@ 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, void_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, void_func cb) const { + dev::await_on(impl().ctx.get_executor(), impl().timerPool, time_point, std::move(cb)); } void scheduler::start() { 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..dd1e91c --- /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, void_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, void_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/macro.hpp similarity index 100% rename from cth_coro/src/utility/asio.hpp rename to cth_coro/src/utility/asio/macro.hpp diff --git a/cth_coro/src/utility/asio/ops.hpp b/cth_coro/src/utility/asio/ops.hpp new file mode 100644 index 0000000..21dba12 --- /dev/null +++ b/cth_coro/src/utility/asio/ops.hpp @@ -0,0 +1,70 @@ +#pragma once +#include "macro.hpp" + +#include "../native_handle_helpers.hpp" +#include "../timer_pool.hpp" + +#include "cth/coro/context.hpp" + +#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 + * @param executor to call the callback on + * @param ctx io_context of the executor, runs the wait + * @param cb to call + * @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), + cb = std::forward(cb) + ](bas::error_code const& ec) mutable { + ASIO_EC_STABLE_THROW(ec, "async wait for handle [{}] failed", h->native_handle()) + + cb(); + } + ) + ); +} + +/** + * calls the callback on the executor once the time point is reached + * @details the timer callback dispatches the callback onto the executor + * @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 + */ +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)]() mutable { bas::dispatch(executor, std::move(cb)); } + ); +} +} diff --git a/cth_coro/src/utility/timer_pool.hpp b/cth_coro/src/utility/timer_pool.hpp index c6136c8..8da197b 100644 --- a/cth_coro/src/utility/timer_pool.hpp +++ b/cth_coro/src/utility/timer_pool.hpp @@ -1,5 +1,5 @@ #pragma once -#include "asio.hpp" +#include "asio/macro.hpp" #include "native_handle_helpers.hpp" #include "cth/coro/utility/fwd.hpp" diff --git a/cth_coro/tests/src/awaiters/awaiters_test.cpp b/cth_coro/tests/src/awaiters/awaiters_test.cpp index 4d29144..214f63d 100644 --- a/cth_coro/tests/src/awaiters/awaiters_test.cpp +++ b/cth_coro/tests/src/awaiters/awaiters_test.cpp @@ -2,6 +2,7 @@ #include "../test.hpp" +#include "cth/coro/scheduler.hpp" #include "cth/meta/coro.hpp" #define AWAITER_TEST(suite, name) CORO_EX_TEST(_awaiters, suite, name) 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..64d010a 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 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..2488d4b 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 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..4931626 100644 --- a/cth_coro/tests/src/executor_test.cpp +++ b/cth_coro/tests/src/executor_test.cpp @@ -234,17 +234,17 @@ CORO_TEST(executor, context_reflects_schedule_hop) { 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) { +// Same defect observed through the context tag: after the hop this_coro::context must name the +// context the coroutine is now running on, not the one baked into the payload at 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; }; @@ -252,7 +252,7 @@ CORO_TEST(executor, context_scheduler_reflects_schedule_hop) { EXPECT_EQ(actual, &other); } -// spawn() runs the child on its scheduler, the caller continues on its own scheduler afterwards +// spawn() runs the child on its context, the caller continues on its own context afterwards CORO_TEST(executor, interop_spawn_caller_continues_on_own_scheduler) { scheduler home{autostart, 1}; scheduler other{autostart, 1}; @@ -261,9 +261,9 @@ 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}; }; diff --git a/cth_coro/tests/src/strand_test.cpp b/cth_coro/tests/src/strand_test.cpp new file mode 100644 index 0000000..5018675 --- /dev/null +++ b/cth_coro/tests/src/strand_test.cpp @@ -0,0 +1,275 @@ +#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}; + atomic_fence done{}; + + scheduler sched{autostart, 4}; + strand serial{sched}; + + serial.await( + fence.native_handle(), + [&] { + inStrand = serial.owns_thread(); + done.signal(); + } + ); + fence.signal(); + + done.wait(); + EXPECT_TRUE(inStrand); +} + +CORO_TEST(strand, await_time_point_calls_inside_strand) { + std::atomic inStrand{false}; + atomic_fence done{}; + + scheduler sched{autostart, 4}; + strand serial{sched}; + + serial.await( + chrono::clock_t::now() + std::chrono::milliseconds{1}, + [&] { + inStrand = serial.owns_thread(); + done.signal(); + } + ); + + done.wait(); + EXPECT_TRUE(inStrand); +} + +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] = sync(executor{sched}.spawn(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] = sync(executor{serial}.spawn(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] = sync(executor{serial}.spawn(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] = sync(executor{serial}.spawn(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] = sync(executor{sched}.spawn(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] = + sync(executor{serial}.spawn(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..c7a9e12 100644 --- a/cth_coro/tests/src/this_coro/payload_test.cpp +++ b/cth_coro/tests/src/this_coro/payload_test.cpp @@ -12,10 +12,10 @@ 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: the context (so the built-in tags keep working) plus a marker value + struct tagged_payload : this_coro::context_payload_base { + constexpr tagged_payload(co::context const& c, int marker) + : this_coro::context_payload_base{c}, _marker{marker} {} [[nodiscard]] constexpr int marker() const { return _marker; } @@ -84,7 +84,7 @@ 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 + // tagged_payload derives context_payload_base, so the built-in tags still resolve against it auto task = []() -> executor_task { co_return co_await this_coro::executor; }; auto const actual = sync(exec.spawn(task(), 1)); 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 81% 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..c5917d7 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,13 +8,13 @@ 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); 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/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