Skip to content

Commit a23aa7a

Browse files
MaxFreedomPollardclaude
MaxFreedomPollard
and
claude
committed
Fix when_all and when_any completing twice under a racing stop request
The stop callbacks that when_all (since #1718) and exec::when_any (since #2115) register on the receiver's stop token increment the count of outstanding children, request stop, and then arrive. If the last child arrives on another thread while such a callback is running, the count is already zero, so the callback's arrival completes the operation a second time. With inplace_stop_source the thread of the last child usually spins forever in ~inplace_stop_callback instead, waiting for a callback that the other thread has already destroyed. The extra count only worked around inplace_stop_source not tolerating its destruction from inside its own request_stop() (#1889), which #2303 fixed. Make both callbacks only forward the request to the internal stop source, like on-stop-request in [exec.when.all]. For when_all this also undoes the rest of the workaround from #1851: it completes with set_stopped only when a child does, and it no longer records values once a child has failed or been stopped. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 8398d5a commit a23aa7a

5 files changed

Lines changed: 342 additions & 75 deletions

File tree

‎include/exec/when_any.hpp‎

Lines changed: 5 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -83,27 +83,8 @@ namespace experimental::execution
8383

8484
STDEXEC_IMMOVABLE(__opstate_base);
8585

86-
struct __forward_stop_request
87-
{
88-
constexpr void operator()() const noexcept
89-
{
90-
// Temporarily increment the count to avoid concurrent/recursive
91-
// arrivals to pull the rug under our feet. Relaxed memory order is
92-
// fine here.
93-
__op_->__count_.fetch_add(1, __std::memory_order_relaxed);
94-
95-
__op_->__stop_source_.request_stop();
96-
97-
// Arrive in order to decrement the count again and complete if
98-
// needed.
99-
__op_->__arrive();
100-
}
101-
102-
__opstate_base* __op_;
103-
};
104-
10586
using __on_stop =
106-
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>&>, __forward_stop_request>;
87+
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>&>, __forward_stop_request<>>;
10788

10889
inplace_stop_source __stop_source_{};
10990
std::optional<__on_stop> __on_stop_{};
@@ -227,8 +208,11 @@ namespace experimental::execution
227208

228209
void start() & noexcept
229210
{
211+
// Only forward the request to the children. The callback can still be
212+
// running on another thread after the last child has arrived, so it
213+
// must not arrive itself.
230214
this->__on_stop_.emplace(get_stop_token(get_env(this->__rcvr_)),
231-
typename __op_base_t::__forward_stop_request{this});
215+
__forward_stop_request{this->__stop_source_});
232216
if (this->__stop_source_.stop_requested())
233217
{
234218
STDEXEC::set_stopped(static_cast<_Receiver&&>(this->__rcvr_));

‎include/stdexec/__detail/__when_all.hpp‎

Lines changed: 34 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -561,35 +561,6 @@ namespace STDEXEC
561561
struct _INVALID_ARGUMENTS_TO_WHEN_ALL_
562562
{};
563563

564-
template <class _State>
565-
struct __forward_stop_request
566-
{
567-
constexpr void operator()() const noexcept
568-
{
569-
// Temporarily increment the count to avoid concurrent/recursive
570-
// arrivals to pull the rug under our feet. Relaxed memory order is fine
571-
// here.
572-
__state_->__count_.fetch_add(1, __std::memory_order_relaxed);
573-
574-
__state_t __expected = __started;
575-
// Transition to the "stopped" state if and only if we're in the
576-
// "started" state. (If this fails, it's because we're in an error
577-
// state, which trumps cancellation.)
578-
if (__state_->__state_.compare_exchange_strong(__expected, __stopped))
579-
{
580-
if constexpr (_State::__uses_stop_source)
581-
{
582-
__state_->__stop_source_.request_stop();
583-
}
584-
}
585-
586-
// Arrive in order to decrement the count again and complete if needed.
587-
__state_->__arrive();
588-
}
589-
590-
_State* __state_;
591-
};
592-
593564
template <class _ErrorsVariant,
594565
class _ValuesTuple,
595566
class _Receiver,
@@ -600,7 +571,7 @@ namespace STDEXEC
600571
using __receiver_t = _Receiver;
601572
static constexpr bool __uses_stop_source = _UsesStopSource;
602573
using __stop_callback_t =
603-
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>>, __forward_stop_request<__state>>;
574+
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>>, __forward_stop_request<>>;
604575
using __stop_source_t = __if_c<_UsesStopSource, inplace_stop_source, __empty>;
605576
using __on_stop_t = __if_c<_UsesStopSource, __optional<__stop_callback_t>, __empty>;
606577

@@ -622,16 +593,6 @@ namespace STDEXEC
622593
// All child operations have completed and arrived at the barrier.
623594
switch (__state_.load(__std::memory_order_relaxed))
624595
{
625-
case __stopped:
626-
if constexpr (_SendsStopped)
627-
{
628-
STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_));
629-
break;
630-
}
631-
// This is reachable because the stop callback sets __stopped whether
632-
// or not any child can send set_stopped, therefore we handle this the
633-
// same as __started
634-
[[fallthrough]];
635596
case __started:
636597
if constexpr (!__std::same_as<_ValuesTuple, __ignore>)
637598
{
@@ -647,6 +608,17 @@ namespace STDEXEC
647608
static_cast<_ErrorsVariant&&>(__errors_));
648609
}
649610
break;
611+
case __stopped:
612+
// Only a child completing with set_stopped puts us in this state.
613+
if constexpr (_SendsStopped)
614+
{
615+
STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_));
616+
}
617+
else
618+
{
619+
STDEXEC_UNREACHABLE();
620+
}
621+
break;
650622
default:;
651623
}
652624
}
@@ -797,8 +769,11 @@ namespace STDEXEC
797769
{
798770
if constexpr (_State::__uses_stop_source)
799771
{
772+
// Only forward the request to the children. The callback can still
773+
// be running on another thread after the last child has arrived, so
774+
// it must not arrive or change __state_ itself.
800775
__state.__on_stop_.emplace(get_stop_token(STDEXEC::get_env(__state.__rcvr_)),
801-
__forward_stop_request<_State>{&__state});
776+
__forward_stop_request{__state.__stop_source_});
802777
}
803778
(STDEXEC::start(__child_ops), ...);
804779
};
@@ -864,24 +839,29 @@ namespace STDEXEC
864839
}
865840
else if constexpr (!__same_as<_ValuesTuple, __ignore>)
866841
{
867-
auto& __opt_values = STDEXEC::__get<_Index::value>(__state.__values_);
868-
using _Tuple = __decayed_tuple<_Args...>;
869-
static_assert(__same_as<decltype(*__opt_values), _Tuple&>,
870-
"One of the senders in this when_all() is fibbing about what types it "
871-
"sends");
872-
if constexpr ((__nothrow_decay_copyable<_Args> && ...))
842+
// We only need to bother recording the completion values if we're
843+
// not already in the "error" or "stopped" state.
844+
if (__state.__state_.load() == __started)
873845
{
874-
__opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...});
875-
}
876-
else
877-
{
878-
STDEXEC_TRY
846+
auto& __opt_values = STDEXEC::__get<_Index::value>(__state.__values_);
847+
using _Tuple = __decayed_tuple<_Args...>;
848+
static_assert(__same_as<decltype(*__opt_values), _Tuple&>,
849+
"One of the senders in this when_all() is fibbing about what types it "
850+
"sends");
851+
if constexpr ((__nothrow_decay_copyable<_Args> && ...))
879852
{
880853
__opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...});
881854
}
882-
STDEXEC_CATCH_ALL
855+
else
883856
{
884-
__set_error(__state, std::current_exception());
857+
STDEXEC_TRY
858+
{
859+
__opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...});
860+
}
861+
STDEXEC_CATCH_ALL
862+
{
863+
__set_error(__state, std::current_exception());
864+
}
885865
}
886866
}
887867
}

‎test/exec/test_when_any.cpp‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,17 @@
1414
* limitations under the License.
1515
*/
1616

17+
#include <atomic>
18+
#include <exec/async_scope.hpp>
1719
#include <exec/single_thread_context.hpp>
1820
#include <exec/when_any.hpp>
1921
#include <numbers>
2022
#include <test_common/receivers.hpp>
2123
#include <test_common/schedulers.hpp>
2224
#include <test_common/senders.hpp>
25+
#include <test_common/stop_tokens.hpp>
2326
#include <test_common/type_helpers.hpp>
27+
#include <thread>
2428

2529
#include <test_common/catch2.hpp>
2630

@@ -332,6 +336,40 @@ namespace
332336
REQUIRE(result == 42);
333337
}
334338

339+
TEST_CASE("when_any handles stop requests from the environment correctly", "[adaptors][when_any]")
340+
{
341+
// Both children complete inside the stop request, and async_scope destroys
342+
// when_any's operation state before the request returns.
343+
auto snd = exec::when_any(completes_if(false), completes_if(false));
344+
345+
exec::async_scope scope;
346+
scope.spawn(snd);
347+
scope.request_stop();
348+
ex::sync_wait(scope.on_empty());
349+
}
350+
351+
TEST_CASE("when_any completes once when a stop request races with its last child",
352+
"[adaptors][when_any]")
353+
{
354+
impulse_scheduler sched;
355+
racing_stop_source source;
356+
std::atomic<int> values{0};
357+
std::atomic<int> stopped{0};
358+
359+
auto op = ex::connect(exec::when_any(ex::schedule(sched)),
360+
racing_stop_receiver{&values, &stopped, source.get_token()});
361+
ex::start(op);
362+
363+
std::thread requester{[&] { source.request_stop(); }};
364+
source.wait_for_request(); // the requester has taken when_any's stop callback
365+
sched.start_next(); // the only child completes on this thread
366+
requester.join();
367+
368+
CHECK(source.destructor_waited());
369+
CHECK(values == 0);
370+
CHECK(stopped == 1);
371+
}
372+
335373
TEST_CASE("when_any completion signatures", "[adaptors][when_any]")
336374
{
337375
struct move_throws

‎test/stdexec/algos/adaptors/test_when_all.cpp‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,16 @@
2525
#include <test_common/receivers.hpp>
2626
#include <test_common/schedulers.hpp>
2727
#include <test_common/senders.hpp>
28+
#include <test_common/stop_tokens.hpp>
2829
#include <test_common/type_helpers.hpp>
2930

31+
#if STDEXEC_USE_MODULES()
32+
import std;
33+
#else
34+
# include <atomic>
35+
# include <thread>
36+
#endif
37+
3038
namespace ex = STDEXEC;
3139

3240
// For testing `when_all_with_variant`, we just check a couple of examples,
@@ -601,6 +609,42 @@ namespace
601609
ex::sync_wait(scope.on_empty());
602610
}
603611

612+
TEST_CASE("when_all leaves stop requests from the environment to its children",
613+
"[adaptors][when_all]")
614+
{
615+
ex::inplace_stop_source source;
616+
source.request_stop();
617+
auto env = ex::prop(ex::get_stop_token, source.get_token());
618+
// In this environment the children can complete with set_stopped, so
619+
// when_all forwards the stop request to them, but they never do.
620+
auto snd = ex::when_all(stop_sensitive_sender{}, stop_sensitive_sender{});
621+
auto op = ex::connect(std::move(snd), expect_void_receiver(env));
622+
ex::start(op);
623+
}
624+
625+
TEST_CASE("when_all completes once when a stop request races with its last child",
626+
"[adaptors][when_all]")
627+
{
628+
impulse_scheduler sched;
629+
racing_stop_source source;
630+
std::atomic<int> values{0};
631+
std::atomic<int> stopped{0};
632+
633+
// just() completes in start(), so the impulse sender is the last child.
634+
auto op = ex::connect(ex::when_all(ex::just(), ex::schedule(sched)),
635+
racing_stop_receiver{&values, &stopped, source.get_token()});
636+
ex::start(op);
637+
638+
std::thread requester{[&] { source.request_stop(); }};
639+
source.wait_for_request(); // the requester has taken when_all's stop callback
640+
sched.start_next(); // the last child completes on this thread
641+
requester.join();
642+
643+
CHECK(source.destructor_waited());
644+
CHECK(values == 1);
645+
CHECK(stopped == 0);
646+
}
647+
604648
TEST_CASE("regression test for #1988", "[adaptors][when_all]")
605649
{
606650
// just test that this compiles:

0 commit comments

Comments
 (0)