diff --git a/.github/workflows/ci-latest-kernel.yml b/.github/workflows/ci-latest-kernel.yml index 800e4442..c035c970 100644 --- a/.github/workflows/ci-latest-kernel.yml +++ b/.github/workflows/ci-latest-kernel.yml @@ -72,6 +72,7 @@ jobs: -DBUILD_EXAMPLES=ON \ -DBUILD_BENCHMARKS=ON \ -DBUILD_TESTS=ON \ + -DENABLE_STDEXEC=ON \ -DCMAKE_C_COMPILER=clang \ -DCMAKE_CXX_COMPILER=clang++ \ -DCMAKE_BUILD_TYPE=Release \ diff --git a/.github/workflows/ci-main.yml b/.github/workflows/ci-main.yml index 8b5e814a..72a3c148 100644 --- a/.github/workflows/ci-main.yml +++ b/.github/workflows/ci-main.yml @@ -65,6 +65,7 @@ jobs: -DBUILD_EXAMPLES=ON \ -DBUILD_BENCHMARKS=ON \ -DBUILD_TESTS=ON \ + -DENABLE_STDEXEC=ON \ $SANITIZER_FLAG \ -DCMAKE_C_COMPILER=${{matrix.compiler.cc}} \ -DCMAKE_CXX_COMPILER=${{matrix.compiler.cxx}} \ diff --git a/.github/workflows/ci-static-check.yml b/.github/workflows/ci-static-check.yml index 6c623207..99d5e3ca 100644 --- a/.github/workflows/ci-static-check.yml +++ b/.github/workflows/ci-static-check.yml @@ -45,6 +45,7 @@ jobs: -DBUILD_EXAMPLES=ON \ -DBUILD_BENCHMARKS=ON \ -DBUILD_TESTS=ON \ + -DENABLE_STDEXEC=ON \ -DCMAKE_C_COMPILER=clang \ -DCMAKE_CXX_COMPILER=clang++ \ -DCMAKE_BUILD_TYPE=Debug \ diff --git a/.github/workflows/ci-toolchain.yml b/.github/workflows/ci-toolchain.yml index 2e261af5..4b76e60e 100644 --- a/.github/workflows/ci-toolchain.yml +++ b/.github/workflows/ci-toolchain.yml @@ -46,6 +46,7 @@ jobs: -DBUILD_EXAMPLES=ON \ -DBUILD_BENCHMARKS=ON \ -DBUILD_TESTS=ON \ + -DENABLE_STDEXEC=ON \ -DCMAKE_C_COMPILER=${{ matrix.compiler.cc }} \ -DCMAKE_CXX_COMPILER=${{ matrix.compiler.cxx }} \ -DCMAKE_BUILD_TYPE=Debug @@ -61,8 +62,10 @@ jobs: - name: Build module-hello # liburing < 2.13 exposes TU-local entities that break GCC modules # https://github.com/axboe/liburing/issues/1457 + # stdexec uses TU-local entities incompatible with GCC C++20 modules if: > matrix.generator == 'Ninja' + && matrix.compiler.cc != 'gcc' && !(matrix.compiler.cc == 'gcc' && matrix.liburing-version == '2.3') run: | cmake -B ${{github.workspace}}/build_module -G Ninja \ @@ -75,5 +78,6 @@ jobs: - name: Run module-hello if: > matrix.generator == 'Ninja' + && matrix.compiler.cc != 'gcc' && !(matrix.compiler.cc == 'gcc' && matrix.liburing-version == '2.3') run: ${{github.workspace}}/build_module/examples/module-hello diff --git a/.gitmodules b/.gitmodules index 8acc6267..7b1c6310 100644 --- a/.gitmodules +++ b/.gitmodules @@ -7,3 +7,6 @@ [submodule "third_party/doxygen-awesome-css"] path = third_party/doxygen-awesome-css url = https://github.com/jothepro/doxygen-awesome-css.git +[submodule "third_party/stdexec"] + path = third_party/stdexec + url = https://github.com/NVIDIA/stdexec.git diff --git a/CMakeLists.txt b/CMakeLists.txt index 28929efa..8c10c777 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -11,6 +11,7 @@ option(TESTS_STATIC_LINK "Use static linking for tests" OFF) option(TESTS_TSAN "Enable thread sanitizer for tests" OFF) option(TESTS_ASAN "Enable address and undefined behavior sanitizers for tests" OFF) option(LINK_LIBURING "Use liburing from third_party and link it to condy" ON) +option(ENABLE_STDEXEC "Enable stdexec execution integration" ON) if (ENABLE_CLANG_TIDY) find_program(CLANG_TIDY_EXE NAMES "clang-tidy") @@ -55,6 +56,12 @@ if(BUILD_EXAMPLES OR BUILD_BENCHMARKS OR BUILD_TESTS OR LINK_LIBURING OR BUILD_M set_target_properties(uring PROPERTIES IMPORTED_LOCATION ${LIBURING_SOURCE_DIR}/src/liburing.a) target_include_directories(uring INTERFACE ${LIBURING_SOURCE_DIR}/src/include) add_dependencies(uring liburing_ext) + + if(ENABLE_STDEXEC) + # stdexec + add_library(stdexec INTERFACE) + target_include_directories(stdexec SYSTEM INTERFACE ${CMAKE_CURRENT_SOURCE_DIR}/third_party/stdexec/include) + endif() endif() if(LINK_LIBURING) @@ -67,6 +74,11 @@ if(BUILD_MODULE) add_subdirectory(module) endif() +if(ENABLE_STDEXEC) + target_link_libraries(condy INTERFACE stdexec) + target_compile_definitions(condy INTERFACE CONDY_HAS_STDEXEC) +endif() + # Warnings as errors add_compile_options(-Wall -Wextra -Werror) if(CMAKE_CXX_COMPILER_ID MATCHES "GNU") diff --git a/include/condy.hpp b/include/condy.hpp index ae49ce40..f2125fc9 100644 --- a/include/condy.hpp +++ b/include/condy.hpp @@ -15,6 +15,7 @@ #include "condy/channel.hpp" // IWYU pragma: export #include "condy/coro.hpp" // IWYU pragma: export #include "condy/cqe_handler.hpp" // IWYU pragma: export +#include "condy/execution.hpp" // IWYU pragma: export #include "condy/futex.hpp" // IWYU pragma: export #include "condy/helpers.hpp" // IWYU pragma: export #include "condy/pmr.hpp" // IWYU pragma: export diff --git a/include/condy/channel.hpp b/include/condy/channel.hpp index 140c8357..1a0663c1 100644 --- a/include/condy/channel.hpp +++ b/include/condy/channel.hpp @@ -20,6 +20,9 @@ #include #include #include +#ifdef CONDY_HAS_STDEXEC +#include "condy/detail/execution.hpp" +#endif namespace condy { @@ -52,6 +55,11 @@ template class Channel { CONDY_DELETE_COPY_MOVE(Channel); +private: + class [[nodiscard]] MovePushSenderImpl; + class [[nodiscard]] CopyPushSenderImpl; + class [[nodiscard]] PopSenderImpl; + public: /** * @brief Try to push an item into the channel. @@ -111,7 +119,11 @@ template class Channel { push_awaiters_.push_back(fake_handle); } - class [[nodiscard]] MovePushSender; +#ifdef CONDY_HAS_STDEXEC + using MovePushSender = detail::StandardSender; +#else + using MovePushSender = MovePushSenderImpl; +#endif /** * @brief Push an item into the channel, awaiting if necessary. * @param item The item to be pushed into the channel. @@ -126,7 +138,11 @@ template class Channel { */ MovePushSender push(T &&item) noexcept { return {*this, std::move(item)}; } - class [[nodiscard]] CopyPushSender; +#ifdef CONDY_HAS_STDEXEC + using CopyPushSender = detail::StandardSender; +#else + using CopyPushSender = CopyPushSenderImpl; +#endif /** * @brief Push an item into the channel, awaiting if necessary. * @param item The item to be pushed into the channel. @@ -140,7 +156,11 @@ template class Channel { return {*this, item}; } - class [[nodiscard]] PopSender; +#ifdef CONDY_HAS_STDEXEC + using PopSender = detail::StandardSender; +#else + using PopSender = PopSenderImpl; +#endif /** * @brief Pop an item from the channel, awaiting if necessary. * @return std::pair 0 and the popped item if successful; -EPIPE @@ -529,12 +549,12 @@ class Channel::PopFinishHandle std::optional stop_callback_; }; -template class Channel::MovePushSender { +template class Channel::MovePushSenderImpl { public: using CondySender = void; using ReturnType = int32_t; - MovePushSender(Channel &channel, T &&item) + MovePushSenderImpl(Channel &channel, T &&item) : channel_(channel), item_(std::move(item)) {} template auto connect_impl(Receiver receiver) noexcept { @@ -562,12 +582,12 @@ template class Channel::MovePushSender { T &&item_; }; -template class Channel::CopyPushSender { +template class Channel::CopyPushSenderImpl { public: using CondySender = void; using ReturnType = int32_t; - CopyPushSender(Channel &channel, const T &item) + CopyPushSenderImpl(Channel &channel, const T &item) : channel_(channel), item_(item) {} template auto connect_impl(Receiver receiver) noexcept { @@ -598,12 +618,12 @@ template class Channel::CopyPushSender { const T &item_; }; -template class Channel::PopSender { +template class Channel::PopSenderImpl { public: using CondySender = void; using ReturnType = std::pair; - PopSender(Channel &channel) : channel_(channel) {} + PopSenderImpl(Channel &channel) : channel_(channel) {} template auto connect_impl(Receiver receiver) noexcept { return OperationState(channel_, std::move(receiver)); diff --git a/include/condy/detail/execution.hpp b/include/condy/detail/execution.hpp new file mode 100644 index 00000000..27adf232 --- /dev/null +++ b/include/condy/detail/execution.hpp @@ -0,0 +1,293 @@ +/** + * @file execution.hpp + * @brief Internal sender/receiver adapters for std::execution integration. + */ + +#pragma once + +#ifdef CONDY_HAS_STDEXEC + +#include "condy/detail/utils.hpp" +#include "condy/runtime.hpp" +#include +#include + +namespace condy { + +namespace ex = stdexec; + +namespace detail { + +class Scheduler { +public: + using scheduler_concept = ex::scheduler_tag; + + Scheduler(Runtime &runtime) : runtime_(&runtime) {} + + bool operator==(const Scheduler &other) const noexcept { + return runtime_ == other.runtime_; + } + + auto schedule() const noexcept { return ScheduleSender{*runtime_}; } + +private: + template + class OperationState + : public InvokerAdapter, WorkInvoker> { + public: + using operation_state_concept = ex::operation_state_tag; + + OperationState(Runtime &runtime, Receiver receiver) + : runtime_(runtime), receiver_(std::move(receiver)) {} + + CONDY_DELETE_COPY_MOVE(OperationState); + + public: + void start() noexcept { runtime_.schedule_internal(this); } + + void invoke() noexcept { ex::set_value(std::move(receiver_)); } + + private: + Runtime &runtime_; + Receiver receiver_; + }; + + class ScheduleSender { + public: + using sender_concept = ex::sender_tag; + + using completion_signatures = + ex::completion_signatures; + + ScheduleSender(Runtime &runtime) : runtime_(runtime) {} + + template auto connect(Receiver receiver) { + return OperationState>(runtime_, + std::move(receiver)); + } + + struct Env { + Runtime &runtime; + Scheduler query(ex::get_completion_scheduler_t) + const noexcept { + return Scheduler{runtime}; + } + }; + Env get_env() const noexcept { return {runtime_}; } + + private: + Runtime &runtime_; + }; + +private: + Runtime *runtime_; +}; + +template struct set_value_traits { + using type = ex::set_value_t(R); +}; +template struct set_value_traits> { + using type = ex::set_value_t(R1, R2); +}; +template struct set_value_traits> { + using type = ex::set_value_t(R...); +}; +template +using set_value_traits_t = typename set_value_traits::type; + +template +class [[nodiscard]] StandardSender : public SenderImpl { +public: + using SenderImpl::SenderImpl; + + using sender_concept = ex::sender_tag; + using completion_signatures = ex::completion_signatures< + set_value_traits_t, + ex::set_error_t(std::error_code), ex::set_stopped_t()>; + + template auto connect(Receiver receiver) noexcept { + using OpState = decltype(this->connect_impl( + ReceiverWrapper{std::move(receiver)})); + return OperationStateWrapper{ + this->connect_impl(ReceiverWrapper{std::move(receiver)})}; + } + +private: + template struct ReceiverWrapper { + Receiver receiver; + + void operator()(int32_t res) noexcept { + if (res >= 0) { + ex::set_value(std::move(receiver), res); + } else if (res == -ECANCELED) { + ex::set_stopped(std::move(receiver)); + } else { + ex::set_error(std::move(receiver), + std::error_code(-res, std::generic_category())); + } + } + + template + void operator()(std::pair res) noexcept { + auto &[res_code, payload] = res; + if (res_code >= 0) { + ex::set_value(std::move(receiver), std::move(res_code), + std::move(payload)); + } else if (res_code == -ECANCELED) { + ex::set_stopped(std::move(receiver)); + } else { + ex::set_error( + std::move(receiver), + std::error_code(-res_code, std::generic_category())); + } + } + + template + void operator()(std::tuple res) noexcept { + std::apply( + [this](auto &&...args) { + ex::set_value(std::move(receiver), + std::forward(args)...); + }, + std::move(res)); + } + + template void operator()(T &&res) noexcept { + ex::set_value(std::move(receiver), std::forward(res)); + } + + auto get_stop_token() const noexcept { + auto env = ex::get_env(receiver); + return ex::get_stop_token(env); + } + }; + + template struct OperationStateWrapper { + using operation_state_concept = ex::operation_state_tag; + + OperationState op_state; + void start() noexcept { op_state.start(0); } + }; +}; + +template struct wait_variant { + static_assert(sizeof...(Ts) == 1, "Requires only one value type"); +}; +template struct wait_variant { + using type = T; +}; +template +using wait_variant_t = typename wait_variant::type; + +template class [[nodiscard]] WaitSenderAwaiter { +private: + struct Env { + Runtime &runtime; + Scheduler query(ex::get_scheduler_t) const noexcept { + return Scheduler{runtime}; + } + Scheduler query(ex::get_start_scheduler_t) const noexcept { + return Scheduler{runtime}; + } + Scheduler query(ex::get_delegation_scheduler_t) const noexcept { + return Scheduler{runtime}; + } + }; + using ValueType = + ex::value_types_of_t; + +public: + WaitSenderAwaiter(Sender sender, Runtime &runtime) + : runtime_(runtime), + op_state_(ex::connect(std::move(sender), Receiver{this})) {} + + CONDY_DELETE_COPY_MOVE(WaitSenderAwaiter); + +public: + bool await_ready() const noexcept { return false; } + + template + bool await_suspend(std::coroutine_handle handle) noexcept { + runtime_.pend_work_internal(); + ex::start(op_state_); + auto h = std::exchange(handle_, handle); + return h == std::noop_coroutine(); + } + + std::optional await_resume() { + runtime_.resume_work_internal(); + auto &result = *result_; + if (result.index() == 1) { + std::rethrow_exception(std::get<1>(std::move(result))); + } + if (result.index() == 2) { + return std::nullopt; + } + return std::get<0>(std::move(result)); + } + +private: + struct Receiver { + using receiver_concept = ex::receiver_tag; + WaitSenderAwaiter *self; + + template void set_value(Args &&...args) noexcept { + self->complete_value_(std::forward(args)...); + } + + template void set_error(E &&e) noexcept { + self->complete_error_(std::forward(e)); + } + + void set_stopped() noexcept { self->complete_stopped_(); } + + auto get_env() const noexcept { return Env{self->runtime_}; } + }; + + template static std::exception_ptr to_exception_(E &&e) { + if constexpr (std::is_same_v, std::exception_ptr>) { + return std::forward(e); + } else if constexpr (std::is_same_v, std::error_code>) { + return std::make_exception_ptr( + std::system_error(std::forward(e))); + } else { + return std::make_exception_ptr(std::forward(e)); + } + } + + template void complete_value_(Args &&...args) noexcept { + result_.emplace(std::in_place_index<0>, + ValueType(std::forward(args)...)); + resume_(); + } + + template void complete_error_(E &&e) noexcept { + result_.emplace(std::in_place_index<1>, + to_exception_(std::forward(e))); + resume_(); + } + + void complete_stopped_() noexcept { + result_.emplace(std::in_place_index<2>, std::monostate{}); + resume_(); + } + + void resume_() noexcept { + auto h = std::exchange(handle_, nullptr); + h.resume(); + } + + using ResultType = + std::variant; + + Runtime &runtime_; + ex::connect_result_t op_state_; + std::coroutine_handle<> handle_ = std::noop_coroutine(); + std::optional result_; +}; + +} // namespace detail + +} // namespace condy + +#endif \ No newline at end of file diff --git a/include/condy/execution.hpp b/include/condy/execution.hpp new file mode 100644 index 00000000..96875044 --- /dev/null +++ b/include/condy/execution.hpp @@ -0,0 +1,42 @@ +/** + * @file execution.hpp + * @brief std::execution integration with condy's runtime. + */ + +#pragma once + +#ifdef CONDY_HAS_STDEXEC + +#include "condy/detail/execution.hpp" +#include "condy/runtime.hpp" +#include + +namespace condy { + +/** + * @brief Get a scheduler for the given runtime. + * @param runtime The runtime to schedule work onto. + * @return A scheduler backed by `runtime`. + */ +inline auto get_scheduler(Runtime &runtime) { + return detail::Scheduler{runtime}; +} + +/** + * @brief Convert a standard sender into a plain awaiter. + * @details Lets a condy coroutine `co_await` any std::execution sender. The + * result follows `ex::sync_wait`. When the sender completes, execution resumes + * on the runtime that started this operation. + * @param sender The std::execution sender to wait for. + * @return Awaiter usable with `co_await` in condy coroutines. + */ +template auto wait_sender(Sender &&sender) { + auto &runtime = current_runtime(); + auto sched = get_scheduler(runtime); + auto s = std::forward(sender) | ex::continues_on(sched); + return detail::WaitSenderAwaiter(std::move(s), runtime); +} + +} // namespace condy + +#endif \ No newline at end of file diff --git a/include/condy/futex.hpp b/include/condy/futex.hpp index f9351837..5baedab6 100644 --- a/include/condy/futex.hpp +++ b/include/condy/futex.hpp @@ -14,6 +14,9 @@ #include #include #include +#ifdef CONDY_HAS_STDEXEC +#include "condy/detail/execution.hpp" +#endif namespace condy { @@ -38,8 +41,15 @@ template class Futex { CONDY_DELETE_COPY_MOVE(Futex); +private: + struct [[nodiscard]] WaitSenderImpl; + public: - struct [[nodiscard]] WaitSender; +#ifdef CONDY_HAS_STDEXEC + using WaitSender = detail::StandardSender; +#else + using WaitSender = WaitSenderImpl; +#endif /** * @brief Wait if the futex value equals to the specified old value. The * awaiting coroutine will be suspended until a notify is received. If the @@ -191,12 +201,12 @@ class Futex::WaitFinishHandle std::optional stop_callback_; }; -template struct Futex::WaitSender { +template struct Futex::WaitSenderImpl { public: using CondySender = void; using ReturnType = int32_t; - WaitSender(Futex &futex, T old) : futex_(futex), old_(old) {} + WaitSenderImpl(Futex &futex, T old) : futex_(futex), old_(old) {} template auto connect_impl(Receiver receiver) noexcept { return OperationState(futex_, old_, std::move(receiver)); diff --git a/include/condy/senders.hpp b/include/condy/senders.hpp index 191350dd..b3ff59fa 100644 --- a/include/condy/senders.hpp +++ b/include/condy/senders.hpp @@ -6,11 +6,77 @@ #pragma once #include "condy/detail/senders.hpp" +#ifdef CONDY_HAS_STDEXEC +#include "condy/detail/execution.hpp" +#endif namespace condy { -// TODO: This re-export is intentional. We may adapt these senders to standard -// sender/receiver concepts in the future. +#ifdef CONDY_HAS_STDEXEC + +template +using OpSender = detail::StandardSender>; + +template +using MultiShotOpSender = detail::StandardSender< + detail::MultiShotOpSender>; + +template +using ZeroCopyOpSender = detail::StandardSender< + detail::ZeroCopyOpSender>; + +template +using FlaggedOpSender = + detail::StandardSender>; + +template +using ParallelAllSender = + detail::StandardSender>; + +template +using ParallelAnySender = + detail::StandardSender>; + +template +using WhenAllSender = detail::StandardSender>; + +template +using WhenAnySender = detail::StandardSender>; + +template +using LinkSender = detail::StandardSender>; + +template +using HardLinkSender = + detail::StandardSender>; + +template +using RangedParallelAllSender = + detail::StandardSender>; + +template +using RangedParallelAnySender = + detail::StandardSender>; + +template +using RangedWhenAllSender = + detail::StandardSender>; + +template +using RangedWhenAnySender = + detail::StandardSender>; + +template +using RangedLinkSender = + detail::StandardSender>; + +template +using RangedHardLinkSender = + detail::StandardSender>; + +#else + using detail::FlaggedOpSender; using detail::HardLinkSender; using detail::LinkSender; @@ -28,4 +94,6 @@ using detail::WhenAllSender; using detail::WhenAnySender; using detail::ZeroCopyOpSender; +#endif + } // namespace condy \ No newline at end of file diff --git a/module/condy.cppm b/module/condy.cppm index cb167d00..fe21c3ac 100644 --- a/module/condy.cppm +++ b/module/condy.cppm @@ -30,6 +30,12 @@ using condy::Task; using condy::default_runtime_options; using condy::sync_wait; +// execution.hpp +#ifdef CONDY_HAS_STDEXEC +using condy::get_scheduler; +using condy::wait_sender; +#endif + // buffers.hpp using condy::buffer; using condy::ConstBuffer; diff --git a/tests/test_execution.cpp b/tests/test_execution.cpp new file mode 100644 index 00000000..7006622c --- /dev/null +++ b/tests/test_execution.cpp @@ -0,0 +1,320 @@ +#ifdef CONDY_HAS_STDEXEC + +#include "condy/async_operations.hpp" +#include "condy/channel.hpp" +#include "condy/execution.hpp" +#include "condy/runtime.hpp" +#include "condy/sender_operations.hpp" +#include "condy/sync_wait.hpp" +#include +#include +#include +#include + +namespace ex = stdexec; + +TEST_CASE("test execution - schedule") { + condy::Runtime runtime; + std::thread::id runtime_thread_id; + std::thread runtime_thread([&] { + runtime_thread_id = std::this_thread::get_id(); + runtime.run(); + }); + + auto scheduler = condy::get_scheduler(runtime); + + bool executed = false; + ex::sender auto sender = ex::schedule(scheduler) | ex::then([&] { + executed = true; + return std::this_thread::get_id(); + }); + + auto [thread_id] = ex::sync_wait(sender).value(); + REQUIRE(executed); + REQUIRE(thread_id == runtime_thread_id); + REQUIRE(runtime_thread_id != std::this_thread::get_id()); + + runtime.allow_exit(); + runtime_thread.join(); +} + +TEST_CASE("test execution - sender") { + condy::Runtime runtime; + std::thread runtime_thread([&] { runtime.run(); }); + + auto scheduler = condy::get_scheduler(runtime); + + bool executed = false; + ex::sender auto sender = ex::schedule(scheduler) | ex::let_value([&] { + executed = true; + return condy::async_nop(); + }); + + auto [r] = ex::sync_wait(sender).value(); + REQUIRE(executed); + REQUIRE(r == 0); + + runtime.allow_exit(); + runtime_thread.join(); +} + +TEST_CASE("test execution - when_all") { + condy::Runtime runtime; + std::thread runtime_thread([&] { runtime.run(); }); + + auto scheduler = condy::get_scheduler(runtime); + + bool executed1 = false; + bool executed2 = false; + + auto sender1 = ex::schedule(scheduler) | ex::then([&] { + executed1 = true; + return 42; + }); + auto sender2 = ex::schedule(scheduler) | ex::then([&] { + executed2 = true; + return 0; + }); + + auto when_all_sender = ex::when_all(sender1, sender2); + auto [r1, r2] = ex::sync_wait(when_all_sender).value(); + + REQUIRE(executed1); + REQUIRE(executed2); + REQUIRE(r1 == 42); + REQUIRE(r2 == 0); + + runtime.allow_exit(); + runtime_thread.join(); +} + +TEST_CASE("test execution - when_any") { + condy::Runtime runtime; + std::thread runtime_thread([&] { runtime.run(); }); + + auto scheduler = condy::get_scheduler(runtime); + + __kernel_timespec ts = { + .tv_sec = 60ll * 60ll, + .tv_nsec = 0, + }; + auto sender = ex::schedule(scheduler) | ex::let_value([&] { + return exec::when_any(condy::async_timeout(&ts, 0, 0), + condy::async_nop()); + }); + + auto [r] = ex::sync_wait(sender).value(); + REQUIRE(r == 0); + + runtime.allow_exit(); + runtime_thread.join(); +} + +TEST_CASE("test execution - when_any with different thread") { + condy::Runtime runtime1, runtime2; + std::thread thread1([&] { runtime1.run(); }); + std::thread thread2([&] { runtime2.run(); }); + + auto scheduler1 = condy::get_scheduler(runtime1); + auto scheduler2 = condy::get_scheduler(runtime2); + + __kernel_timespec ts = { + .tv_sec = 60ll * 60ll, + .tv_nsec = 0, + }; + auto sender1 = ex::schedule(scheduler1) | ex::let_value([&] { + return condy::async_timeout(&ts, 0, 0); + }); + auto sender2 = ex::schedule(scheduler2) | ex::then([] { return 42; }); + auto when_any_sender = exec::when_any(sender1, sender2); + auto [r] = ex::sync_wait(when_any_sender).value(); + REQUIRE(r == 42); + + runtime1.allow_exit(); + runtime2.allow_exit(); + thread1.join(); + thread2.join(); +} + +TEST_CASE("test execution - condy when_all") { + using condy::operators::operator&&; + + condy::Runtime runtime; + std::thread runtime_thread([&] { runtime.run(); }); + auto scheduler = condy::get_scheduler(runtime); + + ex::sender auto sender = ex::schedule(scheduler) | ex::let_value([&] { + return condy::async_nop() && + condy::async_nop() && + condy::async_nop(); + }) | ex::then([](int r1, int r2, int r3) { + REQUIRE(r1 == 0); + REQUIRE(r2 == 0); + REQUIRE(r3 == 0); + return r1 + r2 + r3; + }); + + auto [total] = ex::sync_wait(sender).value(); + REQUIRE(total == 0); + + runtime.allow_exit(); + runtime_thread.join(); +} + +TEST_CASE("test execution - stdexec task") { + condy::Runtime runtime; + std::thread runtime_thread([&] { runtime.run(); }); + + auto sch = condy::get_scheduler(runtime); + + auto my_task = []() -> ex::task { + auto r = co_await condy::async_nop(); + REQUIRE(r == 0); + co_return 42; + }; + + auto [result] = ex::sync_wait(ex::starts_on(sch, my_task())).value(); + REQUIRE(result == 42); + + runtime.allow_exit(); + runtime_thread.join(); +} + +TEST_CASE("test execution - spawn with simple_counting_scope") { + condy::Runtime runtime; + std::thread runtime_thread([&] { runtime.run(); }); + + auto sch = condy::get_scheduler(runtime); + ex::simple_counting_scope scope; + + std::atomic counter{0}; + ex::spawn(ex::schedule(sch) | + ex::then([&]() noexcept { counter.fetch_add(1); }), + scope.get_token()); + ex::spawn(ex::schedule(sch) | + ex::then([&]() noexcept { counter.fetch_add(1); }), + scope.get_token()); + + ex::sync_wait(scope.join()); + + REQUIRE(counter == 2); + + runtime.allow_exit(); + runtime_thread.join(); +} + +namespace { + +struct StoppedSender { + using sender_concept = ex::sender_t; + using completion_signatures = + ex::completion_signatures; + + template struct OperationState { + Receiver receiver; + void start() noexcept { ex::set_stopped(std::move(receiver)); } + }; + + template + OperationState> connect(Receiver &&receiver) { + return {std::forward(receiver)}; + } +}; + +} // namespace + +TEST_CASE("test execution - wait_sender") { + condy::Runtime runtime; + + auto co = []() -> condy::Coro<> { + // single value -> std::optional> + auto v = co_await condy::wait_sender(ex::just(42)); + REQUIRE(v.has_value()); + REQUIRE(std::get<0>(*v) == 42); + + // no value -> std::optional> + auto empty = co_await condy::wait_sender(ex::just()); + REQUIRE(empty.has_value()); + + // multiple values -> std::optional> + auto t = co_await condy::wait_sender(ex::just(1, 2.0)); + REQUIRE(t.has_value()); + REQUIRE(std::get<0>(*t) == 1); + REQUIRE(std::get<1>(*t) == 2.0); + + // condy sender works too (StandardSender is a standard sender) + auto r = co_await condy::wait_sender(condy::async_nop()); + REQUIRE(r.has_value()); + REQUIRE(std::get<0>(*r) == 0); + + // error: throws + REQUIRE_THROWS_AS( + co_await condy::wait_sender(ex::just(42) | ex::then([](int) -> int { + throw std::runtime_error("boom"); + })), + std::runtime_error); + + // stopped: returns nullopt + auto s = co_await condy::wait_sender(StoppedSender{}); + REQUIRE(!s.has_value()); + + // composed sender + auto n = co_await condy::wait_sender( + ex::just(7) | ex::then([](int x) { return x * 2; })); + REQUIRE(n.has_value()); + REQUIRE(std::get<0>(*n) == 14); + }(); + + condy::sync_wait(runtime, std::move(co)); +} + +TEST_CASE("test execution - wait_sender cross runtime") { + condy::Runtime runtime1, runtime2; + std::thread thread2([&] { runtime2.run(); }); + + auto func = [&]() -> condy::Coro<> { + std::thread::id then_tid; + auto before_tid = std::this_thread::get_id(); + auto r = co_await condy::wait_sender( + ex::schedule(condy::get_scheduler(runtime2)) | ex::then([&] { + then_tid = std::this_thread::get_id(); + return 42; + })); + REQUIRE(r.has_value()); + REQUIRE(std::get<0>(*r) == 42); + + auto after_tid = std::this_thread::get_id(); + REQUIRE(after_tid == before_tid); + REQUIRE(then_tid != before_tid); + }; + + condy::sync_wait(runtime1, func()); + + runtime2.allow_exit(); + thread2.join(); +} + +TEST_CASE("test execution - channel pop sender") { + condy::Runtime runtime; + std::thread runtime_thread([&] { runtime.run(); }); + auto scheduler = condy::get_scheduler(runtime); + + condy::Channel ch(1); + + REQUIRE(ch.try_push(42) == 0); + + auto [item] = ex::sync_wait(ex::schedule(scheduler) | + ex::let_value([&] { return ch.pop(); }) | + ex::then([](auto r, auto item) { + REQUIRE(r == 0); + REQUIRE(item == 42); + return item; + })) + .value(); + REQUIRE(item == 42); + + runtime.allow_exit(); + runtime_thread.join(); +} + +#endif \ No newline at end of file diff --git a/third_party/stdexec b/third_party/stdexec new file mode 160000 index 00000000..0a6c3fc4 --- /dev/null +++ b/third_party/stdexec @@ -0,0 +1 @@ +Subproject commit 0a6c3fc4324aed2ce291d49dd228cc3619b24dad