Skip to content

Fix when_all and when_any completing twice under a racing stop request #2358

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MaxFreedomPollard wants to merge 1 commit into
base: main
Choose a base branch
from
+342 −75
Open

Fix when_all and when_any completing twice under a racing stop request #2358

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 5 additions & 21 deletions include/exec/when_any.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,27 +83,8 @@ namespace experimental::execution

STDEXEC_IMMOVABLE(__opstate_base);

struct __forward_stop_request
{
constexpr void operator()() const noexcept
{
// Temporarily increment the count to avoid concurrent/recursive
// arrivals to pull the rug under our feet. Relaxed memory order is
// fine here.
__op_->__count_.fetch_add(1, __std::memory_order_relaxed);

__op_->__stop_source_.request_stop();

// Arrive in order to decrement the count again and complete if
// needed.
__op_->__arrive();
}

__opstate_base* __op_;
};

using __on_stop =
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>&>, __forward_stop_request>;
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>&>, __forward_stop_request<>>;

inplace_stop_source __stop_source_{};
std::optional<__on_stop> __on_stop_{};
Expand Down Expand Up @@ -227,8 +208,11 @@ namespace experimental::execution

void start() & noexcept
{
// Only forward the request to the children. The callback can still be
// running on another thread after the last child has arrived, so it
// must not arrive itself.
this->__on_stop_.emplace(get_stop_token(get_env(this->__rcvr_)),
typename __op_base_t::__forward_stop_request{this});
__forward_stop_request{this->__stop_source_});
if (this->__stop_source_.stop_requested())
{
STDEXEC::set_stopped(static_cast<_Receiver&&>(this->__rcvr_));
Expand Down
88 changes: 34 additions & 54 deletions include/stdexec/__detail/__when_all.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -561,35 +561,6 @@ namespace STDEXEC
struct _INVALID_ARGUMENTS_TO_WHEN_ALL_
{};

template <class _State>
struct __forward_stop_request
{
constexpr void operator()() const noexcept
{
// Temporarily increment the count to avoid concurrent/recursive
// arrivals to pull the rug under our feet. Relaxed memory order is fine
// here.
__state_->__count_.fetch_add(1, __std::memory_order_relaxed);

__state_t __expected = __started;
// Transition to the "stopped" state if and only if we're in the
// "started" state. (If this fails, it's because we're in an error
// state, which trumps cancellation.)
if (__state_->__state_.compare_exchange_strong(__expected, __stopped))
{
if constexpr (_State::__uses_stop_source)
{
__state_->__stop_source_.request_stop();
}
}

// Arrive in order to decrement the count again and complete if needed.
__state_->__arrive();
}

_State* __state_;
};

template <class _ErrorsVariant,
class _ValuesTuple,
class _Receiver,
Expand All @@ -600,7 +571,7 @@ namespace STDEXEC
using __receiver_t = _Receiver;
static constexpr bool __uses_stop_source = _UsesStopSource;
using __stop_callback_t =
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>>, __forward_stop_request<__state>>;
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>>, __forward_stop_request<>>;
using __stop_source_t = __if_c<_UsesStopSource, inplace_stop_source, __empty>;
using __on_stop_t = __if_c<_UsesStopSource, __optional<__stop_callback_t>, __empty>;

Expand All @@ -622,16 +593,6 @@ namespace STDEXEC
// All child operations have completed and arrived at the barrier.
switch (__state_.load(__std::memory_order_relaxed))
{
case __stopped:
if constexpr (_SendsStopped)
{
STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_));
break;
}
// This is reachable because the stop callback sets __stopped whether
// or not any child can send set_stopped, therefore we handle this the
// same as __started
[[fallthrough]];
case __started:
if constexpr (!__std::same_as<_ValuesTuple, __ignore>)
{
Expand All @@ -647,6 +608,17 @@ namespace STDEXEC
static_cast<_ErrorsVariant&&>(__errors_));
}
break;
case __stopped:
// Only a child completing with set_stopped puts us in this state.
if constexpr (_SendsStopped)
{
STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_));
}
else
{
STDEXEC_UNREACHABLE();
}
break;
default:;
}
}
Expand Down Expand Up @@ -797,8 +769,11 @@ namespace STDEXEC
{
if constexpr (_State::__uses_stop_source)
{
// Only forward the request to the children. The callback can still
// be running on another thread after the last child has arrived, so
// it must not arrive or change __state_ itself.
__state.__on_stop_.emplace(get_stop_token(STDEXEC::get_env(__state.__rcvr_)),
__forward_stop_request<_State>{&__state});
__forward_stop_request{__state.__stop_source_});
}
(STDEXEC::start(__child_ops), ...);
};
Expand Down Expand Up @@ -864,24 +839,29 @@ namespace STDEXEC
}
else if constexpr (!__same_as<_ValuesTuple, __ignore>)
{
auto& __opt_values = STDEXEC::__get<_Index::value>(__state.__values_);
using _Tuple = __decayed_tuple<_Args...>;
static_assert(__same_as<decltype(*__opt_values), _Tuple&>,
"One of the senders in this when_all() is fibbing about what types it "
"sends");
if constexpr ((__nothrow_decay_copyable<_Args> && ...))
// We only need to bother recording the completion values if we're
// not already in the "error" or "stopped" state.
if (__state.__state_.load() == __started)
{
__opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...});
}
else
{
STDEXEC_TRY
auto& __opt_values = STDEXEC::__get<_Index::value>(__state.__values_);
using _Tuple = __decayed_tuple<_Args...>;
static_assert(__same_as<decltype(*__opt_values), _Tuple&>,
"One of the senders in this when_all() is fibbing about what types it "
"sends");
if constexpr ((__nothrow_decay_copyable<_Args> && ...))
{
__opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...});
}
STDEXEC_CATCH_ALL
else
{
__set_error(__state, std::current_exception());
STDEXEC_TRY
{
__opt_values.emplace(_Tuple{static_cast<_Args&&>(__args)...});
}
STDEXEC_CATCH_ALL
{
__set_error(__state, std::current_exception());
}
}
}
}
Expand Down
38 changes: 38 additions & 0 deletions test/exec/test_when_any.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,17 @@
* limitations under the License.
*/

#include <atomic>
#include <exec/async_scope.hpp>
#include <exec/single_thread_context.hpp>
#include <exec/when_any.hpp>
#include <numbers>
#include <test_common/receivers.hpp>
#include <test_common/schedulers.hpp>
#include <test_common/senders.hpp>
#include <test_common/stop_tokens.hpp>
#include <test_common/type_helpers.hpp>
#include <thread>

#include <test_common/catch2.hpp>

Expand Down Expand Up @@ -332,6 +336,40 @@ namespace
REQUIRE(result == 42);
}

TEST_CASE("when_any handles stop requests from the environment correctly", "[adaptors][when_any]")
{
// Both children complete inside the stop request, and async_scope destroys
// when_any's operation state before the request returns.
auto snd = exec::when_any(completes_if(false), completes_if(false));

exec::async_scope scope;
scope.spawn(snd);
scope.request_stop();
ex::sync_wait(scope.on_empty());
}

TEST_CASE("when_any completes once when a stop request races with its last child",
"[adaptors][when_any]")
{
impulse_scheduler sched;
racing_stop_source source;
std::atomic<int> values{0};
std::atomic<int> stopped{0};

auto op = ex::connect(exec::when_any(ex::schedule(sched)),
racing_stop_receiver{&values, &stopped, source.get_token()});
ex::start(op);

std::thread requester{[&] { source.request_stop(); }};
source.wait_for_request(); // the requester has taken when_any's stop callback
sched.start_next(); // the only child completes on this thread
requester.join();

CHECK(source.destructor_waited());
CHECK(values == 0);
CHECK(stopped == 1);
}

TEST_CASE("when_any completion signatures", "[adaptors][when_any]")
{
struct move_throws
Expand Down
44 changes: 44 additions & 0 deletions test/stdexec/algos/adaptors/test_when_all.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,16 @@
#include <test_common/receivers.hpp>
#include <test_common/schedulers.hpp>
#include <test_common/senders.hpp>
#include <test_common/stop_tokens.hpp>
#include <test_common/type_helpers.hpp>

#if STDEXEC_USE_MODULES()
import std;
#else
# include <atomic>
# include <thread>
#endif

namespace ex = STDEXEC;

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

TEST_CASE("when_all leaves stop requests from the environment to its children",
"[adaptors][when_all]")
{
ex::inplace_stop_source source;
source.request_stop();
auto env = ex::prop(ex::get_stop_token, source.get_token());
// In this environment the children can complete with set_stopped, so
// when_all forwards the stop request to them, but they never do.
auto snd = ex::when_all(stop_sensitive_sender{}, stop_sensitive_sender{});
auto op = ex::connect(std::move(snd), expect_void_receiver(env));
ex::start(op);
}

TEST_CASE("when_all completes once when a stop request races with its last child",
"[adaptors][when_all]")
{
impulse_scheduler sched;
racing_stop_source source;
std::atomic<int> values{0};
std::atomic<int> stopped{0};

// just() completes in start(), so the impulse sender is the last child.
auto op = ex::connect(ex::when_all(ex::just(), ex::schedule(sched)),
racing_stop_receiver{&values, &stopped, source.get_token()});
ex::start(op);

std::thread requester{[&] { source.request_stop(); }};
source.wait_for_request(); // the requester has taken when_all's stop callback
sched.start_next(); // the last child completes on this thread
requester.join();

CHECK(source.destructor_waited());
CHECK(values == 1);
CHECK(stopped == 0);
}

TEST_CASE("regression test for #1988", "[adaptors][when_all]")
{
// just test that this compiles:
Expand Down
Loading
Loading