diff --git a/include/exec/sequence/merge_each.hpp b/include/exec/sequence/merge_each.hpp index 923b28ddd..397e312aa 100644 --- a/include/exec/sequence/merge_each.hpp +++ b/include/exec/sequence/merge_each.hpp @@ -383,16 +383,12 @@ namespace experimental::execution } void set_break() noexcept { - switch (__completion_.exchange(__completion_t::__stopped)) + auto __expected = __completion_t::__started; + if (__completion_.compare_exchange_strong(__expected, __completion_t::__stopped)) { - case __completion_t::__started: - // We must request stop. When the previous state is __error or __stopped, then stop has - // already been requested. + // We transitioned from started to stopped, so we must request stop. When the state is + // already error or stopped, stop has already been requested. __nested_stop_.request_stop(); - break; - case __completion_t::__stopped: - [[fallthrough]]; // We're already in the "stopped" state. Ignore the break. - case __completion_t::__error:; // We're already in the "error" state. Ignore the break. } } @@ -455,8 +451,9 @@ namespace experimental::execution } void error_complete() noexcept override { - // do not double report error - STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_)); + // The error has been delivered as an item. Complete the sequence so the consumer can + // publish it, unless the operation was independently stopped. + exec::__set_value_unless_stopped(static_cast<_Receiver&&>(__rcvr_)); } void complete_if_none_active() noexcept diff --git a/test/exec/sequence/test_merge_each.cpp b/test/exec/sequence/test_merge_each.cpp index 2160a49c4..090c1262d 100644 --- a/test/exec/sequence/test_merge_each.cpp +++ b/test/exec/sequence/test_merge_each.cpp @@ -33,6 +33,7 @@ #include #include +#include using namespace std::chrono_literals; using namespace exec; @@ -227,6 +228,86 @@ namespace auto op = subscribe(std::move(merged), null_receiver{}); } + template + struct error_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_error(static_cast(receiver_), static_cast(error_)); + } + + Receiver receiver_; + Error error_; + }; + + template + auto subscribe(Receiver receiver) && noexcept -> operation + { + return {static_cast(receiver), static_cast(error_)}; + } + + Error error_; + }; + + struct error_state + { + std::optional error_{}; + bool completed_ = false; + bool stopped_ = false; + }; + + struct record_error_receiver + { + using receiver_concept = ex::receiver_tag; + + template + auto set_next(Item&& item) & + { + auto record_error = [state = state_](Error&& error) noexcept + { + if constexpr (std::same_as, int>) + { + state->error_ = static_cast(error); + } + }; + return ex::upon_stopped(ex::upon_error(static_cast(item), record_error), + []() noexcept {}); + } + + void set_value() noexcept + { + state_->completed_ = true; + } + + template + void set_error(Error&& error) noexcept + { + if constexpr (std::same_as, int>) + { + state_->error_ = static_cast(error); + } + } + + void set_stopped() noexcept + { + state_->stopped_ = true; + } + + auto get_env() const noexcept -> ex::env<> + { + return {}; + } + + error_state* state_; + }; + TEST_CASE("merge_each - merge two sequence senders of no elements", "[sequence_senders][merge_each][empty_sequence]") { @@ -368,31 +449,46 @@ namespace CHECK(v.has_value() == true); } -// TODO - fix problem with stopping -#if 0 - TEST_CASE( - "merge_each - merge_each sender stops when a nested sequence fails", - "[sequence_senders][static_thread_pool][merge_each][merge][iterate]") { + TEST_CASE("merge_each - preserves errors from nested value senders", + "[sequence_senders][merge_each]") + { + error_state state{}; + auto merged = merge_each(ex::just(ex::just_error(42))); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); - auto sequences = merge( - log_start(range(100, 120), "range 100-120"), - ex::just(emits_error(std::runtime_error{"failed sequence "})), - log_start(range(200, 220), "range 200-220") - ); + ex::start(op); - [[maybe_unused]] auto merged = merge_each(std::move(sequences)); + CHECK(state.error_ == 42); + CHECK(state.completed_); + CHECK_FALSE(state.stopped_); + } - int count = 0; + TEST_CASE("merge_each - preserves errors from nested sequences", "[sequence_senders][merge_each]") + { + error_state state{}; + auto nested_sequence = error_sequence{42}; + auto merged = merge_each(ex::just(std::move(nested_sequence))); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); + + ex::start(op); + + CHECK(state.error_ == 42); + CHECK(state.completed_); + CHECK_FALSE(state.stopped_); + } + + TEST_CASE("merge_each - preserves errors from the outer sequence", + "[sequence_senders][merge_each]") + { + error_state state{}; + auto merged = merge_each(error_sequence{42}); + auto op = subscribe(std::move(merged), record_error_receiver{&state}); - auto v = ex::sync_wait(ignore_all_values(merged | then_each([&count](int x){ - ++count; - UNSCOPED_INFO("item: " << x - << ", on thread id: " << std::this_thread::get_id()); - }))); + ex::start(op); - CHECK(count == 20); - CHECK(v.has_value() == false); + CHECK(state.error_ == 42); + CHECK_FALSE(state.completed_); + CHECK_FALSE(state.stopped_); } -#endif // 0 } // namespace