Skip to content

Fix bulk to pass the child sender's results to the callback as lvalue references #325

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 10, 2026
+82 −23
Merged

Fix bulk to pass the child sender's results to the callback as lvalue references #325

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
bfb9393
Fix `bulk` to pass the child sender's results to the callback as lval…
Cra3z Oct 9, 2026
98ae23d
Format code
Cra3z Oct 9, 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
12 changes: 6 additions & 6 deletions include/beman/execution/detail/bulk.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,20 +62,20 @@ struct bulk_traits<true, F, Shape, Args...> { // for bulk_chunked

static auto invoke(F& fn, Shape shape, Args&... args) noexcept(is_nothrow_invocable) -> void {
if (shape > static_cast<Shape>(0)) [[likely]] {
std::invoke(fn, 0, shape, args...);
::std::invoke(fn, 0, shape, args...);
}
}
};

template <typename F, typename Shape, typename... Args>
struct bulk_traits<false, F, Shape, Args...> { // for bulk_unchunked
static constexpr bool is_invocable = ::std::invocable<F, Shape, Args...>;
static constexpr bool is_invocable = ::std::invocable<F&, Shape, Args&...>;

static constexpr bool is_nothrow_invocable = ::std::is_nothrow_invocable_v<F, Shape, Args...>;
static constexpr bool is_nothrow_invocable = ::std::is_nothrow_invocable_v<F&, Shape, Args&...>;

static auto invoke(F& fn, Shape shape, Args&... args) noexcept(is_nothrow_invocable) -> void {
for (auto i = static_cast<Shape>(0); i < shape; ++i) {
std::invoke(fn, i, args...);
::std::invoke(fn, i, args...);
}
}
};
Expand Down Expand Up @@ -242,10 +242,10 @@ struct bulk_t {

template <typename Shape, typename Fn>
static auto wrap_chunked(Fn f) noexcept {
return [f = std::move(f)]<typename... Args>(Shape begin, Shape end, Args&&... args) noexcept(
return [f = ::std::move(f)]<typename... Args>(Shape begin, Shape end, Args&&... args) noexcept(
::std::is_nothrow_invocable_v<Fn&, Shape, Args&...>) {
while (begin != end) {
std::invoke(f, begin++, args...);
::std::invoke(f, begin++, args...);
}
};
}
Expand Down
12 changes: 6 additions & 6 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 @@ -106,12 +106,12 @@ struct psched_bulk_sender {
assert(IsChunked || !is_parallel_policy || end - begin == 1uz);
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>(const Args&... args) {
const auto call_fn = [=, this]<typename... Args>(Args&... args) {
if constexpr (IsChunked) {
static_assert(::std::invocable<Fn, Shape, Shape, const Args&...>);
static_assert(::std::invocable<Fn, Shape, Shape, Args&...>);
fn(first, last, args...);
} else {
static_assert(::std::invocable<Fn, Shape, const Args&...>);
static_assert(::std::invocable<Fn, Shape, Args&...>);
for (Shape i = first; i < last; ++i) {
fn(i, args...);
}
Expand All @@ -122,7 +122,7 @@ struct psched_bulk_sender {
if constexpr (::std::same_as<T, ::std::monostate>) {
::beman::execution::detail::unreachable();
} else {
::std::apply(call_fn, ::std::move(tpl));
::std::apply(call_fn, tpl);
}
},
result);
Expand All @@ -135,7 +135,7 @@ struct psched_bulk_sender {
try {
::std::visit(
[&]<typename T>(T tpl) {
if constexpr (::std::same_as<T, std::monostate>) {
if constexpr (::std::same_as<T, ::std::monostate>) {
::beman::execution::detail::unreachable();
} else {
::std::apply(call_set_value, ::std::move(tpl));
Expand Down Expand Up @@ -212,7 +212,7 @@ struct psched_bulk_sender {
using sub_state_t = ::beman::execution::connect_result_t<Child, receiver_ref>;

state(::std::shared_ptr<backend_type> backend, Child child, Rcvr rcvr, Policy policy, Shape shape, Fn fn)
: proxy(std::move(backend), ::std::move(rcvr), ::std::move(policy), ::std::move(shape), ::std::move(fn)),
: proxy(::std::move(backend), ::std::move(rcvr), ::std::move(policy), ::std::move(shape), ::std::move(fn)),
sub_state(::beman::execution::connect(::std::move(child), receiver_ref{&proxy})) {}

auto start() & noexcept -> void { ::beman::execution::start(sub_state); }
Expand Down
18 changes: 9 additions & 9 deletions include/beman/execution/detail/task.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ struct allocator_support {
}

template <typename... Args>
auto operator new(std::size_t size, const Args&... args) -> void* {
auto operator new(::std::size_t size, const Args&... args) -> void* {
if constexpr (::std::same_as<Allocator, ::std::allocator<::std::byte>>) {
Allocator alloc{};
return allocator_traits::allocate(alloc, size);
Expand Down Expand Up @@ -313,7 +313,7 @@ class result_type<Value, ::beman::execution::completion_signatures<::beman::exec
::beman::execution::detail::sub_visit<2uz>(
[]<typename E>(E& error) {
if constexpr (::std::same_as<::std::remove_cvref_t<E>, ::std::exception_ptr>) {
std::rethrow_exception(::std::move(error));
::std::rethrow_exception(::std::move(error));
} else if constexpr (::std::same_as<::std::remove_cvref_t<E>, ::std::error_code>) {
throw ::std::system_error{::std::move(error)};
} else {
Expand All @@ -339,7 +339,7 @@ class result_type<Value, ::beman::execution::completion_signatures<>> {
private:
using value_type = ::std::conditional_t<::std::same_as<void, Value>, void_type, Value>;

::std::variant<std::monostate, value_type> result;
::std::variant<::std::monostate, value_type> result;

public:
template <typename T>
Expand Down Expand Up @@ -463,7 +463,7 @@ auto make_sched(const Env& env, const Alloc& alloc) noexcept -> Sched {
::std::make_obj_using_allocator<Sched>(alloc, ::beman::execution::get_scheduler(env));
}) {
return ::std::make_obj_using_allocator<Sched>(alloc, ::beman::execution::get_scheduler(env));
} else if constexpr (std::default_initializable<Sched>) {
} else if constexpr (::std::default_initializable<Sched>) {
return Sched();
} else {
static_assert(false,
Expand Down Expand Up @@ -494,7 +494,7 @@ class state_base : public result_type<Value, error_types_of_t<Environment>> {

auto operator=(state_base&&) -> state_base& = delete;

auto complete() noexcept -> std::coroutine_handle<> { return this->do_complete(); }
auto complete() noexcept -> ::std::coroutine_handle<> { return this->do_complete(); }
auto get_allocator() noexcept -> allocator_type { return this->do_get_allocator(); }
auto get_stop_token() noexcept -> stop_token_type { return this->do_get_stop_token(); }
auto get_environment() noexcept -> Environment& { return this->do_get_environment(); }
Expand All @@ -521,7 +521,7 @@ struct state : state_base<Value, Environment> {
template <typename H, typename Env>
state(Receiver r, H h, const Env& env) noexcept
: rcvr(::std::move(r)),
handle(std::move(h)),
handle(::std::move(h)),
holder(env),
scheduler(::beman::execution::detail::task::make_sched<scheduler_type>(env, do_get_allocator())),
propagator(::beman::execution::get_stop_token(env)) {}
Expand Down Expand Up @@ -570,7 +570,7 @@ class awaiter : public state_base<Value, Environment> {

explicit awaiter(::std::coroutine_handle<ParentPromise> parent, handle<OwnPromise> h, const auto& env) noexcept
: parent(parent),
handle(std::move(h)),
handle(::std::move(h)),
holder(env),
scheduler(::beman::execution::detail::task::make_sched<scheduler_type>(env, do_get_allocator())),
propagator(::beman::execution::get_stop_token(env)) {}
Expand All @@ -586,7 +586,7 @@ class awaiter : public state_base<Value, Environment> {
auto await_resume() -> Value { return this->result_resume(); }

private:
auto do_complete() noexcept -> std::coroutine_handle<> final {
auto do_complete() noexcept -> ::std::coroutine_handle<> final {
if (this->no_completion_set()) {
return this->parent.promise().unhandled_stopped();
}
Expand Down Expand Up @@ -709,7 +709,7 @@ class promise_type : public ::beman::execution::detail::task::promise_base<::std
}
}

auto unhandled_stopped() -> std::coroutine_handle<> { return this->get_state()->complete(); }
auto unhandled_stopped() -> ::std::coroutine_handle<> { return this->get_state()->complete(); }

auto get_return_object() noexcept {
return Coroutine(::beman::execution::detail::task::handle<promise_type>(this));
Expand Down
4 changes: 2 additions & 2 deletions include/beman/execution/detail/task_scheduler.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ struct task_scheduler_backend : ::beman::execution::parallel_scheduler_replaceme
allocated_state_holder(Sndr sndr,
::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy,
Alloc alloc)
: alloc(std::move(alloc)),
: alloc(::std::move(alloc)),
op(::beman::execution::connect(::std::move(sndr), rcvr_wrapper{proxy, *this})) {}

auto destroy() noexcept -> void override { ::beman::execution::detail::delete_object(alloc, this); }
Expand Down Expand Up @@ -378,7 +378,7 @@ class task_scheduler {

private:
explicit task_scheduler(::std::shared_ptr<::beman::execution::detail::task_scheduler_backend> backend) noexcept
: backend_(std::move(backend)) {}
: backend_(::std::move(backend)) {}

::std::shared_ptr<::beman::execution::detail::task_scheduler_backend> backend_;
};
Expand Down
39 changes: 39 additions & 0 deletions tests/beman/execution/exec-bulk.test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,44 @@ auto test_bulk() {
}
}

auto test_bulk_predecessor_result() {
{
auto sndr = test_std::just(std::make_unique<int>(114)) |
test_std::bulk(test_std::par, 1, [](int i, std::unique_ptr<int>& ptr) noexcept {
ASSERT(ptr && *ptr == 114);
ASSERT(i == 0);
});
auto [ptr] = test_std::sync_wait(std::move(sndr)).value();
ASSERT(ptr && *ptr == 114);
}
{
auto sndr = test_std::just(std::make_unique<int>(514)) |
test_std::bulk_chunked(test_std::par, 1, [](int i, int j, std::unique_ptr<int>& ptr) noexcept {
ASSERT(ptr && *ptr == 514);
ASSERT(i == 0 && j == 1);
});
auto [ptr] = test_std::sync_wait(std::move(sndr)).value();
ASSERT(ptr && *ptr == 514);
}
{
auto sndr = test_std::just(std::make_unique<int>(114)) |
test_std::bulk_unchunked(test_std::par, 1, [](int i, std::unique_ptr<int>& ptr) noexcept {
ASSERT(ptr && *ptr == 114);
ASSERT(i == 0);
});
auto [ptr] = test_std::sync_wait(std::move(sndr)).value();
ASSERT(ptr && *ptr == 114);
}
{
auto [vec] =
test_std::sync_wait(
test_std::just(std::vector{1, 2, 3, 4}) |
test_std::bulk(test_std::par, 4uz, [](std::size_t i, std::vector<int>& vec) noexcept { vec[i] *= 2; }))
.value();
ASSERT((vec == std::vector{2, 4, 6, 8}));
}
}

auto test_bulk_noexcept() {
auto b0 = test_std::bulk(test_std::just(), test_std::seq, 1, [](int) noexcept {});
auto b0_env = test_std::get_env(b0);
Expand Down Expand Up @@ -560,6 +598,7 @@ TEST(exec_bulk) {
try {

test_bulk();
test_bulk_predecessor_result();
test_bulk_noexcept();
test_bulk_pipeable();
test_bulk_chunked();
Expand Down
20 changes: 20 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 @@ -169,6 +169,26 @@ auto test_parallel_scheduler_schedule() -> void {
}
}
}
{
auto sndr = test_std::schedule(sch) | test_std::then([] { return std::make_unique<int>(114); }) |
test_std::bulk(test_std::par, 1, [](int i, std::unique_ptr<int>& ptr) noexcept {
ASSERT(ptr && *ptr == 114);
ASSERT(i == 0);
});
if (auto result = test_std::sync_wait(std::move(sndr))) {
auto [ptr] = std::move(*result);
ASSERT(ptr && *ptr == 114);
}
}
{
if (auto result = test_std::sync_wait(
test_std::schedule(sch) | test_std::then([] { return std::vector{1, 2, 3, 4}; }) |
test_std::bulk(
test_std::par, 4uz, [](std::size_t i, std::vector<int>& vec) noexcept { vec[i] *= 2; }))) {
auto [vec] = std::move(*result);
ASSERT((vec == std::vector{2, 4, 6, 8}));
}
}
}
} // namespace

Expand Down
Loading