diff --git a/include/exec/fork_join.hpp b/include/exec/fork_join.hpp index 921c84d86..c23a5c8cc 100644 --- a/include/exec/fork_join.hpp +++ b/include/exec/fork_join.hpp @@ -39,8 +39,24 @@ namespace experimental::execution } }; + template ::value> + struct _cache_completions + { + using type = Completions; + }; + + template + struct _cache_completions + { + using type = + STDEXEC::__concat_completion_signatures_t; + }; + + // Account that decay-copying into the cache may throw. template - using _variant_t = STDEXEC::__mapply_q; + using _variant_t = STDEXEC::__mapply_q::type>; template struct _env_t @@ -281,16 +297,38 @@ namespace experimental::execution struct fork_join_t { + /// No closure given. + template + STDEXEC_ATTRIBUTE(host, device) + constexpr auto operator()(Sndr&& sndr) const noexcept(STDEXEC::__nothrow_decay_copyable) + { + return static_cast(sndr); + } + + /// Unary closure. + template + requires(!STDEXEC::sender) + STDEXEC_ATTRIBUTE(host, device) + constexpr auto operator()(Sndr&& sndr, Closure&& clsr) const + noexcept(STDEXEC::__nothrow_callable) + { + return static_cast(clsr)(static_cast(sndr)); + } + + /// One sender and multiple closures. template + requires(sizeof...(Closures) > 1) STDEXEC_ATTRIBUTE(host, device) - constexpr auto operator()(Sndr&& sndr, Closures&&... closures) const // - -> STDEXEC::__well_formed_sender auto + constexpr auto operator()(Sndr&& sndr, Closures&&... closures) const + noexcept(STDEXEC::__nothrow_decay_copyable) + -> STDEXEC::__well_formed_sender auto { return STDEXEC::__sexpr{fork_join_t(), STDEXEC::__tuple{static_cast(closures)...}, static_cast(sndr)}; } + /// One or more closures. template requires((!STDEXEC::sender) && ...) STDEXEC_ATTRIBUTE(host, device) diff --git a/include/stdexec/__detail/__then.hpp b/include/stdexec/__detail/__then.hpp index 1a86eece8..9a7ddbff1 100644 --- a/include/stdexec/__detail/__then.hpp +++ b/include/stdexec/__detail/__then.hpp @@ -239,7 +239,7 @@ namespace STDEXEC //! sender `then(sndr, std::move(__fun))`. template <__movable_value _Fun> STDEXEC_ATTRIBUTE(always_inline, host, device) - constexpr auto operator()(_Fun __fun) const + constexpr auto operator()(_Fun __fun) const noexcept(__nothrow_decay_copyable<_Fun>) { return __closure(*this, static_cast<_Fun&&>(__fun)); } diff --git a/test/exec/test_fork_join.cpp b/test/exec/test_fork_join.cpp index a2d6d41b7..dc23cc399 100644 --- a/test/exec/test_fork_join.cpp +++ b/test/exec/test_fork_join.cpp @@ -43,6 +43,30 @@ namespace completion_signatures>); } + TEST_CASE("fork_join coalesces empty and unary calls", "[adaptors][fork_join]") + { + /// Empty (no closure given). + STDEXEC::sender auto empty = exec::fork_join(STDEXEC::just()); + using empty_t = decltype(empty); + STATIC_REQUIRE(std::same_as); + STATIC_REQUIRE(!exec::sender_for); + STATIC_REQUIRE(noexcept(exec::fork_join(STDEXEC::just()))); + + auto then = STDEXEC::then([]() noexcept {}); + + /// Unary closure. + STDEXEC::sender auto unary = exec::fork_join(STDEXEC::just(), then); + using unary_t = decltype(unary); + STATIC_REQUIRE(std::same_as); + STATIC_REQUIRE(!exec::sender_for); + STATIC_REQUIRE(noexcept(exec::fork_join(STDEXEC::just(), then))); + + /// Multiple closures. + STDEXEC::sender auto multiple = STDEXEC::just() | exec::fork_join(then, then); + STATIC_REQUIRE(exec::sender_for); + STATIC_REQUIRE(noexcept(exec::fork_join(STDEXEC::just(), then, then))); + } + struct ForwardingThen { template @@ -69,6 +93,35 @@ namespace }; #endif + //! Count how many copies and moves are performed. + struct counter + { + inline static std::atomic copy_constructions{0}; + inline static std::atomic copy_assignments{0}; + inline static std::atomic move_constructions{0}; + inline static std::atomic move_assignments{0}; + + counter() = default; + counter(counter const &) noexcept + { + copy_constructions.fetch_add(1, std::memory_order_relaxed); + } + counter &operator=(counter const &) noexcept + { + copy_assignments.fetch_add(1, std::memory_order_relaxed); + return *this; + } + counter(counter &&) noexcept + { + move_constructions.fetch_add(1, std::memory_order_relaxed); + } + counter &operator=(counter &&) noexcept + { + move_assignments.fetch_add(1, std::memory_order_relaxed); + return *this; + } + }; + template struct identifiable_domain : public STDEXEC::default_domain {}; @@ -140,18 +193,43 @@ namespace #if !STDEXEC_NO_STDCPP_EXCEPTIONS() TEST_CASE("fork_join reports failures while caching results", "[adaptors][fork_join]") { + std::atomic witness{0}; + auto sndr = exec::fork_join(exec::just_from( [](auto sink) noexcept { static throwing_copy value; return sink(value); }), - then([](throwing_copy const &) noexcept {})); + then([&witness](throwing_copy const &) noexcept { ++witness; }), + then([&witness](throwing_copy const &) noexcept { ++witness; })); CHECK_THROWS_AS(sync_wait(std::move(sndr)), int); + + CHECK(witness == 0); } #endif + TEST_CASE("fork_join caches without copying when results are movable and replays by reference to " + "children", + "[adaptors][fork_join]") + { + std::atomic witness{0}; + + auto sndr = exec::fork_join(exec::just_from([](auto sink) noexcept { return sink(counter{}); }), + STDEXEC::then([&witness](counter const &) noexcept { ++witness; }), + STDEXEC::then([&witness](counter const &) noexcept { ++witness; })); + + STDEXEC::sync_wait(std::move(sndr)); + + CHECK(counter::copy_constructions == 0); + CHECK(counter::copy_assignments == 0); + CHECK(counter::move_constructions == 1); + CHECK(counter::move_assignments == 0); + + CHECK(witness == 2); + } + TEST_CASE("fork_join can be nested", "[adaptors][fork_join]") { std::atomic witness = 0; diff --git a/test/stdexec/algos/adaptors/test_then.cpp b/test/stdexec/algos/adaptors/test_then.cpp index 88e85e0c3..1292d522d 100644 --- a/test/stdexec/algos/adaptors/test_then.cpp +++ b/test/stdexec/algos/adaptors/test_then.cpp @@ -179,6 +179,17 @@ namespace | ex::then([] { return std::string{"hello"}; })); } + TEST_CASE("then noexceptness", "[adaptors][then]") + { + auto func = [](){}; + + STATIC_REQUIRE(noexcept(ex::then(func))); + + STATIC_REQUIRE(noexcept(STDEXEC::just() | STDEXEC::then(func))); + + STATIC_REQUIRE(noexcept(STDEXEC::then(STDEXEC::just(), func))); + } + TEST_CASE("then keeps error_types from input sender", "[adaptors][then]") { inline_scheduler sched1{};