exec.split: implementation and test file

Author: maikel

include/beman/execution26/detail/split.hpp

@@ -0,0 +1,151 @@
+// include/beman/execution26/detail/split.hpp                         -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#ifndef INCLUDED_BEMAN_EXECUTION26_DETAIL_SPLIT
+#define INCLUDED_BEMAN_EXECUTION26_DETAIL_SPLIT
+
+#include <beman/execution26/detail/atomic_intrusive_queue.hpp>
+#include <beman/execution26/detail/connect_result_t.hpp>
+#include <beman/execution26/detail/default_impls.hpp>
+#include <beman/execution26/detail/get_stop_token.hpp>
+#include <beman/execution26/detail/impls_for.hpp>
+#include <beman/execution26/detail/inplace_stop_source.hpp>
+#include <beman/execution26/detail/meta_combine.hpp>
+#include <beman/execution26/detail/meta_unique.hpp>
+#include <beman/execution26/detail/receiver.hpp>
+#include <beman/execution26/detail/value_types_of_t.hpp>
+
+#include <atomic>
+#include <optional>
+#include <variant>
+#include <tuple>
+
+// ----------------------------------------------------------------------------
+
+namespace beman::execution26::detail {
+
+struct split_t {};
+
+template <>
+struct impls_for<split_t> : ::beman::execution26::detail::default_impls {
+    struct local_state_base {
+        local_state_base* next;
+        virtual auto      complete() noexcept -> void = 0;
+
+      protected:
+        ~local_state_base() = default;
+    };
+
+    // [exec.split-9]
+    template <class Sndr>
+    struct shared_state;
+
+    template <class Sndr>
+    struct split_receiver {
+        using receiver_concept = ::beman::execution26::receiver_t;
+
+        split_receiver(shared_state<Sndr>* state) noexcept : sh_state(state) { state->inc_ref(); }
+
+        template <class Tag, class... Args>
+        void complete(Tag, Args&&... args) noexcept {
+            using tuple_t = ::beman::execution26::detail::decayed_tuple<Tag, Args...>;
+            try {
+                sh_state->result.template emplace<tuple_t>(Tag(), ::std::forward<Args>(args)...);
+            } catch (...) {
+                using tuple_err = ::std::tuple<::beman::execution26::set_error_t, ::std::exception_ptr>;
+                sh_state->result.template emplace<tuple_err>(::beman::execution26::set_error,
+                                                             ::std::current_exception());
+            }
+            const auto state = sh_state;
+            state->inc_ref();
+            state->notify();
+            state->dec_ref();
+        }
+
+        template <class... Args>
+        void set_value(Args&&... args) && noexcept {
+            complete(::beman::execution26::set_value, ::std::forward<Args>(args)...);
+        }
+
+        template <class Error>
+        void set_error(Error&& err) && noexcept {
+            complete(::beman::execution26::set_error, ::std::forward<Error>(err));
+        }
+
+        void set_stopped() && noexcept { complete(::beman::execution26::set_stopped); }
+
+        struct env {
+            shared_state<Sndr>* sh_state;
+
+            ::beman::execution26::inplace_stop_token query(::beman::execution26::get_stop_token_t) const noexcept {
+                return sh_state->stop_src.get_token();
+            }
+        };
+
+        env get_env() const noexcept { return env{sh_state}; }
+
+        shared_state<Sndr>* sh_state;
+    };
+
+    // [exec.split-10]
+    template <class Sndr>
+    struct shared_state {
+        template <class Tag>
+        struct tagged_tuple {
+            template <class... Args>
+            using invoke = ::std::tuple<Tag, ::std::decay_t<Args>...>;
+        };
+
+        using variant_type = ::beman::execution26::detail::meta::unique<::beman::execution26::detail::meta::combine<
+            ::std::variant<::std::monostate>,
+            ::std::variant<::std::tuple<::beman::execution26::set_stopped_t>>,
+            ::std::variant<::std::tuple<::beman::execution26::set_error_t, ::std::exception_ptr>>,
+            ::beman::execution26::error_types_of_t<Sndr,
+                                                   typename shared_state<Sndr>::env,
+                                                   tagged_tuple<::beman::execution26::set_error_t>::template invoke>,
+            ::beman::execution26::value_types_of_t<Sndr,
+                                                   typename shared_state<Sndr>::env,
+                                                   tagged_tuple<::beman::execution26::set_value_t>::template invoke>>>;
+
+        using state_list_type = ::beman::execution26::detail::atomic_intrusive_queue<&local_state_base::next>;
+
+        explicit shared_state(Sndr&& sndr);
+
+        void start() noexcept {
+            assert(op_state);
+            ::beman::execution26::start(*op_state);
+        }
+
+        void notify() noexcept {
+            op_state.reset();
+            ::beman::execution26::detail::intrusive_queue listeners = waiting_states.pop_all_and_shutdown();
+            while (auto listener = listeners.pop()) {
+                listener->complete();
+            }
+        }
+
+        void inc_ref() noexcept { ref_count.fetch_add(1); }
+
+        void dec_ref() noexcept {
+            if (1 == ref_count.fetch_sub(1)) {
+                delete this;
+            }
+        }
+
+        ::beman::execution26::inplace_stop_source                                           stop_src{};
+        variant_type                                                                        result{};
+        state_list_type                                                                     waiting_states{};
+        ::std::atomic<::std::size_t>                                                        ref_count{0};
+        ::std::optional<::beman::execution26::connect_result_t<Sndr, split_receiver<Sndr>>> op_state{};
+    };
+};
+
+} // namespace beman::execution26::detail
+
+namespace beman::execution26 {
+using split_t = ::beman::execution26::detail::split_t;
+
+inline constexpr ::beman::execution26::split_t split{};
+} // namespace beman::execution26
+
+#endif

tests/beman/execution26/CMakeLists.txt

@@ -49,6 +49,7 @@ list(
   exec-snd-expos.test
   exec-snd-transform.test
   exec-starts-on.test
+  exec-split.test
   exec-sync-wait.test
   exec-then.test
   exec-utils-cmplsigs.test

tests/beman/execution26/exec-split.test.cpp

@@ -0,0 +1,11 @@
+// src/beman/execution26/tests/split.test.cpp                         -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/execution26/detail/split.hpp>
+#include <beman/execution26/execution.hpp>
+#include <test/execution.hpp>
+#include <concepts>
+
+// ----------------------------------------------------------------------------
+
+TEST(exec_split) {}