Add stdexec::__stop_when

This diff defines `stdexec::__stop_when` as the implementation of _`stop-when`_ and adds tests to validate the algorithm. Co-authored-by: Eric Niebler <eniebler@nvidia.com>
ericniebler · Feb 10, 2026 · Dec 14, 2025 · Dec 15, 2025 · Dec 16, 2025 · Dec 22, 2025
commit 08d1308911a00ab61702529da6c398ebd74f060c
diff --git a/include/stdexec/__detail/__stop_when.hpp b/include/stdexec/__detail/__stop_when.hpp
@@ -0,0 +1,185 @@
+/*
+ * Copyright (c) 2025 Ian Petersen
+ * Copyright (c) 2025 NVIDIA Corporation
+ *
+ * Licensed under the Apache License Version 2.0 with LLVM Exceptions
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *   https://llvm.org/LICENSE.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include "__execution_fwd.hpp"
+
+#include "__atomic.hpp"
+#include "__basic_sender.hpp"
+#include "__concepts.hpp"
+#include "__env.hpp"
+#include "__sender_concepts.hpp"
+#include "__sender_introspection.hpp"
+#include "__stop_token.hpp"
+
+#include <type_traits>
+#include <utility>
+
+namespace STDEXEC {
+  /////////////////////////////////////////////////////////////////////////////
+  // [exec.stop.when]
+  namespace __stop_when_ {
+
+    ////////////////////////////////////////////////////////////////////////////////////////////////
+    template <class _Token, class _Receiver>
+    struct __state {
+      _Token __token_;
+      _Receiver __rcvr_;
+    };
+
+    template <class _Token, class _Receiver>
+    __state(_Token, _Receiver) -> __state<_Token, _Receiver>;
+
+    struct __stop_when_t {
+      template <sender _Sender, unstoppable_token _Token>
+      constexpr _Sender&& operator()(_Sender&& __sndr, _Token&&) const noexcept {
+        return static_cast<_Sender&&>(__sndr);
+      }
+
+      template <sender _Sender, stoppable_token _Token>
+      constexpr auto operator()(_Sender&& __sndr, _Token&& __token) const noexcept(
+        __nothrow_constructible_from<std::remove_cvref_t<_Sender>, _Sender>
+        && __nothrow_constructible_from<std::remove_cvref_t<_Token>, _Token>) {
+        return __make_sexpr<__stop_when_t>(
+          static_cast<_Token&&>(__token), static_cast<_Sender&&>(__sndr));
+      }
+    };
+
+    struct __stop_when_impl : __sexpr_defaults {
+      template <class _Sender, class... _Env>
+      static consteval auto get_completion_signatures() //
+        -> __completion_signatures_of_t<__child_of<_Sender>, _Env...> {
+        static_assert(sender_expr_for<_Sender, __stop_when_t>);
+        return {};
+      };
+
+      static constexpr auto get_env = [](__ignore, const auto& __state) noexcept {
+        return __env::__join(
+          prop(get_stop_token, __state.__token_), STDEXEC::get_env(__state.__rcvr_));
+      };
+
+      template <stoppable_token _Token1, stoppable_token _Token2>
+      struct __fused_token {
+        _Token1 __tkn1_;
+        _Token2 __tkn2_;
+
+        friend constexpr bool operator==(const __fused_token&, const __fused_token&) = default;
+
+        [[nodiscard]]
+        bool stop_requested() const noexcept {
+          return __tkn1_.stop_requested() || __tkn2_.stop_requested();
+        }
+
+        [[nodiscard]]
+        bool stop_possible() const noexcept {
+          return __tkn1_.stop_possible() || __tkn2_.stop_possible();
+        }
+
+        template <class _Fn>
+        struct callback_type {
+          template <class _C>
+            requires __std::constructible_from<_Fn, _C>
+          explicit callback_type(__fused_token&& __ftkn, _C&& __fn)
+            noexcept(__nothrow_constructible_from<_Fn, _C>)
+            : __fn_(static_cast<_C&&>(__fn))
+            , __cb1_(std::move(__ftkn.__tkn1_), __cb(this))
+            , __cb2_(std::move(__ftkn.__tkn2_), __cb(this)) {
+          }
+
+          template <class _C>
+            requires __std::constructible_from<_Fn, _C>
+          explicit callback_type(const __fused_token& __ftkn, _C&& __fn)
+            noexcept(__nothrow_constructible_from<_Fn, _C>)
+            : __fn_(static_cast<_C&&>(__fn))
+            , __cb1_(__ftkn.__tkn1_, __cb(this))
+            , __cb2_(__ftkn.__tkn2_, __cb(this)) {
+          }
+
+          callback_type(callback_type&&) = delete;
+
+         private:
+          struct __cb {
+            callback_type* self;
+
+            void operator()() noexcept {
+              (*self)();
+            }
+          };
+
+          using __cb1_t = _Token1::template callback_type<__cb>;
+          using __cb2_t = _Token2::template callback_type<__cb>;
+
+          void operator()() noexcept {
+            if (!__called_.exchange(true, __std::memory_order_relaxed)) {
+              __fn_();
+            }
+          }
+
+          _Fn __fn_;
+          __std::atomic<bool> __called_{false};
+          __cb1_t __cb1_;
+          __cb2_t __cb2_;
+        };
+      };
+
+      struct __make_token_fn {
+        template <class _SenderToken, class _ReceiverToken>
+          requires stoppable_token<std::remove_cvref_t<_SenderToken>>
+                && unstoppable_token<std::remove_cvref_t<_ReceiverToken>>
+        [[nodiscard]]
+        std::remove_cvref_t<_SenderToken>
+          operator()(_SenderToken&& __sndrToken, _ReceiverToken&&) const noexcept {
+          // when the receiver's stop token is unstoppable, the net token is just
+          // the sender's captured token
+          return __sndrToken;
+        }
+
+        template <class _SenderToken, class _ReceiverToken>
+          requires stoppable_token<std::remove_cvref_t<_SenderToken>>
+                && stoppable_token<std::remove_cvref_t<_ReceiverToken>>
+        [[nodiscard]]
+        __fused_token<std::remove_cvref_t<_SenderToken>, std::remove_cvref_t<_ReceiverToken>>
+          operator()(_SenderToken&& __sndrToken, _ReceiverToken&& __rcvrToken) const noexcept {
+          // when the receiver's stop token is stoppable, the net token must be
+          // a fused token that responds to signals from both the sender's captured
+          // token and the receiver's token
+          return {
+            static_cast<_SenderToken&&>(__sndrToken), static_cast<_ReceiverToken&&>(__rcvrToken)};
+        }
+      };
+
+      static constexpr auto get_state =
+        []<class _Self, class _Receiver>(_Self&& __self, _Receiver __rcvr) noexcept {
+          auto& [__tag, __token, __child] = __self;
+          auto __new_token = __make_token_fn{}(
+            __forward_like<_Self>(__token), get_stop_token(STDEXEC::get_env(__rcvr)));
+          return __state{std::move(__new_token), std::move(__rcvr)};
+        };
+    };
+  } // namespace __stop_when_
+
+  using __stop_when_::__stop_when_t;
+
+  /// @brief The stop-when sender adaptor, which fuses an additional stop token
+  ///        into its child sender such that the sender responds to stop
+  ///        requests from both the given stop token and the receiver's token
+  /// @hideinitializer
+  inline constexpr __stop_when_t __stop_when{};
+
+  template <>
+  struct __sexpr_impl<__stop_when_t> : __stop_when_::__stop_when_impl { };
+} // namespace STDEXEC
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
@@ -60,6 +60,7 @@ set(stdexec_test_sources
     stdexec/algos/adaptors/test_stopped_as_error.cpp
     stdexec/algos/adaptors/test_ensure_started.cpp
     stdexec/algos/adaptors/test_write_env.cpp
+    stdexec/algos/adaptors/test_stop_when.cpp
     stdexec/algos/consumers/test_start_detached.cpp
     stdexec/algos/consumers/test_sync_wait.cpp
     stdexec/algos/other/test_execute.cpp

diff --git a/test/stdexec/algos/adaptors/test_stop_when.cpp b/test/stdexec/algos/adaptors/test_stop_when.cpp
@@ -0,0 +1,127 @@
+/*
+ * Copyright (c) 2025 Ian Petersen
+ * Copyright (c) 2025 NVIDIA Corporation
+ *
+ * Licensed under the Apache License Version 2.0 with LLVM Exceptions
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *   https://llvm.org/LICENSE.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <catch2/catch.hpp>
+#include <stdexec/execution.hpp>
+#include <stdexec/stop_token.hpp>
+#include <stdexec/__detail/__stop_when.hpp>
+#include <test_common/receivers.hpp>
+
+namespace ex = STDEXEC;
+
+namespace {
+  TEST_CASE("stop-when of an unstoppable token is the identity", "[adaptors][stop-when]") {
+    auto snd = ex::just(42);
+    auto checkIdentity = [](auto&& snd) {
+      auto&& result = ex::__stop_when(std::forward<decltype(snd)>(snd), ex::never_stop_token{});
+
+      REQUIRE(&snd == &result);
+    };
+
+    checkIdentity(snd);
+    checkIdentity(std::as_const(snd));
+    checkIdentity(std::move(snd));
+  }
+
+  TEST_CASE("stop-when(just(), token) returns a sender", "[adaptors][stop-when]") {
+    ex::inplace_stop_source source;
+    auto snd = ex::__stop_when(ex::just(), source.get_token());
+    STATIC_REQUIRE(ex::sender<decltype(snd)>);
+    wait_for_value(snd);
+  }
+
+  auto isTokenStoppable() {
+    return ex::read_env(ex::get_stop_token)
+         | ex::then([](auto token) noexcept { return token.stop_possible(); });
+  }
+
+  TEST_CASE(
+    "stop-when substitutes its token when the receiver's token is unstoppable",
+    "[adaptors][stop-when]") {
+
+    // check that, by default, wait_for_value provides an unstoppable stop token
+    wait_for_value(isTokenStoppable(), false);
+
+    // now, check that stop-when mixes in a stoppable token
+
+    ex::inplace_stop_source source;
+
+    REQUIRE(source.get_token().stop_possible());
+
+    wait_for_value(ex::__stop_when(isTokenStoppable(), source.get_token()), true);
+  }
+
+  TEST_CASE(
+    "stop-when fuses its token with the receiver's when both are stoppable",
+    "[adaptors][stop-when]") {
+    ex::inplace_stop_source source;
+    wait_for_value(
+      ex::__stop_when(ex::__stop_when(isTokenStoppable(), source.get_token()), source.get_token()),
+      true);
+  }
+
+  template <std::invocable Fn>
+  ex::sender auto make_stop_callback(Fn&& fn) noexcept {
+    return ex::read_env(ex::get_stop_token)
+         | ex::then([fn = std::forward<Fn>(fn)](auto token) mutable noexcept {
+             using cb_t = decltype(token)::template callback_type<std::remove_cvref_t<Fn>>;
+             return std::optional<cb_t>(std::in_place, std::move(token), std::move(fn));
+           });
+  }
+
+  TEST_CASE("callbacks registered with stop-when's token can be invoked", "[adaptors][stop-when]") {
+    int invokeCount = 0;
+    auto snd = make_stop_callback([&invokeCount]() noexcept { invokeCount++; });
+
+    {
+      ex::inplace_stop_source source;
+      wait_for_value(
+        snd | ex::then([&](auto&& optCallback) noexcept {
+          source.request_stop();
+          optCallback.reset();
+          return invokeCount;
+        }),
+        0);
+    }
+
+    {
+      ex::inplace_stop_source source;
+
+      wait_for_value(
+        ex::__stop_when(snd, source.get_token()) | ex::then([&](auto&& optCallback) noexcept {
+          source.request_stop();
+          optCallback.reset();
+          return invokeCount;
+        }),
+        1);
+    }
+
+    {
+      ex::inplace_stop_source source1;
+      ex::inplace_stop_source source2;
+
+      wait_for_value(
+        ex::__stop_when(snd, source2.get_token()) | ex::then([&](auto&& optCallback) noexcept {
+          source1.request_stop();
+          source2.request_stop();
+          optCallback.reset();
+          return invokeCount;
+        }) | ex::write_env(ex::prop(ex::get_stop_token, source1.get_token())),
+        2);
+    }
+  }
+} // namespace