diff --git a/include/exec/libdispatch_queue.hpp b/include/exec/libdispatch_queue.hpp index 24bf27221..5f717cacf 100644 --- a/include/exec/libdispatch_queue.hpp +++ b/include/exec/libdispatch_queue.hpp @@ -37,6 +37,7 @@ import stdexec; # if !STDEXEC_USE_MODULES() # include +# include # include # endif @@ -54,16 +55,16 @@ namespace experimental::execution template struct operation; - template + template struct bulk_sender; template struct bulk_shared_state; template - requires STDEXEC::__callable &...> + requires STDEXEC::__callable &...> using bulk_non_throwing_t = - STDEXEC::__mbool &...> + STDEXEC::__mbool &...> && STDEXEC::__nothrow_decay_copyable>; template @@ -74,16 +75,36 @@ namespace experimental::execution struct transform_bulk { - template - auto operator()(STDEXEC::bulk_t, Data &&data, Sender &&sndr) + template Tag, + class Data, + class Sender> + auto operator()(Tag, Data &&data, Sender &&sndr) { auto [pol, shape, fun] = static_cast(data); - // TODO: handle non-par execution policies - return bulk_sender, decltype(shape), decltype(fun)>{ - queue_, - static_cast(sndr), - shape, - std::move(fun)}; + using policy_type = STDEXEC::__decay_t; + constexpr bool parallelize = + STDEXEC::__same_as + || STDEXEC::__same_as; + + if constexpr (STDEXEC::__same_as) + { + return bulk_sender, decltype(shape), decltype(fun), true>{ + queue_, + static_cast(sndr), + shape, + std::move(fun), + parallelize}; + } + else + { + using fun_t = STDEXEC::__bulk::__as_bulk_chunked_fn; + return bulk_sender, decltype(shape), fun_t, false>{ + queue_, + static_cast(sndr), + shape, + fun_t{std::move(fun)}, + parallelize}; + } } libdispatch_queue &queue_; @@ -101,7 +122,9 @@ namespace experimental::execution struct domain { // transform the generic bulk sender into a parallel libdispatch bulk sender - template Sender, class Env> + template + requires sender_for + || sender_for auto transform_sender(STDEXEC::set_value_t, Sender &&sndr, Env const &env) const noexcept { if constexpr (STDEXEC::__completes_on) @@ -117,7 +140,7 @@ namespace experimental::execution return STDEXEC::__not_a_sender< STDEXEC::_WHAT_(CANNOT_DISPATCH_THE_BULK_ALGORITHM_TO_THE_LIBDISPATCH_SCHEDULER), STDEXEC::_WHY_(BECAUSE_THERE_IS_NO_LIBDISPATCH_SCHEDULER_IN_THE_ENVIRONMENT), - STDEXEC::_WHERE_(STDEXEC::_IN_ALGORITHM_, STDEXEC::bulk_t), + STDEXEC::_WHERE_(STDEXEC::_IN_ALGORITHM_, STDEXEC::tag_of_t), STDEXEC::_TO_FIX_THIS_ERROR_( ADD_A_CONTINUES_ON_TRANSITION_TO_THE_LIBDISPATCH_SCHEDULER_BEFORE_THE_BULK_ALGORITHM), STDEXEC::_WITH_PRETTY_SENDER_, @@ -257,7 +280,7 @@ namespace experimental::execution ////////////////////////////////////////////////////////////////////////////////////////////////// // What follows is the implementation for parallel bulk execution on // libdispatch queue. - template + template struct bulk_sender { using sender_concept = STDEXEC::sender_tag; @@ -278,6 +301,7 @@ namespace experimental::execution return bulk_op_state_t{self.queue_, self.shape_, self.fun_, + self.parallelize_, std::forward(self).sndr_, std::forward(rcvr)}; } @@ -291,16 +315,14 @@ namespace experimental::execution STDEXEC::get_completion_signatures<__copy_cvref_t, Env...>(), []() { + using bulk_tag_t = std::conditional_t; using value_sig_t = set_value_t(__decay_t...); - using arg_pack_t = __tuple &...>; - // using arg_pack_t = __if_c<__same_as<_AlgoTag, bulk_chunked_t>, - // __tuple, - // __tuple>; + using arg_pack_t = __tuple &...>; if constexpr (!__decay_copyable) { return exec::throw_compile_time_error< _WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), - _WHERE_(_IN_ALGORITHM_, bulk_t), + _WHERE_(_IN_ALGORITHM_, bulk_tag_t), _WITH_ARGUMENTS_(Args...), _WITH_PRETTY_SENDER_<__copy_cvref_t>, _WITH_ENVIRONMENT_(Env...)>(); @@ -318,7 +340,7 @@ namespace experimental::execution { return STDEXEC::__throw_compile_time_error< _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), - _WHERE_(_IN_ALGORITHM_, bulk_t), + _WHERE_(_IN_ALGORITHM_, bulk_tag_t), _WITH_FUNCTION_(Fun &), __mapply<__qf<_WITH_ARGUMENTS_>, arg_pack_t>>(); } @@ -335,6 +357,7 @@ namespace experimental::execution Sender sndr_; Shape shape_; Fun fun_; + bool parallelize_; }; template @@ -355,9 +378,19 @@ namespace experimental::execution auto task_id = static_cast(t)->task_id_; auto total_tasks = static_cast(sh_state.num_tasks()); - auto computation = [&sh_state, task_id](auto &...args) + auto computation = [&sh_state, task_id, total_tasks](auto &...args) { - sh_state.fun_(task_id, args...); + static_cast(total_tasks); + if (!sh_state.parallelize_) + { + // There should only be a single task + STDEXEC_ASSERT(task_id == 0 && total_tasks == 1); + sh_state.fun_(static_cast(0), sh_state.shape_, args...); + } + else + { + sh_state.fun_(task_id, task_id + static_cast(1), args...); + } }; auto completion = [&](auto &...args) @@ -419,16 +452,17 @@ namespace experimental::execution STDEXEC::__q, STDEXEC::__q>; - bulk_shared_state(Receiver rcvr, Shape shape, Fun fun) + bulk_shared_state(Receiver rcvr, Shape shape, Fun fun, bool parallelize) : rcvr_{std::move(rcvr)} , shape_{shape} , fun_{fun} + , parallelize_{parallelize} , task_with_exception_{static_cast(num_tasks())} {} - Shape num_tasks() const + Shape num_tasks() const noexcept { - return shape_; + return parallelize_ ? shape_ : (std::min) (shape_, static_cast(1)); } template @@ -443,6 +477,7 @@ namespace experimental::execution Receiver rcvr_; Shape shape_; Fun fun_; + bool parallelize_; STDEXEC::__std::atomic finished_tasks_{0}; STDEXEC::__std::atomic task_with_exception_{0}; @@ -459,9 +494,10 @@ namespace experimental::execution void enqueue() noexcept { - using bulk_task = shared_state::bulk_task; - shared_state_.tasks_.reserve(static_cast(shared_state_.shape_)); - for (Shape i{}; i != shared_state_.shape_; ++i) + using bulk_task = shared_state::bulk_task; + auto const total_tasks = shared_state_.num_tasks(); + shared_state_.tasks_.reserve(static_cast(total_tasks)); + for (Shape i{}; i != total_tasks; ++i) { shared_state_.tasks_.push_back(bulk_task(&shared_state_, i)); queue_.submit(&(shared_state_.tasks_.back())); @@ -535,8 +571,13 @@ namespace experimental::execution using shared_state = bulk_shared_state; using inner_op_state = STDEXEC::connect_result_t; - bulk_op_state(libdispatch_queue &queue, Shape shape, Fun fun, CvSender &&sndr, Receiver rcvr) - : shared_state_(std::move(rcvr), shape, fun) + bulk_op_state(libdispatch_queue &queue, + Shape shape, + Fun fun, + bool parallelize, + CvSender &&sndr, + Receiver rcvr) + : shared_state_(std::move(rcvr), shape, fun, parallelize) , inner_op_{ STDEXEC::connect(static_cast(sndr), bulk_rcvr{shared_state_, queue})} {} diff --git a/test/exec/test_libdispatch.cpp b/test/exec/test_libdispatch.cpp index 465411407..7251a57d6 100644 --- a/test/exec/test_libdispatch.cpp +++ b/test/exec/test_libdispatch.cpp @@ -19,6 +19,7 @@ #include "test_common/catch2.hpp" #include "test_common/type_helpers.hpp" +#include #include #include #include @@ -108,6 +109,92 @@ namespace CHECK(res == 30); } + TEST_CASE("libdispatch queue bulk_chunked uses one task per index for parallel policies") + { + exec::libdispatch_queue queue; + auto sch = queue.get_scheduler(); + + std::vector visited(5, 0); + std::atomic chunks{0}; + + auto sender = STDEXEC::schedule(sch) + | STDEXEC::bulk_chunked(STDEXEC::par, + 5, + [&](int begin, int end) + { + ++chunks; + for (; begin != end; ++begin) + visited[begin] = 1; + }); + + REQUIRE(STDEXEC::sync_wait(std::move(sender)).has_value()); + + CHECK(chunks.load() == 5); + CHECK(visited == std::vector{1, 1, 1, 1, 1}); + } + + TEST_CASE("libdispatch queue runs a non-parallel bulk_chunked in a single task") + { + exec::libdispatch_queue queue; + auto sch = queue.get_scheduler(); + + std::vector bounds; + + auto sender = STDEXEC::schedule(sch) + | STDEXEC::bulk_chunked(STDEXEC::seq, + 5, + [&](int begin, int end) + { + bounds.push_back(begin); + bounds.push_back(end); + }); + + REQUIRE(STDEXEC::sync_wait(std::move(sender)).has_value()); + + // `seq` forbids splitting the index space, so a single chunk covers all of it + CHECK(bounds == std::vector{0, 5}); + } + + TEST_CASE("libdispatch queue bulk_unchunked should call callback function with every index") + { + exec::libdispatch_queue queue; + auto sch = queue.get_scheduler(); + + std::vector data{1, 2, 3, 4, 5}; + auto size = data.size(); + auto expensive_computation = [](auto i, auto &data) + { + data[i] = 2 * data[i]; + }; + auto add = [](auto const &data) + { + return std::accumulate(std::begin(data), std::end(data), 0); + }; + auto sender = STDEXEC::just(std::move(data)) | STDEXEC::continues_on(sch) + | STDEXEC::bulk_unchunked(STDEXEC::par, size, expensive_computation) + | STDEXEC::then(add); + + auto [res] = STDEXEC::sync_wait(sender).value(); + CHECK(res == 30); + } + + TEST_CASE("libdispatch queue runs a non-parallel bulk_unchunked in a single task") + { + exec::libdispatch_queue queue; + auto sch = queue.get_scheduler(); + + std::vector indices; + + auto sender = STDEXEC::schedule(sch) + | STDEXEC::bulk_unchunked(STDEXEC::seq, + 4, + [&](int idx) { indices.push_back(idx); }); + + REQUIRE(STDEXEC::sync_wait(std::move(sender)).has_value()); + + CHECK(indices == std::vector{0, 1, 2, 3}); + } + #if !STDEXEC_NO_STDCPP_EXCEPTIONS() TEST_CASE("libdispatch bulk should handle exceptions gracefully") { @@ -241,10 +328,11 @@ namespace TEST_CASE("libdispatch bulk connects an lvalue child sender as an lvalue") { exec::libdispatch_queue queue; - auto fun = [](int, int &) noexcept {}; - using sender_t = exec::__libdispatch::bulk_sender; + auto fun = [](int, int, int &) noexcept {}; + using sender_t = + exec::__libdispatch::bulk_sender; - sender_t sender{queue, lvalue_connect_sender{}, 0, std::move(fun)}; + sender_t sender{queue, lvalue_connect_sender{}, 0, std::move(fun), true}; auto result = STDEXEC::sync_wait(sender); REQUIRE(result.has_value());