Async scope

CMakeLists.txt

@@ -40,6 +40,7 @@ option(
 include(GNUInstallDirs)
 set(INSTALL_CONFIGDIR ${CMAKE_INSTALL_LIBDIR}/cmake/${PROJECT_NAME})
 
+if(FALSE)
 if(PROJECT_IS_TOP_LEVEL AND NOT BEMAN_EXECUTION_ENABLE_INSTALL OR CMAKE_SKIP_INSTALL_RULES)
   set(CMAKE_SKIP_INSTALL_RULES ON)
 
@@ -90,6 +91,7 @@ if(PROJECT_IS_TOP_LEVEL AND NOT BEMAN_EXECUTION_ENABLE_INSTALL OR CMAKE_SKIP_INS
     # LINKER "lld"
   )
 endif()
+endif()
 
 add_subdirectory(src/beman/execution)
 

include/beman/execution/detail/as_tuple.hpp

@@ -0,0 +1,25 @@
+// include/beman/execution/detail/as_tuple.hpp                        -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_AS_TUPLE
+#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_AS_TUPLE
+
+#include <beman/execution/detail/decayed_tuple.hpp>
+
+// ----------------------------------------------------------------------------
+
+namespace beman::execution::detail {
+template <typename T>
+struct as_tuple;
+template <typename Rc, typename... A>
+struct as_tuple<Rc(A...)> {
+    using type = ::beman::execution::detail::decayed_tuple<Rc, A...>;
+};
+
+template <typename T>
+using as_tuple_t = typename ::beman::execution::detail::as_tuple<T>::type;
+} // namespace beman::execution::detail
+
+// ----------------------------------------------------------------------------
+
+#endif

include/beman/execution/detail/associate.hpp

@@ -0,0 +1,161 @@
+// include/beman/execution/detail/associate.hpp                       -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_ASSOCIATE
+#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_ASSOCIATE
+
+#include <beman/execution/detail/scope_token.hpp>
+#include <beman/execution/detail/sender.hpp>
+#include <beman/execution/detail/connect.hpp>
+#include <beman/execution/detail/transform_sender.hpp>
+#include <beman/execution/detail/get_domain_early.hpp>
+#include <beman/execution/detail/make_sender.hpp>
+#include <beman/execution/detail/default_impls.hpp>
+#include <beman/execution/detail/impls_for.hpp>
+#include <type_traits>
+#include <optional>
+#include <utility>
+#include <iostream> //-dk:TODO remove
+
+// ----------------------------------------------------------------------------
+
+namespace beman::execution::detail {
+template <::beman::execution::scope_token Token, ::beman::execution::sender Sender>
+struct associate_data {
+    using wrap_sender = ::std::remove_cvref_t<decltype(::std::declval<Token&>().wrap(::std::declval<Sender>()))>;
+
+    explicit associate_data(Token t, Sender&& s) : token(t), sender(this->token.wrap(::std::forward<Sender>(s))) {
+        if (!token.try_associate()) {
+            this->sender.reset();
+        }
+    }
+    associate_data(const associate_data& other) noexcept(::std::is_nothrow_copy_constructible_v<wrap_sender> &&
+                                                         noexcept(token.try_associate()))
+        : token(other.token), sender() {
+        if (other.sender && this->token.try_associate()) {
+            try {
+                this->sender.emplace(*other.sender);
+            } catch (...) {
+                this->token.disassociate();
+            }
+        }
+    }
+    associate_data(associate_data&& other) noexcept(::std::is_nothrow_move_constructible_v<wrap_sender>)
+        : token(other.token), sender(::std::move(other.sender)) {
+        other.sender.reset();
+    }
+    auto operator=(const associate_data&) -> associate_data& = delete;
+    auto operator=(associate_data&&) -> associate_data&      = delete;
+    ~associate_data() {
+        if (this->sender) {
+            this->sender.reset();
+            this->token.disassociate();
+        }
+    }
+
+    auto release() -> ::std::optional<::std::pair<Token, wrap_sender>> {
+        return this->sender ? (std::unique_ptr<std::optional<wrap_sender>, decltype([](auto* opt) { opt->reset(); })>(
+                                   &this->sender),
+                               ::std::optional{::std::pair{::std::move(this->token), ::std::move(*this->sender)}})
+                            : ::std::optional<::std::pair<Token, wrap_sender>>{};
+    }
+
+    Token                        token;
+    ::std::optional<wrap_sender> sender;
+};
+template <::beman::execution::scope_token Token, ::beman::execution::sender Sender>
+associate_data(Token, Sender&&) -> associate_data<Token, Sender>;
+
+struct associate_t {
+    template <::beman::execution::sender Sender, ::beman::execution::scope_token Token>
+    auto operator()(Sender&& sender, Token&& token) const {
+        auto domain(::beman::execution::detail::get_domain_early(sender));
+        return ::beman::execution::transform_sender(
+            domain,
+            ::beman::execution::detail::make_sender(
+                *this,
+                ::beman::execution::detail::associate_data(::std::forward<Token>(token),
+                                                           ::std::forward<Sender>(sender))));
+    }
+};
+
+template <>
+struct impls_for<associate_t> : ::beman::execution::detail::default_impls {
+    template <typename, typename>
+    struct get_noexcept : ::std::false_type {};
+    template <typename Tag, typename Data, typename Receiver>
+    struct get_noexcept<::beman::execution::detail::basic_sender<Tag, Data>, Receiver>
+        : ::std::bool_constant<
+              ::std::is_nothrow_move_constructible_v<typename ::std::remove_cvref_t<Data>::wrap_sender>&& ::beman::
+                  execution::detail::nothrow_callable<::beman::execution::connect_t,
+                                                      typename ::std::remove_cvref_t<Data>::wrap_sender,
+                                                      Receiver>> {};
+
+    static constexpr auto get_state =
+        []<typename Sender, typename Receiver>(Sender&& sender, Receiver& receiver) noexcept(
+            ::std::is_nothrow_constructible_v<::std::remove_cvref_t<Sender>, Sender>&&
+                get_noexcept<::std::remove_cvref_t<Sender>, Receiver>::value) {
+            auto [_, data] = ::std::forward<Sender>(sender);
+            auto dataParts{data.release()};
+
+            using scope_token = decltype(dataParts->first);
+            using wrap_sender = decltype(dataParts->second);
+            using op_t        = decltype(::beman::execution::connect(::std::move(dataParts->second),
+                                                              ::std::forward<Receiver>(receiver)));
+
+            struct op_state {
+                using sop_t        = op_t;
+                using sscope_token = scope_token;
+                bool associated{false};
+                union {
+                    Receiver* rcvr;
+                    struct {
+                        sscope_token tok;
+                        sop_t        op;
+                    } assoc;
+                };
+                explicit op_state(Receiver& r) noexcept : rcvr(::std::addressof(r)) {}
+                explicit op_state(sscope_token tk, wrap_sender&& sndr, Receiver& r) try
+                    : associated(true), assoc(tk, ::beman::execution::connect(::std::move(sndr), ::std::move(r))) {
+                } catch (...) {
+                    tk.disassociate();
+                    throw;
+                }
+                op_state(op_state&&) = delete;
+                ~op_state() {
+                    if (this->associated) {
+                        this->assoc.op.~sop_t();
+                        this->assoc.tok.disassociate();
+                        this->assoc.tok.~sscope_token();
+                    }
+                }
+                auto run() noexcept -> void {
+                    if (this->associated) {
+                        ::beman::execution::start(this->assoc.op);
+                    } else {
+                        ::beman::execution::set_stopped(::std::move(*this->rcvr));
+                    }
+                }
+            };
+            return dataParts ? op_state(::std::move(dataParts->first), ::std::move(dataParts->second), receiver)
+                             : op_state(receiver);
+        };
+    static constexpr auto start = [](auto& state, auto&&) noexcept -> void { state.run(); };
+};
+
+template <typename Data, typename Env>
+struct completion_signatures_for_impl<
+    ::beman::execution::detail::basic_sender<::beman::execution::detail::associate_t, Data>,
+    Env> {
+    using type = ::beman::execution::completion_signatures<::beman::execution::set_value_t()>;
+};
+} // namespace beman::execution::detail
+
+namespace beman::execution {
+using associate_t = ::beman::execution::detail::associate_t;
+inline constexpr associate_t associate{};
+} // namespace beman::execution
+
+// ----------------------------------------------------------------------------
+
+#endif

include/beman/execution/detail/counting_scope.hpp

@@ -0,0 +1,60 @@
+// include/beman/execution/detail/counting_scope.hpp                  -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_COUNTING_SCOPE
+#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_COUNTING_SCOPE
+
+#include <beman/execution/detail/counting_scope_base.hpp>
+#include <beman/execution/detail/counting_scope_join.hpp>
+#include <beman/execution/detail/sender.hpp>
+#include <beman/execution/detail/inplace_stop_source.hpp>
+#include <beman/execution/detail/stop_when.hpp>
+#include <utility>
+#include <cstdlib>
+
+// ----------------------------------------------------------------------------
+
+namespace beman::execution {
+class counting_scope;
+}
+
+// ----------------------------------------------------------------------------
+
+class beman::execution::counting_scope : public ::beman::execution::detail::counting_scope_base {
+  public:
+    class token;
+
+    auto join() noexcept -> ::beman::execution::sender auto {
+        return ::beman::execution::detail::counting_scope_join(this);
+    }
+    auto get_token() noexcept -> token;
+    auto request_stop() noexcept -> void { this->stop_source.request_stop(); }
+
+  private:
+    ::beman::execution::inplace_stop_source stop_source{};
+};
+
+// ----------------------------------------------------------------------------
+
+class beman::execution::counting_scope::token : public ::beman::execution::detail::counting_scope_base::token {
+  public:
+    template <::beman::execution::sender Sender>
+    auto wrap(Sender&& sender) const noexcept -> ::beman::execution::sender auto {
+        return ::beman::execution::detail::stop_when(
+            ::std::forward<Sender>(sender),
+            static_cast<::beman::execution::counting_scope*>(this->scope)->stop_source.get_token());
+    }
+
+  private:
+    friend class beman::execution::counting_scope;
+    explicit token(::beman::execution::counting_scope* s)
+        : ::beman::execution::detail::counting_scope_base::token(s) {}
+};
+
+// ----------------------------------------------------------------------------
+
+inline auto beman::execution::counting_scope::get_token() noexcept -> beman::execution::counting_scope::token {
+    return beman::execution::counting_scope::token(this);
+}
+
+#endif