Skip to content

Propagate exceptions thrown by parallel_scheduler bulk functions #328

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dietmarkuehl merged 2 commits into from
Oct 17, 2026
Merged

Propagate exceptions thrown by parallel_scheduler bulk functions #328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8dfea05
Propagate exceptions thrown by parallel_scheduler bulk functions
Cra3z Oct 17, 2026
289daa3
Format code
Cra3z Oct 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 0 additions & 40 deletions include/beman/execution/detail/logger.hpp
View file
Open in desktop

This file was deleted.

27 changes: 20 additions & 7 deletions include/beman/execution/detail/psched_bulk_sender.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
#ifdef BEMAN_HAS_IMPORT_STD
import std;
#else
#include <atomic>
#include <cstddef>
#include <concepts>
#include <exception>
Expand Down Expand Up @@ -107,13 +108,19 @@ struct psched_bulk_sender {
const Shape first = is_parallel_policy ? static_cast<Shape>(begin) : Shape(0);
const Shape last = is_parallel_policy ? static_cast<Shape>(end) : shape;
const auto call_fn = [=, this]<typename... Args>(Args&... args) {
if constexpr (IsChunked) {
static_assert(::std::invocable<Fn, Shape, Shape, Args&...>);
fn(first, last, args...);
} else {
static_assert(::std::invocable<Fn, Shape, Args&...>);
for (Shape i = first; i < last; ++i) {
fn(i, args...);
try {
if constexpr (IsChunked) {
static_assert(::std::invocable<Fn, Shape, Shape, Args&...>);
fn(first, last, args...);
} else {
static_assert(::std::invocable<Fn, Shape, Args&...>);
for (Shape i = first; i < last; ++i) {
fn(i, args...);
}
}
} catch (...) {
if (!has_exception.exchange(true, ::std::memory_order_relaxed)) {
exception = ::std::current_exception();
}
}
};
Expand All @@ -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]<typename... Args>(Args&&... args) {
::beman::execution::set_value(::std::move(rcvr), ::std::forward<Args>(args)...);
};
Expand Down Expand Up @@ -170,6 +181,8 @@ struct psched_bulk_sender {
Policy policy;
Shape shape;
Fn fn;
::std::atomic<bool> has_exception{false};
::std::exception_ptr exception;
};

template <typename Rcvr>
Expand Down
1 change: 0 additions & 1 deletion src/beman/execution/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
23 changes: 8 additions & 15 deletions tests/beman/execution/exec-inline-scheduler.test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{}; }
Expand Down Expand Up @@ -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::set_value_t>(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);
Expand Down
29 changes: 29 additions & 0 deletions tests/beman/execution/exec-parallel-scheduler.test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -234,4 +262,5 @@ TEST(exec_parallel_scheduler) {
test_parallel_scheduler_synopsis();
test_replaceability_synopsis();
test_parallel_scheduler_schedule();
test_parallel_scheduler_exception_propagation();
}
Loading