diff --git a/include/exec/sequence/merge_each.hpp b/include/exec/sequence/merge_each.hpp index 8700b67ef..1db401404 100644 --- a/include/exec/sequence/merge_each.hpp +++ b/include/exec/sequence/merge_each.hpp @@ -1001,21 +1001,34 @@ namespace experimental::execution if constexpr (!__is_nothrow) { __op_->store_error(std::current_exception()); - __op_->nested_sequence_break(); + __next_seq_op_->nested_sequence_break(); } } } + // The sender of a nested sequence completed without producing a nested + // sequence. The next receiver must still be completed, or else the input + // sequence would never learn that this item is finished. template void set_error(_Error&& __error) noexcept { __op_->store_error(static_cast<_Error&&>(__error)); - __op_->nested_sequence_break(); + __next_seq_op_->nested_sequence_break(); } void set_stopped() noexcept { - __op_->nested_sequence_complete(); + if (__op_->__nested_stop_.stop_requested()) + { + // Stop has been requested, so ask the input sequence to stop + // producing nested sequences. + __next_seq_op_->nested_sequence_break(); + } + else + { + // Only this sender was stopped; skip it and keep merging the rest. + __next_seq_op_->nested_sequence_complete(); + } } using __env_t = typename _OperationBase::__nested_stop_env_t; diff --git a/test/exec/sequence/test_merge_each.cpp b/test/exec/sequence/test_merge_each.cpp index faeabcec7..55ab405b4 100644 --- a/test/exec/sequence/test_merge_each.cpp +++ b/test/exec/sequence/test_merge_each.cpp @@ -35,6 +35,7 @@ #include #include #include +#include using namespace std::chrono_literals; using namespace exec; @@ -488,7 +489,134 @@ namespace CHECK_FALSE(state.stopped_); } + // A sender of a nested sequence that can also complete without producing one + template + using maybe_nested_sequence_t = variant_sender; + + TEST_CASE("merge_each - completes when the sender of a nested sequence fails", + "[sequence_senders][merge_each]") + { + using nested_sequence_sender_t = maybe_nested_sequence_t; + + error_state state{}; + int count = 0; + auto merged = merge_each(range(0, 3) + | exec::transform_each(ex::let_value( + [](int i) noexcept -> nested_sequence_sender_t + { + if (i == 1) + { + return ex::just_error(42); + } + return ex::just(ex::just(i)); + }))) + | then_each([&count](int) noexcept { ++count; }); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); + + ex::start(op); + + CHECK(count == 1); + CHECK(state.nested_error_ == 42); + CHECK(state.completed_); + CHECK_FALSE(state.stopped_); + } + + TEST_CASE("merge_each - continues when the sender of a nested sequence is stopped", + "[sequence_senders][merge_each]") + { + using nested_sequence_sender_t = maybe_nested_sequence_t; + + error_state state{}; + int count = 0; + auto merged = merge_each(range(0, 3) + | exec::transform_each(ex::let_value( + [](int i) noexcept -> nested_sequence_sender_t + { + if (i == 1) + { + return ex::just_stopped(); + } + return ex::just(ex::just(i)); + }))) + | then_each([&count](int) noexcept { ++count; }); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); + + ex::start(op); + + CHECK(count == 2); + CHECK_FALSE(state.nested_error_.has_value()); + CHECK(state.completed_); + CHECK_FALSE(state.stopped_); + } + + TEST_CASE("merge_each - completes with stopped when stop is requested between nested sequences", + "[sequence_senders][merge_each]") + { + ex::inplace_stop_source stop_source; + int count = 0; + bool executed = false; + auto merged = merge_each(range(0, 3) | then_each([](int) noexcept { return range(0, 3); })) + | then_each( + [&](int) noexcept + { + if (++count == 4) + { + stop_source.request_stop(); + } + }) + | ignore_all_values() | with_stop_token_from(stop_source); + auto op = ex::connect(std::move(merged), expect_stopped_receiver_ex{executed}); + + ex::start(op); + + CHECK(count == 4); + CHECK(executed); + } + #if !STDEXEC_NO_STDCPP_EXCEPTIONS() + struct throwing_subscribe_sequence + { + using sender_concept = sequence_sender_tag; + using item_types = exec::item_types<>; + using completion_signatures = ex::completion_signatures; + + template + struct operation + { + void start() & noexcept + { + ex::set_value(static_cast(receiver_)); + } + + Receiver receiver_; + }; + + template + auto subscribe(Receiver receiver) && -> operation + { + if (throws_) + { + throw std::runtime_error("subscribe failed"); + } + return {static_cast(receiver)}; + } + + bool throws_ = true; + }; + + TEST_CASE("merge_each - completes when subscribing to a nested sequence throws", + "[sequence_senders][merge_each]") + { + error_state state{}; + auto merged = merge_each(ex::just(throwing_subscribe_sequence{})); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); + + ex::start(op); + + CHECK(state.completed_); + CHECK_FALSE(state.stopped_); + } + TEST_CASE("merge_each - preserves errors from the outer sequence", "[sequence_senders][merge_each]") {