refactor the exec::repeat_n algorithm

Author: ericniebler

examples/nvexec/maxwell/snr.cuh

@@ -18,42 +18,27 @@
 
 #pragma once
 
-#include "common.cuh"
-#include "stdexec/execution.hpp" // IWYU pragma: export
+#include <stdexec/execution.hpp> // IWYU pragma: export
 
-namespace ex = stdexec;
+#include <exec/inline_scheduler.hpp>
+#include <exec/repeat_n.hpp>
+
+#include "./common.cuh"
 
 #if STDEXEC_CUDA_COMPILATION()
 #  include <nvexec/stream_context.cuh>    // IWYU pragma: export
 #  include <nvexec/multi_gpu_context.cuh> // IWYU pragma: export
 #else
 namespace nvexec {
-  struct stream_receiver_base {
-    using receiver_concept = ex::receiver_t;
-  };
-
-  struct stream_sender_base {
-    using sender_concept = ex::sender_t;
-  };
-
-  namespace detail {
-    struct stream_op_state_base { };
-  } // namespace detail
-
-  inline auto is_on_gpu() -> bool {
+  inline constexpr bool is_on_gpu() noexcept {
     return false;
   }
 } // namespace nvexec
 #endif
 
-#include <optional>
-#include <exec/inline_scheduler.hpp>
-
-STDEXEC_PRAGMA_PUSH()
-STDEXEC_PRAGMA_IGNORE_GNU("-Wmissing-braces")
-
 namespace ex = stdexec;
 
+<<<<<<< HEAD
 namespace repeat_n_detail {
   template <class SenderId, class Closure>
   struct repeat_n_sender_t;
@@ -454,6 +439,8 @@ namespace nvexec::_strm {
 
 #endif // STDEXEC_CUDA_COMPILATION()
 
+=======
+>>>>>>> f6ba2af5 (refactor the `exec::repeat_n` algorithm)
 template <class SchedulerT>
 [[nodiscard]]
 auto is_gpu_scheduler([[maybe_unused]] SchedulerT&& scheduler) -> bool {
@@ -469,13 +456,12 @@ auto maxwell_eqs_snr(
   std::size_t n_iterations,
   fields_accessor accessor,
   ex::scheduler auto&& computer) {
-  return ex::just()
-       | ex::on(
+  return ex::on(
            computer,
-           repeat_n(
-             n_iterations,
-             ex::bulk(ex::par, accessor.cells, update_h(accessor))
-               | ex::bulk(ex::par, accessor.cells, update_e(time, dt, accessor))))
+           ex::just() //
+             | ex::bulk(ex::par, accessor.cells, update_h(accessor))
+             | ex::bulk(ex::par, accessor.cells, update_e(time, dt, accessor))
+             | exec::repeat_n(n_iterations))
        | ex::then(dump_vtk(write_results, accessor));
 }
 
@@ -489,8 +475,8 @@ void run_snr(
   time_storage_t time{is_gpu_scheduler(computer)};
   fields_accessor accessor = grid.accessor();
 
-  auto init = ex::just()
-            | ex::on(computer, ex::bulk(ex::par, grid.cells, grid_initializer(dt, accessor)));
+  auto init =
+    ex::on(computer, ex::just() | ex::bulk(ex::par, grid.cells, grid_initializer(dt, accessor)));
   ex::sync_wait(init);
 
   auto snd = maxwell_eqs_snr(dt, time.get(), write_vtk, n_iterations, accessor, computer);
@@ -499,5 +485,3 @@ void run_snr(
     ex::sync_wait(std::move(snd));
   });
 }
-
-STDEXEC_PRAGMA_POP()

include/exec/repeat_n.hpp

@@ -19,16 +19,19 @@
 #include "../stdexec/execution.hpp"
 #include "../stdexec/__detail/__meta.hpp"
 #include "../stdexec/__detail/__basic_sender.hpp"
-#include "../stdexec/__detail/__manual_lifetime.hpp"
+#include "../stdexec/__detail/__optional.hpp"
 
 #include "trampoline_scheduler.hpp"
 #include "sequence.hpp"
 
-#include "../stdexec/__detail/__atomic.hpp"
 #include <cstddef>
 #include <exception>
 #include <type_traits>
 
+STDEXEC_PRAGMA_PUSH()
+STDEXEC_PRAGMA_IGNORE_EDG(expr_has_no_effect)
+STDEXEC_PRAGMA_IGNORE_GNU("-Wunused-value")
+
 namespace exec {
   namespace __repeat_n {
     using namespace stdexec;
@@ -46,28 +49,13 @@ namespace exec {
         using receiver_concept = stdexec::receiver_t;
         __repeat_n_state<_Sender, _Receiver> *__state_;
 
-        template <class... _Args>
-        void set_value(_Args &&...__args) noexcept {
-          STDEXEC_TRY {
-            __state_->__complete(set_value_t(), static_cast<_Args &&>(__args)...);
-          }
-          STDEXEC_CATCH_ALL {
-            if constexpr (!__nothrow_decay_copyable<_Args...>) {
-              __state_->__complete(set_error_t(), std::current_exception());
-            }
-          }
+        void set_value() noexcept {
+          __state_->__complete(set_value_t());
         }
 
         template <class _Error>
         void set_error(_Error &&__err) noexcept {
-          STDEXEC_TRY {
-            __state_->__complete(set_error_t(), static_cast<_Error &&>(__err));
-          }
-          STDEXEC_CATCH_ALL {
-            if constexpr (!__nothrow_decay_copyable<_Error>) {
-              __state_->__complete(set_error_t(), std::current_exception());
-            }
-          }
+          __state_->__complete(set_error_t(), static_cast<_Error &&>(__err));
         }
 
         void set_stopped() noexcept {
@@ -89,10 +77,6 @@ namespace exec {
     template <class _Child>
     __child_count_pair(_Child, std::size_t) -> __child_count_pair<_Child>;
 
-    STDEXEC_PRAGMA_PUSH()
-
-    STDEXEC_PRAGMA_IGNORE_GNU("-Wtsan")
-
     template <class _Sender, class _Receiver>
     struct __repeat_n_state
       : stdexec::__enable_receiver_from_this<
@@ -107,73 +91,60 @@ namespace exec {
         __result_of<exec::sequence, schedule_result_t<trampoline_scheduler &>, __child_t &>;
       using __child_op_t = stdexec::connect_result_t<__child_on_sched_sender_t, __receiver_t>;
 
-      __child_count_pair<__child_t> __pair_;
-      __std::atomic_flag __started_{};
-      stdexec::__manual_lifetime<__child_op_t> __child_op_;
-      trampoline_scheduler __sched_;
-
       __repeat_n_state(_Sender &&__sndr, _Receiver &)
         : __pair_(__sexpr_apply(static_cast<_Sender &&>(__sndr), stdexec::__detail::__get_data())) {
         // Q: should we skip __connect() if __count_ == 0?
         __connect();
       }
 
-      ~__repeat_n_state() {
-        if (!__started_.test(__std::memory_order_acquire)) {
-          __std::atomic_thread_fence(__std::memory_order_release);
-          // TSan does not support __std::atomic_thread_fence, so we
-          // need to use the TSan-specific __tsan_release instead:
-          STDEXEC_WHEN(STDEXEC_TSAN(), __tsan_release(&__started_));
-          __child_op_.__destroy();
-        }
-      }
-
-      void __connect() {
-        __child_op_.__construct_from([this] {
-          return stdexec::connect(
-            exec::sequence(stdexec::schedule(__sched_), __pair_.__child_), __receiver_t{this});
-        });
+      auto __connect() -> __child_op_t & {
+        return __child_op_.__emplace_from(
+          stdexec::connect,
+          exec::sequence(stdexec::schedule(__sched_), __pair_.__child_),
+          __receiver_t{this});
       }
 
       void __start() noexcept {
         if (__pair_.__count_ == 0) {
           stdexec::set_value(static_cast<_Receiver &&>(this->__receiver()));
         } else {
-          [[maybe_unused]]
-          const bool __already_started = __started_.test_and_set(__std::memory_order_relaxed);
-          STDEXEC_ASSERT(!__already_started);
-          stdexec::start(__child_op_.__get());
+          stdexec::start(*__child_op_);
         }
       }
 
       template <class _Tag, class... _Args>
-      void __complete(_Tag, _Args... __args) noexcept { // Intentionally by value...
+      void __complete(_Tag, _Args &&...__args) noexcept { // Intentionally by value...
+        static_assert(sizeof...(_Args) <= 1);
+        static_assert(sizeof...(_Args) == 0 || std::is_same_v<_Tag, stdexec::set_error_t>);
         STDEXEC_ASSERT(__pair_.__count_ > 0);
-        __child_op_.__destroy(); // ... because this could potentially invalidate them.
-        if constexpr (same_as<_Tag, set_value_t>) {
-          STDEXEC_TRY {
+
+        STDEXEC_TRY {
+          auto __arg_copy = (0, ..., static_cast<_Args &&>(__args)); // copy any arg...
+          __child_op_.reset(); // ... because this could potentially invalidate it.
+
+          if constexpr (same_as<_Tag, set_value_t>) {
             if (--__pair_.__count_ == 0) {
-              stdexec::set_value(static_cast<_Receiver &&>(this->__receiver()));
+              stdexec::set_value(std::move(this->__receiver()));
             } else {
-              __connect();
-              stdexec::start(__child_op_.__get());
+              stdexec::start(__connect());
             }
+          } else {
+            _Tag()(std::move(this->__receiver()), static_cast<__decay_t<_Args> &&>(__arg_copy)...);
           }
-          STDEXEC_CATCH_ALL {
-            stdexec::set_error(
-              static_cast<_Receiver &&>(this->__receiver()), std::current_exception());
-          }
-        } else {
-          _Tag()(static_cast<_Receiver &&>(this->__receiver()), static_cast<_Args &&>(__args)...);
+        }
+        STDEXEC_CATCH_ALL {
+          stdexec::set_error(std::move(this->__receiver()), std::current_exception());
         }
       }
+
+      __child_count_pair<__child_t> __pair_;
+      stdexec::__optional<__child_op_t> __child_op_;
+      trampoline_scheduler __sched_;
     };
 
     template <class _Sender, class _Receiver>
     __repeat_n_state(_Sender &&, _Receiver &) -> __repeat_n_state<_Sender, _Receiver>;
 
-    STDEXEC_PRAGMA_POP()
-
     struct repeat_n_t;
     struct _REPEAT_N_EXPECTS_A_SENDER_OF_VOID_;
 
@@ -236,11 +207,11 @@ namespace exec {
         return {{__count}, {}, {}};
       }
 
-      template <class _Sender>
-      auto transform_sender(set_value_t, _Sender &&__sndr, __ignore) {
+      template <class _Sender, bool _NoThrow = __nothrow_decay_copyable<_Sender>>
+      auto transform_sender(set_value_t, _Sender &&__sndr, __ignore) noexcept(_NoThrow) {
         return __sexpr_apply(
           static_cast<_Sender &&>(__sndr),
-          []<class _Child>(__ignore, std::size_t __count, _Child __child) {
+          []<class _Child>(__ignore, std::size_t __count, _Child __child) noexcept(_NoThrow) {
             return __make_sexpr<__repeat_n_tag>(__child_count_pair{std::move(__child), __count});
           });
       }
@@ -263,3 +234,5 @@ namespace stdexec {
     };
   };
 } // namespace stdexec
+
+STDEXEC_PRAGMA_POP()