From bfb9393c7d52f20bf7daffeddb34190a5546e00b Mon Sep 17 00:00:00 2001 From: Cra3z Date: Fri, 9 Oct 2026 14:27:20 +0800 Subject: [PATCH 1/2] Fix `bulk` to pass the child sender's results to the callback as lvalue references --- include/beman/execution/detail/bulk.hpp | 12 +++--- .../execution/detail/psched_bulk_sender.hpp | 12 +++--- include/beman/execution/detail/task.hpp | 18 ++++----- .../beman/execution/detail/task_scheduler.hpp | 4 +- tests/beman/execution/exec-bulk.test.cpp | 39 +++++++++++++++++++ .../exec-parallel-scheduler.test.cpp | 20 ++++++++++ 6 files changed, 82 insertions(+), 23 deletions(-) diff --git a/include/beman/execution/detail/bulk.hpp b/include/beman/execution/detail/bulk.hpp index 6d8a97f6..63c2ff13 100644 --- a/include/beman/execution/detail/bulk.hpp +++ b/include/beman/execution/detail/bulk.hpp @@ -62,20 +62,20 @@ struct bulk_traits { // for bulk_chunked static auto invoke(F& fn, Shape shape, Args&... args) noexcept(is_nothrow_invocable) -> void { if (shape > static_cast(0)) [[likely]] { - std::invoke(fn, 0, shape, args...); + ::std::invoke(fn, 0, shape, args...); } } }; template struct bulk_traits { // for bulk_unchunked - static constexpr bool is_invocable = ::std::invocable; + static constexpr bool is_invocable = ::std::invocable; - static constexpr bool is_nothrow_invocable = ::std::is_nothrow_invocable_v; + static constexpr bool is_nothrow_invocable = ::std::is_nothrow_invocable_v; static auto invoke(F& fn, Shape shape, Args&... args) noexcept(is_nothrow_invocable) -> void { for (auto i = static_cast(0); i < shape; ++i) { - std::invoke(fn, i, args...); + ::std::invoke(fn, i, args...); } } }; @@ -242,10 +242,10 @@ struct bulk_t { template static auto wrap_chunked(Fn f) noexcept { - return [f = std::move(f)](Shape begin, Shape end, Args&&... args) noexcept( + return [f = ::std::move(f)](Shape begin, Shape end, Args&&... args) noexcept( ::std::is_nothrow_invocable_v) { while (begin != end) { - std::invoke(f, begin++, args...); + ::std::invoke(f, begin++, args...); } }; } diff --git a/include/beman/execution/detail/psched_bulk_sender.hpp b/include/beman/execution/detail/psched_bulk_sender.hpp index e2dd5fd1..9bd8c9d4 100644 --- a/include/beman/execution/detail/psched_bulk_sender.hpp +++ b/include/beman/execution/detail/psched_bulk_sender.hpp @@ -106,12 +106,12 @@ struct psched_bulk_sender { assert(IsChunked || !is_parallel_policy || end - begin == 1uz); 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](const Args&... args) { + const auto call_fn = [=, this](Args&... args) { if constexpr (IsChunked) { - static_assert(::std::invocable); + static_assert(::std::invocable); fn(first, last, args...); } else { - static_assert(::std::invocable); + static_assert(::std::invocable); for (Shape i = first; i < last; ++i) { fn(i, args...); } @@ -122,7 +122,7 @@ struct psched_bulk_sender { if constexpr (::std::same_as) { ::beman::execution::detail::unreachable(); } else { - ::std::apply(call_fn, ::std::move(tpl)); + ::std::apply(call_fn, tpl); } }, result); @@ -135,7 +135,7 @@ struct psched_bulk_sender { try { ::std::visit( [&](T tpl) { - if constexpr (::std::same_as) { + if constexpr (::std::same_as) { ::beman::execution::detail::unreachable(); } else { ::std::apply(call_set_value, ::std::move(tpl)); @@ -212,7 +212,7 @@ struct psched_bulk_sender { using sub_state_t = ::beman::execution::connect_result_t; state(::std::shared_ptr 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); } diff --git a/include/beman/execution/detail/task.hpp b/include/beman/execution/detail/task.hpp index e839621d..c0ce55d2 100644 --- a/include/beman/execution/detail/task.hpp +++ b/include/beman/execution/detail/task.hpp @@ -141,7 +141,7 @@ struct allocator_support { } template - 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 alloc{}; return allocator_traits::allocate(alloc, size); @@ -313,7 +313,7 @@ class result_type( [](E& error) { if constexpr (::std::same_as<::std::remove_cvref_t, ::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, ::std::error_code>) { throw ::std::system_error{::std::move(error)}; } else { @@ -339,7 +339,7 @@ class result_type> { private: using value_type = ::std::conditional_t<::std::same_as, void_type, Value>; - ::std::variant result; + ::std::variant<::std::monostate, value_type> result; public: template @@ -463,7 +463,7 @@ auto make_sched(const Env& env, const Alloc& alloc) noexcept -> Sched { ::std::make_obj_using_allocator(alloc, ::beman::execution::get_scheduler(env)); }) { return ::std::make_obj_using_allocator(alloc, ::beman::execution::get_scheduler(env)); - } else if constexpr (std::default_initializable) { + } else if constexpr (::std::default_initializable) { return Sched(); } else { static_assert(false, @@ -494,7 +494,7 @@ class state_base : public result_type> { 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(); } @@ -521,7 +521,7 @@ struct state : state_base { template 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(env, do_get_allocator())), propagator(::beman::execution::get_stop_token(env)) {} @@ -570,7 +570,7 @@ class awaiter : public state_base { explicit awaiter(::std::coroutine_handle parent, handle h, const auto& env) noexcept : parent(parent), - handle(std::move(h)), + handle(::std::move(h)), holder(env), scheduler(::beman::execution::detail::task::make_sched(env, do_get_allocator())), propagator(::beman::execution::get_stop_token(env)) {} @@ -586,7 +586,7 @@ class awaiter : public state_base { 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(); } @@ -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(this)); diff --git a/include/beman/execution/detail/task_scheduler.hpp b/include/beman/execution/detail/task_scheduler.hpp index 69b6a865..1ac636ca 100644 --- a/include/beman/execution/detail/task_scheduler.hpp +++ b/include/beman/execution/detail/task_scheduler.hpp @@ -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); } @@ -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_; }; diff --git a/tests/beman/execution/exec-bulk.test.cpp b/tests/beman/execution/exec-bulk.test.cpp index e6c38078..c03ebabf 100644 --- a/tests/beman/execution/exec-bulk.test.cpp +++ b/tests/beman/execution/exec-bulk.test.cpp @@ -82,6 +82,44 @@ auto test_bulk() { } } +auto test_bulk_predecessor_result() { + { + auto sndr = test_std::just(std::make_unique(114)) | + test_std::bulk(test_std::par, 1, [](int i, std::unique_ptr& 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(514)) | + test_std::bulk_chunked(test_std::par, 1, [](int i, int j, std::unique_ptr& 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(114)) | + test_std::bulk_unchunked(test_std::par, 1, [](int i, std::unique_ptr& 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& 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); @@ -560,6 +598,7 @@ TEST(exec_bulk) { try { test_bulk(); + test_bulk_predecessor_result(); test_bulk_noexcept(); test_bulk_pipeable(); test_bulk_chunked(); diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp index beb4daf5..288b0d97 100644 --- a/tests/beman/execution/exec-parallel-scheduler.test.cpp +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -169,6 +169,26 @@ auto test_parallel_scheduler_schedule() -> void { } } } + { + auto sndr = test_std::schedule(sch) | test_std::then([] { return std::make_unique(114); }) | + test_std::bulk(test_std::par, 1, [](int i, std::unique_ptr& 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& vec) noexcept { vec[i] *= 2; }))) { + auto [vec] = std::move(*result); + ASSERT((vec == std::vector{2, 4, 6, 8})); + } + } } } // namespace From 98ae23dedf889830059627c3efcf316144786188 Mon Sep 17 00:00:00 2001 From: Cra3z Date: Fri, 9 Oct 2026 14:56:12 +0800 Subject: [PATCH 2/2] Format code --- tests/beman/execution/exec-bulk.test.cpp | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/tests/beman/execution/exec-bulk.test.cpp b/tests/beman/execution/exec-bulk.test.cpp index c03ebabf..626f1f2d 100644 --- a/tests/beman/execution/exec-bulk.test.cpp +++ b/tests/beman/execution/exec-bulk.test.cpp @@ -84,29 +84,29 @@ auto test_bulk() { auto test_bulk_predecessor_result() { { - auto sndr = test_std::just(std::make_unique(114)) | - test_std::bulk(test_std::par, 1, [](int i, std::unique_ptr& ptr) noexcept { + auto sndr = test_std::just(std::make_unique(114)) | + test_std::bulk(test_std::par, 1, [](int i, std::unique_ptr& 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(514)) | - test_std::bulk_chunked(test_std::par, 1, [](int i, int j, std::unique_ptr& ptr) noexcept { + auto sndr = test_std::just(std::make_unique(514)) | + test_std::bulk_chunked(test_std::par, 1, [](int i, int j, std::unique_ptr& 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(114)) | - test_std::bulk_unchunked(test_std::par, 1, [](int i, std::unique_ptr& ptr) noexcept { + auto sndr = test_std::just(std::make_unique(114)) | + test_std::bulk_unchunked(test_std::par, 1, [](int i, std::unique_ptr& ptr) noexcept { ASSERT(ptr && *ptr == 114); ASSERT(i == 0); - }); + }); auto [ptr] = test_std::sync_wait(std::move(sndr)).value(); ASSERT(ptr && *ptr == 114); }