diff --git a/include/beman/execution/detail/logger.hpp b/include/beman/execution/detail/logger.hpp deleted file mode 100644 index 4fea369e..00000000 --- a/include/beman/execution/detail/logger.hpp +++ /dev/null @@ -1,40 +0,0 @@ -// include/beman/execution/detail/logger.hpp -*-C++-*- -// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception - -#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_LOGGER -#define INCLUDED_BEMAN_EXECUTION_DETAIL_LOGGER - -#include -#include -#include - -// ---------------------------------------------------------------------------- - -namespace beman::execution::detail { -struct logger { - static auto level(int i) -> int { - static int rc{}; - return rc += i; - } - ::std::ostream& log(const char* pre, const char* msg) { - ::std::fill_n(::std::ostreambuf_iterator(std::cout.rdbuf()), level(0), ' '); - std::cout << pre; - return std::cout << msg << "\n"; - } - ::std::ostream& log(const char* msg) { return log("| ", msg); } - const char* msg; - explicit logger(const char* m) : msg(m) { - level(1); - log("\\ ", this->msg); - } - logger(logger&&) = delete; - ~logger() { - log("/ ", this->msg); - level(-1); - } -}; -} // namespace beman::execution::detail - -// ---------------------------------------------------------------------------- - -#endif diff --git a/include/beman/execution/detail/psched_bulk_sender.hpp b/include/beman/execution/detail/psched_bulk_sender.hpp index 9bd8c9d4..30228f74 100644 --- a/include/beman/execution/detail/psched_bulk_sender.hpp +++ b/include/beman/execution/detail/psched_bulk_sender.hpp @@ -9,6 +9,7 @@ #ifdef BEMAN_HAS_IMPORT_STD import std; #else +#include #include #include #include @@ -107,13 +108,19 @@ struct psched_bulk_sender { const Shape first = is_parallel_policy ? static_cast(begin) : Shape(0); const Shape last = is_parallel_policy ? static_cast(end) : shape; const auto call_fn = [=, this](Args&... args) { - if constexpr (IsChunked) { - static_assert(::std::invocable); - fn(first, last, args...); - } else { - static_assert(::std::invocable); - for (Shape i = first; i < last; ++i) { - fn(i, args...); + try { + if constexpr (IsChunked) { + static_assert(::std::invocable); + fn(first, last, args...); + } else { + static_assert(::std::invocable); + for (Shape i = first; i < last; ++i) { + fn(i, args...); + } + } + } catch (...) { + if (!has_exception.exchange(true, ::std::memory_order_relaxed)) { + exception = ::std::current_exception(); } } }; @@ -129,6 +136,10 @@ struct psched_bulk_sender { } auto set_value() noexcept -> void final { + if (exception) [[unlikely]] { + this->set_error(exception); + return; + } const auto call_set_value = [this](Args&&... args) { ::beman::execution::set_value(::std::move(rcvr), ::std::forward(args)...); }; @@ -170,6 +181,8 @@ struct psched_bulk_sender { Policy policy; Shape shape; Fn fn; + ::std::atomic has_exception{false}; + ::std::exception_ptr exception; }; template diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index 3df1a8c5..51cec4b2 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -131,7 +131,6 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/join_env.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/just.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/let.hpp - ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/logger.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/make_env.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/make_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/matching_sig.hpp diff --git a/tests/beman/execution/exec-inline-scheduler.test.cpp b/tests/beman/execution/exec-inline-scheduler.test.cpp index fb0a74c1..fba01583 100644 --- a/tests/beman/execution/exec-inline-scheduler.test.cpp +++ b/tests/beman/execution/exec-inline-scheduler.test.cpp @@ -96,18 +96,7 @@ struct sched_env_receiver { bool& called; struct env { - auto query(test_std::get_scheduler_t) const noexcept { return custom_scheduler{42}; } - }; - auto set_value() && noexcept -> void { called = true; } - auto get_env() const noexcept { return env{}; } -}; - -struct domain_env_receiver { - using receiver_concept = test_std::receiver_tag; - bool& called; - - struct env { - auto query(test_std::get_scheduler_t) const noexcept { return sched_with_domain{}; } + auto query(test_std::get_start_scheduler_t) const noexcept { return custom_scheduler{42}; } }; auto set_value() && noexcept -> void { called = true; } auto get_env() const noexcept { return env{}; } @@ -186,9 +175,13 @@ auto test_inline_attrs_completion_domain() { } auto test_connect_with_scheduler_env() { - bool called = false; - auto sndr = test_std::schedule(test_std::inline_scheduler{}); - auto op = test_std::connect(sndr, sched_env_receiver{called}); + bool called = false; + auto sndr = test_std::schedule(test_std::inline_scheduler{}); + sched_env_receiver rcvr{called}; + auto sch = + test_std::get_completion_scheduler(test_std::get_env(sndr), test_std::get_env(rcvr)); + ASSERT(sch == custom_scheduler{42}); + auto op = test_std::connect(sndr, rcvr); ASSERT(!called); test_std::start(op); ASSERT(called); diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp index 2acbb546..c7daa1b6 100644 --- a/tests/beman/execution/exec-parallel-scheduler.test.cpp +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -219,6 +219,34 @@ auto test_parallel_scheduler_schedule() -> void { } } } + +auto test_parallel_scheduler_exception_propagation() -> void { + auto sch = test_std::get_parallel_scheduler(); + try { + test_std::sync_wait(test_std::schedule(sch) | test_std::then([] { throw 114; })); + } catch (int n) { + ASSERT(n == 114); + } + try { + test_std::sync_wait(test_std::schedule(sch) | test_std::bulk(test_std::par, 4, [](int i) { + if (i == 2) { + throw 514; + } + })); + } catch (int n) { + ASSERT(n == 514); + } + try { + test_std::sync_wait(test_std::schedule(sch) | test_std::bulk(test_std::seq, 3, [](int i) { + if (i == 1) { + throw 114514; + } + })); + } catch (int n) { + ASSERT(n == 114514); + } +} + } // namespace #if !BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND @@ -234,4 +262,5 @@ TEST(exec_parallel_scheduler) { test_parallel_scheduler_synopsis(); test_replaceability_synopsis(); test_parallel_scheduler_schedule(); + test_parallel_scheduler_exception_propagation(); }