added a write-up of creating a simple sender

examples/CMakeLists.txt

@@ -21,6 +21,7 @@ set(TODO stop_token) #-dk:TODO including that causes a linker error
 set(TODO suspend_never) #-dk:TODO including that causes ASAN errors
 
 set(EXAMPLES
+    tutorial-create-a-sender
     allocator
     doc-just
     doc-just_error

examples/tutorial-create-a-sender.cpp

@@ -0,0 +1,124 @@
+// examples/tutorial/create-a-sender.cpp                              -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <iostream>
+#include <optional>
+#include <stack>
+#ifdef BEMAN_HAS_MODULES
+import beman.execution;
+#else
+#include <beman/execution/execution.hpp>
+#endif
+
+namespace ex = beman::execution;
+
+namespace {
+template <typename T>
+class asynchronous_stack {
+    struct node {
+        node*        next{};
+        virtual void complete(T) = 0;
+    };
+    std::stack<T> stack;
+    node*         awaiting{};
+
+    template <ex::receiver Rcvr>
+    struct state : node {
+        using operation_state_concept = ex::operation_state_tag;
+        struct stop_fun {
+            state& st;
+            void   operator()() noexcept {
+                state& s = this->st;
+                this->st.callback.reset();
+                for (auto it{&this->st.self.awaiting}; it; it = &(*it)->next) {
+                    if (*it == &this->st) {
+                        *it = this->st.next;
+                        break;
+                    }
+                }
+                ex::set_stopped(std::move(s.rcvr));
+            }
+        };
+        using stop_token_t = ex::stop_token_of_t<decltype(ex::get_env(std::declval<Rcvr&>()))>;
+        using callback_t   = ex::stop_callback_for_t<stop_token_t, stop_fun>;
+        std::remove_cvref_t<Rcvr> rcvr;
+        asynchronous_stack&       self;
+        std::optional<callback_t> callback;
+        state(Rcvr&& r, asynchronous_stack& s) : rcvr(std::forward<Rcvr>(r)), self(s) {}
+        void start() & noexcept {
+            if (not this->self.stack.empty()) {
+                T value(std::move(this->self.stack.top()));
+                this->self.stack.pop();
+                ex::set_value(std::move(rcvr), std::move(value));
+            } else {
+                this->next = std::exchange(this->self.awaiting, this);
+                this->callback.emplace(ex::get_stop_token(ex::get_env(this->rcvr)), stop_fun{*this});
+            }
+        }
+        void complete(T value) override {
+            this->callback.reset();
+            ex::set_value(std::move(rcvr), std::move(value));
+        }
+    };
+
+  public:
+    struct pop_sender {
+        using sender_concept = ex::sender_tag;
+        template <typename...>
+        static consteval auto get_completion_signatures() {
+            return ex::completion_signatures<ex::set_value_t(T), ex::set_stopped_t()>{};
+        }
+
+        asynchronous_stack& self;
+        template <ex::receiver Rcvr>
+        auto connect(Rcvr&& rcvr) const {
+            static_assert(ex::operation_state<state<Rcvr>>);
+            return state<Rcvr>{std::forward<Rcvr>(rcvr), self};
+        }
+    };
+
+    void push(T value) {
+        if (this->awaiting) {
+            std::exchange(this->awaiting, this->awaiting->next)->complete(std::move(value));
+        } else {
+            this->stack.push(std::move(value));
+        }
+    }
+    pop_sender pop() { return pop_sender{*this}; }
+};
+
+static_assert(ex::sender<asynchronous_stack<int>::pop_sender>);
+static_assert(ex::sender_in<asynchronous_stack<int>::pop_sender>);
+} // namespace
+// ----------------------------------------------------------------------------
+
+int main() {
+    ex::counting_scope      scope;
+    asynchronous_stack<int> st;
+    [[maybe_unused]] auto   sender = st.pop() | ex::then([](int v) { std::cout << "got value=" << v << "\n"; });
+
+    for (int value{1}; value < 4; ++value) {
+        st.push(value);
+    }
+    std::cout << "pushed 1,2,3\n";
+
+    int count{8};
+    for (int value{1}; value < count; ++value) {
+        ex::spawn(st.pop() | ex::then([value](int v) noexcept {
+                      std::cout << "got value=" << v << " for request " << value << "\n";
+                  }) | ex::upon_stopped([value] noexcept { std::cout << "request " << value << " was stopped\n"; }),
+                  scope.get_token());
+    }
+
+    std::cout << "requested " << (count - 1) << " values\n";
+
+    for (int value{4}; value < 7; ++value) {
+        st.push(value);
+    }
+    std::cout << "pushed 4,5,6\n";
+
+    scope.request_stop();
+    std::cout << "requested stop\n";
+    ex::sync_wait(scope.join());
+    std::cout << "joined\n";
+}

include/beman/execution/detail/call_result_t.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/call_result_t.hpp                 -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_CALL_RESULT
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_CALL_RESULT
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_CALL_RESULT_T
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_CALL_RESULT_T
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
@@ -25,4 +25,4 @@ using call_result_t = decltype(::std::declval<Fun>()(std::declval<Args>()...));
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_CALL_RESULT
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_CALL_RESULT_T

include/beman/execution/detail/completion_signatures_of_t.hpp

@@ -1,8 +1,8 @@
-// include/beman/execution/detail/completion_signaturess_of_t.hpp   -*-C++-*-
+// include/beman/execution/detail/completion_signatures_of_t.hpp   -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_COMPLETION_SIGNATURES_OF
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_COMPLETION_SIGNATURES_OF
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_COMPLETION_SIGNATURES_OF_T
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_COMPLETION_SIGNATURES_OF_T
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_MODULES
@@ -31,4 +31,4 @@ using completion_signatures_of_t = decltype(::beman::execution::get_completion_s
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_COMPLETION_SIGNATURES_OF
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_COMPLETION_SIGNATURES_OF_T

include/beman/execution/detail/connect_result_t.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/connect_result_t.hpp              -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_CONNECT_RESULT
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_CONNECT_RESULT
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_CONNECT_RESULT_T
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_CONNECT_RESULT_T
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
@@ -29,4 +29,4 @@ using connect_result_t = decltype(::beman::execution::connect(::std::declval<Sen
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_CONNECT_RESULT
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_CONNECT_RESULT_T

include/beman/execution/detail/dependent_sender_error.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/dependent_sender_error.hpp          -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_DEPENDENT_SENDER_ERROR
-#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_DEPENDENT_SENDER_ERROR
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_DEPENDENT_SENDER_ERROR
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_DEPENDENT_SENDER_ERROR
 
 // ----------------------------------------------------------------------------
 

include/beman/execution/detail/env_of_t.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/env_of_t.hpp                      -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_ENV_OF
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_ENV_OF
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_ENV_OF_T
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_ENV_OF_T
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
@@ -29,4 +29,4 @@ using env_of_t = decltype(::beman::execution::get_env(::std::declval<T>()));
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_ENV_OF
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_ENV_OF_T

include/beman/execution/detail/error_types_of.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/error_types_of.hpp                       -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_CONTEXT_ERROR_TYPES_OF
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_CONTEXT_ERROR_TYPES_OF
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF
 
 #include <beman/execution/execution.hpp>
 #include <exception>
@@ -25,4 +25,4 @@ using error_types_of_t = typename error_types_of<Context>::type;
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_CONTEXT_ERROR_TYPES_OF
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF

include/beman/execution/detail/error_types_of_t.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/error_types_of_t.hpp              -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF_T
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF_T
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
@@ -46,4 +46,4 @@ using error_types_of_t =
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_ERROR_TYPES_OF_T

include/beman/execution/detail/get_completion_domain.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/get_completion_domain.hpp           -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_GET_COMPLETION_DOMAIN
-#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_GET_COMPLETION_DOMAIN
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_GET_COMPLETION_DOMAIN
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_GET_COMPLETION_DOMAIN
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD

include/beman/execution/detail/hide_sched.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/hide_sched.hpp                      -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_HIDE_SCHED
-#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_HIDE_SCHED
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_HIDE_SCHED
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_HIDE_SCHED
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
@@ -48,4 +48,4 @@ auto hide_sched(const Q& q) noexcept {
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_HIDE_SCHED
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_HIDE_SCHED

include/beman/execution/detail/indeterminate_domain.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/indeterminate_domain.hpp            -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_INDETERMINATE_DOMAIN
-#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_INDETERMINATE_DOMAIN
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_INDETERMINATE_DOMAIN
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_INDETERMINATE_DOMAIN
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD

include/beman/execution/detail/inplace_stop_source.hpp

@@ -152,12 +152,14 @@ inline auto beman::execution::inplace_stop_source::request_stop() -> bool {
 }
 
 inline auto beman::execution::inplace_stop_source::add(callback_base* cb) -> void {
-    if (this->stopped) {
-        cb->call();
-    } else {
+    {
         ::std::lock_guard guard(this->lock);
-        cb->next = ::std::exchange(this->callbacks, cb);
+        if (!this->stopped) {
+            cb->next = ::std::exchange(this->callbacks, cb);
+            return;
+        }
     }
+    cb->call();
 }
 
 inline auto beman::execution::inplace_stop_source::deregister(callback_base* cb) -> void {

include/beman/execution/detail/intrusive_stack.hpp

@@ -1,8 +1,8 @@
-// include/beman/execution/detail/intrusive_queue.hpp               -*-C++-*-
+// include/beman/execution/detail/intrusive_stack.hpp               -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_INTRUSIVE_QUEUE
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_INTRUSIVE_QUEUE
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_INTRUSIVE_STACK
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_INTRUSIVE_STACK
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
@@ -17,20 +17,20 @@ namespace beman::execution::detail {
 template <auto Next>
 class intrusive_stack;
 
-//! @brief  This data structure is an intrusive queue that is not thread-safe.
+//! @brief  This data structure is an intrusive stack that is not thread-safe.
 template <class Item, Item* Item::* Next>
 class intrusive_stack<Next> {
   public:
     intrusive_stack() = default;
 
     explicit intrusive_stack(Item* head) noexcept : head_{head} {}
 
-    //! @brief  Pushes an item to the queue.
+    //! @brief  Pushes an item to the stack.
     auto push(Item* item) noexcept -> void { item->*Next = std::exchange(head_, item); }
 
-    //! @brief  Pops one item from the queue.
+    //! @brief  Pops one item from the stack.
     //!
-    //! @return  The item that was popped from the queue, or nullptr if the queue is empty.
+    //! @return  The item that was popped from the stack, or nullptr if the stack is empty.
     auto pop() noexcept -> Item* {
         if (head_) {
             auto item = head_;
@@ -40,7 +40,7 @@ class intrusive_stack<Next> {
         return nullptr;
     }
 
-    //! @brief  Tests if the queue is empty.
+    //! @brief  Tests if the stack is empty.
     auto empty() const noexcept -> bool { return !head_; }
 
   private:
@@ -49,4 +49,4 @@ class intrusive_stack<Next> {
 
 } // namespace beman::execution::detail
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_INTRUSIVE_QUEUE
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_INTRUSIVE_STACK

include/beman/execution/detail/is_constant.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/is_constant.hpp                     -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_IS_CONSTANT
-#define INCLUDED_INCLUDE_BEMAN_EXECUTION_DETAIL_IS_CONSTANT
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_IS_CONSTANT
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_IS_CONSTANT
 
 // ----------------------------------------------------------------------------
 

include/beman/execution/detail/schedule_result_t.hpp

@@ -1,8 +1,8 @@
 // include/beman/execution/detail/schedule_result_t.hpp             -*-C++-*-
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
-#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_SCHEDULE_RESULT
-#define INCLUDED_BEMAN_EXECUTION_DETAIL_SCHEDULE_RESULT
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_SCHEDULE_RESULT_T
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_SCHEDULE_RESULT_T
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
@@ -27,4 +27,4 @@ using schedule_result_t = decltype(::beman::execution::schedule(::std::declval<S
 
 // ----------------------------------------------------------------------------
 
-#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_SCHEDULE_RESULT
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_SCHEDULE_RESULT_T