Merge branch 'main' into fix/read-with-default-connect-noexcept

Author: ericniebler

.github/workflows/ci.cpu.yml

@@ -142,11 +142,46 @@ jobs:
           path: /tmp/sccache*.log
           compression-level: 9
 
+  valgrind-run-loop:
+    runs-on: ubuntu-latest
+    name: CPU (gcc 14, Valgrind run_loop)
+    container:
+      options: -u root
+      image: rapidsai/devcontainers:26.08-cpp-gcc14-cuda12.9
+    steps:
+      - name: Checkout stdexec
+        uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+
+      - name: Install Valgrind
+        run: |
+          apt-get update
+          DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends valgrind
+
+      - name: Build run_loop reproducer
+        run: |
+          cmake -S . -B build -GNinja \
+            -DCMAKE_BUILD_TYPE=Debug \
+            -DCMAKE_CXX_STANDARD=20 \
+            -DCMAKE_CXX_EXTENSIONS=OFF \
+            -DSTDEXEC_BUILD_TESTS=ON
+          cmake --build build --target test.run_loop_finish_repro -v
+
+      - name: Run run_loop reproducer under Valgrind
+        run: |
+          timeout 60s valgrind \
+            --fair-sched=no \
+            --error-exitcode=1 \
+            --quiet \
+            ./build/test/test.run_loop_finish_repro
+
   ci-cpu:
     runs-on: ubuntu-latest
     name: CI (CPU)
     needs:
       - build-cpu
+      - valgrind-run-loop
     steps:
       - run: echo "CI (CPU) success"
 

include/exec/env.hpp

@@ -61,16 +61,19 @@ namespace experimental::execution
     struct __without_t
     {
       template <class _Env, class _Query>
-      constexpr auto operator()(_Env&& __env, _Query) const noexcept
+        requires STDEXEC::__queryable_with<_Env, _Query>
+      constexpr auto operator()(_Env&& __env, _Query) const
+        noexcept(STDEXEC::__nothrow_constructible_from<__without<_Env, _Query>, _Env>)
+          -> __without<_Env, _Query>
       {
-        if constexpr (STDEXEC::__queryable_with<_Env, _Query>)
-        {
-          return __without<_Env, _Query>{static_cast<_Env&&>(__env)};
-        }
-        else
-        {
-          return static_cast<_Env&&>(__env);
-        }
+        return __without<_Env, _Query>{static_cast<_Env&&>(__env)};
+      }
+
+      template <class _Env, class _Query>
+      constexpr auto operator()(_Env&& __env, _Query) const
+        noexcept(STDEXEC::__nothrow_move_constructible<_Env>) -> _Env
+      {
+        return static_cast<_Env&&>(__env);
       }
     };
 

include/exec/just_from.hpp

@@ -30,17 +30,17 @@ namespace experimental::execution
 
   namespace detail
   {
-    auto _just_from(just_from_t*) -> STDEXEC::set_value_t;
-    auto _just_from(just_error_from_t*) -> STDEXEC::set_error_t;
-    auto _just_from(just_stopped_from_t*) -> STDEXEC::set_stopped_t;
+    auto _just_from(just_from_t *) -> STDEXEC::set_value_t;
+    auto _just_from(just_error_from_t *) -> STDEXEC::set_error_t;
+    auto _just_from(just_stopped_from_t *) -> STDEXEC::set_stopped_t;
   }  // namespace detail
 
   template <class JustTag>
   struct _just_from
   {  // NOLINT(bugprone-crtp-constructor-accessibility)
    private:
     friend JustTag;
-    using _set_tag_t = decltype(detail::_just_from(static_cast<JustTag*>(nullptr)));
+    using _set_tag_t = decltype(detail::_just_from(static_cast<JustTag *>(nullptr)));
 
     using _diag_t = STDEXEC::__if_c<STDEXEC_IS_SAME(_set_tag_t, STDEXEC::set_error_t),
                                     AN_ERROR_COMPLETION_MUST_HAVE_EXACTLY_ONE_ERROR_ARGUMENT,
@@ -54,11 +54,11 @@ namespace experimental::execution
     struct _probe_fn
     {
       template <class... Ts>
-      auto operator()(Ts&&... ts) const noexcept -> _error_t<Ts...>;
+      auto operator()(Ts &&...ts) const noexcept -> _error_t<Ts...>;
 
       template <class... Ts>
         requires STDEXEC::__cmplsigs::__is_compl_sig<_set_tag_t(Ts...)>
-      auto operator()(Ts&&...) const noexcept -> STDEXEC::completion_signatures<_set_tag_t(Ts...)>
+      auto operator()(Ts &&...) const noexcept -> STDEXEC::completion_signatures<_set_tag_t(Ts...)>
       {
         return {};
       }
@@ -67,13 +67,13 @@ namespace experimental::execution
     template <class Rcvr>
     struct _complete_fn
     {
-      Rcvr& _rcvr;
+      Rcvr &_rcvr;
 
       template <class... Ts>
       STDEXEC_ATTRIBUTE(always_inline, host, device)
-      void operator()(Ts&&... ts) const noexcept
+      void operator()(Ts &&...ts) const noexcept
       {
-        _set_tag_t()(static_cast<Rcvr&&>(_rcvr), static_cast<Ts&&>(ts)...);
+        _set_tag_t()(static_cast<Rcvr &&>(_rcvr), static_cast<Ts &&>(ts)...);
       }
     };
 
@@ -89,17 +89,17 @@ namespace experimental::execution
       {
         if constexpr (STDEXEC::__nothrow_callable<Fn, _complete_fn<Rcvr>>)
         {
-          static_cast<Fn&&>(_fn)(_complete_fn<Rcvr>{_rcvr});
+          static_cast<Fn &&>(_fn)(_complete_fn<Rcvr>{_rcvr});
         }
         else
         {
           STDEXEC_TRY
           {
-            static_cast<Fn&&>(_fn)(_complete_fn<Rcvr>{_rcvr});
+            static_cast<Fn &&>(_fn)(_complete_fn<Rcvr>{_rcvr});
           }
           STDEXEC_CATCH_ALL
           {
-            STDEXEC::set_error(static_cast<Rcvr&&>(_rcvr), std::current_exception());
+            STDEXEC::set_error(static_cast<Rcvr &&>(_rcvr), std::current_exception());
           }
         }
       }
@@ -149,15 +149,15 @@ namespace experimental::execution
       auto connect(Rcvr rcvr) && noexcept(STDEXEC::__nothrow_decay_copyable<Rcvr, Fn>)
         -> _opstate<Rcvr, Fn>
       {
-        return _opstate<Rcvr, Fn>{static_cast<Rcvr&&>(rcvr), static_cast<Fn&&>(_fn)};
+        return _opstate<Rcvr, Fn>{static_cast<Rcvr &&>(rcvr), static_cast<Fn &&>(_fn)};
       }
 
       template <class Rcvr>
       STDEXEC_ATTRIBUTE(host, device)
       auto connect(Rcvr rcvr) const & noexcept(STDEXEC::__nothrow_decay_copyable<Rcvr, Fn const &>)
         -> _opstate<Rcvr, Fn>
       {
-        return _opstate<Rcvr, Fn>{static_cast<Rcvr&&>(rcvr), _fn};
+        return _opstate<Rcvr, Fn>{static_cast<Rcvr &&>(rcvr), _fn};
       }
 
       [[nodiscard]]
@@ -172,17 +172,17 @@ namespace experimental::execution
 
       template <class Rcvr>
       STDEXEC_ATTRIBUTE(host, device)
-      auto submit(Rcvr rcvr) && noexcept -> void
+      auto submit(Rcvr rcvr) && noexcept(STDEXEC::__nothrow_decay_copyable<Fn>) -> void
       {
-        auto op = static_cast<_sndr_base&&>(*this).connect(static_cast<Rcvr&&>(rcvr));
+        auto op = static_cast<_sndr_base &&>(*this).connect(static_cast<Rcvr &&>(rcvr));
         STDEXEC::start(op);
       }
 
       template <class Rcvr>
       STDEXEC_ATTRIBUTE(host, device)
-      auto submit(Rcvr rcvr) const & noexcept -> void
+      auto submit(Rcvr rcvr) const & noexcept(STDEXEC::__nothrow_decay_copyable<Fn const &>) -> void
       {
-        auto op = this->connect(static_cast<Rcvr&&>(rcvr));
+        auto op = this->connect(static_cast<Rcvr &&>(rcvr));
         STDEXEC::start(op);
       }
     };
@@ -203,7 +203,7 @@ namespace experimental::execution
                       "must return an instance of a specialization of " STDEXEC_PP_STRINGIZE(
                         STDEXEC) "::completion_signatures<>.");
         return _sndr<Fn>{
-          {{}, static_cast<Fn&&>(fn)}
+          {{}, static_cast<Fn &&>(fn)}
         };
       }
       else

include/exec/libdispatch_queue.hpp

@@ -537,7 +537,8 @@ namespace experimental::execution
 
       bulk_op_state(libdispatch_queue &queue, Shape shape, Fun fun, CvSender &&sndr, Receiver rcvr)
         : shared_state_(std::move(rcvr), shape, fun)
-        , inner_op_{STDEXEC::connect(std::move(sndr), bulk_rcvr{shared_state_, queue})}
+        , inner_op_{
+            STDEXEC::connect(static_cast<CvSender &&>(sndr), bulk_rcvr{shared_state_, queue})}
       {}
 
       void start() & noexcept

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

include/exec/static_thread_pool.hpp

@@ -930,8 +930,10 @@ namespace experimental::execution
       for (std::uint32_t i = 0; i < tasks.size(); ++i)
       {
         std::uint32_t index = i % this->available_parallelism();
-        queue.queues_[index].push_front(&tasks[i]);
-        thread_states_[index]->notify();
+        if (queue.queues_[index].push_front(&tasks[i]))
+        {
+          thread_states_[index]->notify();
+        }
       }
       // At this point the calling thread can exit and the pool will take over.
       // Ultimately, the last completing thread passes the result forward.

include/exec/when_any.hpp

@@ -34,39 +34,39 @@ namespace experimental::execution
     template <class _Env>
     using __env_t = __join_env_t<prop<get_stop_token_t, inplace_stop_token>, _Env>;
 
-    template <class... _Ts>
-    using __nothrow_decay_copyable_and_move_constructible_t = __mbool<(
-      (__nothrow_decay_copyable<_Ts> && __nothrow_move_constructible<__decay_t<_Ts>>) && ...)>;
-
     template <class... Args>
     using __as_rvalues = set_value_t (*)(__decay_t<Args>...);
 
     template <class... E>
-    using __as_error = set_error_t (*)(E...);
+    using __as_error = set_error_t (*)(__decay_t<E>...);
 
-    // Here we convert all set_value(Args...) to set_value(__decay_t<Args>...). Note, we keep all
-    // error types as they are and unconditionally add set_stopped(). The indirection through the
-    // __completions_fn is to avoid a pack expansion bug in nvc++.
+    // Here we convert all set_value(Args...) and set_error(Args...) to use decayed arguments and
+    // unconditionally add set_stopped(). The indirection through the __completions_fn is to avoid
+    // a pack expansion bug in nvc++.
     template <class... _Env>
     struct __completions_fn
     {
+      template <class _CvSender>
+      using __raw_completions_t = __completion_signatures_of_t<_CvSender, __env_t<_Env>...>;
+
+      template <class _CvSender>
+      using __decayed_completions_t =
+        __transform_reduce_completion_signatures_t<__raw_completions_t<_CvSender>,
+                                                   __as_rvalues,
+                                                   __as_error,
+                                                   set_stopped_t (*)(),
+                                                   __completion_signature_ptrs_t>;
+
       template <class... _CvSenders>
-      using __all_value_args_nothrow_decay_copyable =
-        __minvoke_q<__mand_t,
-                    __value_types_t<__completion_signatures_of_t<_CvSenders, __env_t<_Env>...>,
-                                    __qq<__nothrow_decay_copyable_and_move_constructible_t>,
-                                    __qq<__mand_t>>...>;
+      using __all_nothrow_decay_copyable_results =
+        __mand<__nothrow_decay_copyable_results_t<__raw_completions_t<_CvSenders>>...,
+               __nothrow_decay_copyable_results_t<__decayed_completions_t<_CvSenders>>...>;
 
       template <class... _CvSenders>
       using __f = __mtry_q<__concat_completion_signatures_t>::__f<
-        __eptr_completion_unless_t<__all_value_args_nothrow_decay_copyable<_CvSenders...>>,
+        __eptr_completion_unless_t<__all_nothrow_decay_copyable_results<_CvSenders...>>,
         completion_signatures<set_stopped_t()>,
-        __transform_reduce_completion_signatures_t<
-          __completion_signatures_of_t<_CvSenders, __env_t<_Env>...>,
-          __as_rvalues,
-          __as_error,
-          set_stopped_t (*)(),
-          __completion_signature_ptrs_t>...>;
+        __decayed_completions_t<_CvSenders>...>;
     };
 
     template <class _Env, class... _CvSenders>

include/nvexec/nvtx.cuh

@@ -110,7 +110,7 @@ namespace nv::execution
           static_cast<Self&&>(self).sndr_,
           static_cast<Receiver&&>(rcvr),
           [&](_strm::opstate_base<Receiver>& stream_provider) -> receiver_t<Receiver>
-          { return receiver_t<Receiver>(stream_provider, std::move(self.name_)); });
+          { return receiver_t<Receiver>(stream_provider, static_cast<Self&&>(self).name_); });
       }
       STDEXEC_EXPLICIT_THIS_END(connect)
 

include/stdexec/__detail/__any.hpp

@@ -48,7 +48,7 @@ import stdexec;
 STDEXEC_PRAGMA_IGNORE_GNU("-Wredundant-consteval-if")
 STDEXEC_PRAGMA_IGNORE_GNU("-Warray-bounds")
 
-// NOLINTBEGIN(moderize-use-override)
+// NOLINTBEGIN(modernize-use-override)
 
 STDEXEC_MODULE_EXPORT_AUTHORING
 namespace STDEXEC::__any
@@ -2283,7 +2283,7 @@ namespace STDEXEC::__any
 
 }  // namespace STDEXEC::__any
 
-// NOLINTEND(moderize-use-override)
+// NOLINTEND(modernize-use-override)
 
 #  include "__epilogue.hpp"
 #endif  // !STDEXEC_USE_MODULES() || defined(STDEXEC_IN_MODULE_PURVIEW)

include/stdexec/__detail/__bulk.hpp

@@ -281,7 +281,10 @@ namespace STDEXEC
 
           STDEXEC_TRY
           {
-            __state.__data_.__fun_(__shape_t(), __shape_t(__state.__data_.__shape_), __args...);
+            if (__shape_t{} < __state.__data_.__shape_)
+            {
+              __state.__data_.__fun_(__shape_t(), __shape_t(__state.__data_.__shape_), __args...);
+            }
             STDEXEC::set_value(static_cast<_State&&>(__state).__rcvr_,
                                static_cast<_Args&&>(__args)...);
           }