From b5458e5a7a8bc1054889dbbd39e04af469b9dc49 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Tue, 7 Jul 2026 00:26:06 +0100 Subject: [PATCH 1/5] added a write-up of creating a simple sender --- docs/create-a-sender.md | 418 ++++++++++++++++++++++++++ examples/CMakeLists.txt | 1 + examples/tutorial-create-a-sender.cpp | 118 ++++++++ 3 files changed, 537 insertions(+) create mode 100644 docs/create-a-sender.md create mode 100644 examples/tutorial-create-a-sender.cpp diff --git a/docs/create-a-sender.md b/docs/create-a-sender.md new file mode 100644 index 00000000..4c3ef33c --- /dev/null +++ b/docs/create-a-sender.md @@ -0,0 +1,418 @@ +# Create A Sender + +This page describes how to create senders. The examples used aren't +necessarily super useful but are intended to describe the different +aspects of creating a sender. + +## Asynchronous Stack + +The first example is an asynchronous stack: popping from the stack +is an asynchronous operation which either completes immediately if +there is work or completes when new work becomes available. Pushing +to the asynchronous stack succeeds immediately. It could be an +extension to impose a limit on the stack size and make pushing also +a asynchronous operation making the two operation somewhat symmetric. + +The interface to the asynchronous stack could look like this: + +```c++ +template +class asynchronous_stack { + std::stack stack; + // TODO +public: + void push(T value); + pop_sender pop(); +}; +``` + +This immediately raises the question what is `pop_sender`? Well, +it is a sender, i.e., `ex::sender::pop_sender>` +is `true` which completes successfully with a `T` value: + +```c++ +template +class asynchronous_stack { +public: + struct pop_sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::completion_signatures{}; + } + // TODO + }; + + void push(T); + pop_sender pop() { return pop_sender{/* TODO */}; } +}; + +static_assert(ex::sender::pop_sender>); +static_assert(ex::sender_in::pop_sender>); +``` + +The declaration of `sender_concept` indicates to the `ex::execution` +library that `pop_sender` actually implements (models) the `concept` +`ex::sender`. That is required for various interfaces taking senders +as arguments. + +The function `get_completion_signatures()` is a compile-time function +which provides access to the possible completions of the sender +`pop_sender`. The current declaration states that `pop_sender` +always completes successfully with a value of type `T`. For example, +this sender could be used in a work graph where the result is sent to +a function: + +```c++ +asynchronous_stack st; +auto sender = st.pop() | ex::then([](int v) noexcept { std::cout << "got value=" << v << "\n"; }); +``` + +Of course, `sender` doesn't actually do anything, yet. The above +code is just showing how to create a work graph using this sender. +To actually make it do some work we'll need to implement its logic. +The logic for `pop_sender` will somehow need to look at an +`asynchronous_stack` to either extract a value if one is present +or to register interest in values becoming available. In both cases +it needs a reference to the `asynchronous_stack`: + +```c++ +template +class asynchronous_stack { +public: + struct pop_sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::completion_signatures{}; + } + + asynchronous_stack& self; + // TODO + }; + + void push(T); + pop_sender pop() { return pop_sender{*this}; } +}; +``` + +The `pop_sender` itself actually doesn't really do any work: it +wouldn't even know where to send any results to. Instead, it is the +interface used to `ex::connect` to a receiver which is provides a +way to notify the completion. When `ex::connect(sndr, rcvr)` is +invoked (and there are no customizations) it behaves as if +`sndr.connect(rcvr)` was invoked. Thus, `pop_sender` should have a +`connect` member function taking a receiver. The result of invoking +this function is an operation state: + +```c++ +template +class asynchronous_stack { + template + struct state { + using operation_state_concept = ex::operation_state_tag; + /* TODO */ + void start() & noexcept { /* TODO */ } + }; +public: + struct pop_sender { + // ... + + asynchronous_stack& self; + template + auto connect(Rcvr&& rcvr) const { + static_assert(ex::operation_state>); + return state{/* TODO */}; + } + }; + + // ... +}; +``` + +The `state` is identified as an `ex::operation_state` similarly to +a sender: by declaring `operation_state_concept` as an alias for +`ex::operation_state_tag` (or a class derived thereof) it is +identified as an `ex::operation_state`. To implement the `concept` +`ex::operation_state` the `state` class also needs a `start()` +function which never throws (any errors would be reported via +suitable completion signature on the erceiver). + +The actual work happens when `state`s `start()` function is +invoked: at that point it is checked whether there is any value +already available in the `asynchronous_stack` and, if that is +the case, it is passed to the receiver's `set_value` function. +Thus, the `state` will need a reference to the `asynchronous_stack` +and the receiver: + +```c++ +template +class asynchronous_stack { + template + struct state { + using operation_state_concept = ex::operation_state_tag; + std::remove_cvref_t rcvr; + asynchronous_stack& self; + void start() & noexcept { /* TODO */ } + }; +public: + struct pop_sender { + // ... + + asynchronous_stack& self; + template + auto connect(Rcvr&& rcvr) const { + return state{std::forward(rcvr), self}; + } + }; + + // ... +}; +``` + +With that we have created enough of the scaffolding to actually give +sender an initial try: for a test we can just use `asynchronous_stack` +and upon `start()`ing the operation state we could just complete with a +known value, e.g.: + +```c++ +template +class asynchronous_stack { + template + struct state { + using operation_state_concept = ex::operation_state_tag; + std::remove_cvref_t rcvr; + asynchronous_stack& self; + void start() & noexcept { ex::set_value(std::move(rcvr), 17); } + }; + // ... +}; + +int main() { + asynchronous_stack st; + auto sender = st.pop() | ex::then([](int v) noexcept { std::cout << "got value=" << v << "\n"; }); + ex::sync_wait(std::move(sender)); +} +``` + +Running the resulting executable (on my Mac that is +`./build/simple-Darwin/stagedir/bin/beman_execution.example.tutorial-create-a-sender`) +results in printing that it got value 17: + +``` +got value=17 +``` + +So, we got something running but it isn't quite the correct logic. +It is easy to implement the necessary logic when there are already +values in the stack: for that the `push(T)` function just adds +values to a `std::stack` and `state::start` checks whether there +is a value and produces a corresponding result if that is the case: + +```c++ +template +class asynchronous_stack { + std::stack stack; + + template + struct state { + // ... + 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)); + } + // TODO + } + }; + void push(T value) { this->stack.push(std::move(value)); } + // ... +}; + +int main() { + asynchronous_stack st; + auto sender = st.pop() | ex::then([](int v) noexcept { std::cout << "got value=" << v << "\n"; }); + + for (int value{1}; value < 4; ++ value) { + st.push(value); + } + + for (int value{1}; value < 4; ++ value) { + ex::sync_wait(sender); + } +} +``` + +The `push` function unconditionally pushes elements into the `stack` +and the `start` function doesn't do anything if there are no values. +If the `stack` is `empty`, the `start` function should register the +operation state with the `asynchronous_stack` and `push` should +verify if there is a registered operation state. Since the operation +state objects live at least until they complete they can be used +for an intrusive list: there is no need to do a separate allocation. +The list is built by deriving from a `struct node` which simply has +a `next` pointer to another `node` and a `virtual` function `complete` +taking a `T` object as argument. The `asynchronous_stack` keeps a +list of `node`s named `awaiting` in the example. This example doesn't +to be fair: the most recently started `pop_sender` is the first one +to complete. Since the `node` becomes a base class of `state`, it +is easiest to give `state` a constructor: + +```c++ +template +class asynchronous_stack { + struct node { + node* next{}; + virtual void complete(T) = 0; + }; + std::stack stack; + node* awaiting{}; + + template + struct state: node { + // ... + state(Rcvr&& r, asynchronous_stack& s): rcvr(std::forward(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); + } + } + void complete(T value) override { + ex::set_value(std::move(rcvr), std::move(value)); + } + }; + void push(T value) { this->stack.push(std::move(value)); } + // ... +}; +``` + +The new logic added is in `start` which now adds itself to the list +of `awaiting` operation state by replacing `awaiting` with itself +an setting its `next` pointer to the previous head of the list. To +take advantage of already `awaiting` operation states the `push` +function needs to check if there is any `node` and, if so, extract +this node `n` and invoke `n->complete(std::move(value))`: + +```c++ +template +class asynchronous_stack { + // ... + 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)); + } + } +}; +``` + +To actually demonstrate the functionality it becomes necessary to +`start` some `pop_sender`s before adding more work. To do so, the +`main` function uses a `ex::counting_scope`: invoking `ex::spawn(work, +scope.get_token())` results in `connect`ing `work` to a receiver +and `start`ing the result operation state. So, work is `push`ed in +two groups of three items. After the first three items are `push`ed +six `pop()` requests are `spawn`ed. The first three of them will +complete immediately. The second three will complete once more +work gets `push`ed: + +```c++ +int main() { + ex::counting_scope scope; + asynchronous_stack st; + auto sender = st.pop() | ex::then([](int v) noexcept { std::cout << "got value=" << v << "\n"; }); + + for (int value{1}; value < 4; ++ value) { + st.push(value); + } + std::cout << "pushed 1,2,3\n"; + + for (int value{1}; value < 7; ++ value) { + ex::spawn(st.pop() | ex::then([value](int v) noexcept { + std::cout << "got value=" << v << " for request " << value << "\n"; + }), scope.get_token()); + } + + std::cout << "requested 6 values\n"; + + for (int value{4}; value < 7; ++ value) { + st.push(value); + } + std::cout << "pushed 4,5,6\n"; + + ex::sync_wait(scope.join()); +} +``` + +The `counting_scope` needs to be `join`ed before it can be destroyed. +`scope.join()` returns a sender which completes once all held work +completes. The example is carefully setup to contain no outstanding +work when synchronously awaiting `scope.join()`. However, if more +`pop()` requests get `spawn`ed into the `scope` than there are +elements which are `push`ed, the operation will never complete! To +deal with that situation it is possibly to `scope.request_stop()` +before awaiting completion of `scope.join()`. The `scope.request_stop()` +will stop all outstanding work - assuming it can be stopped. But +`pop_sender` has no support, yet, for getting stopped. + +The way to support stop requests is to set up a stop callback using +the stop token associated with a receiver `r`. To get the stop token +`ex::get_stop_token(ex::get_env(r))` can be used. Such a stop token +support a callback which can then be used to complete with +`set_stopped`: + +```c++ +template +class asynchronous_stack { + // ... + template + struct state: node { + // ... + struct stop_fun { + state& self; + void operator()() noexcept { + state& self = this->self; + self.callback.reset(); + ex::set_stopped(std::move(self.rcvr)); + } + }; + using stop_token_t = ex::stop_token_of_t()))>; + using callback_t = ex::stop_callback_for_t; + std::optional callback; + state(Rcvr&& r, asynchronous_stack& s): rcvr(std::forward(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->callback.emplace(ex::get_stop_token(ex::get_env(this->rcvr)), *this); + this->next = std::exchange(this->self.awaiting, this); + } + } + void complete(T value) override { + this->callback.reset(); + ex::set_value(std::move(rcvr), std::move(value)); + } + }; + struct pop_sender { + using sender_concept = ex::sender_tag; + template + static consteval auto get_completion_signatures() { + return ex::completion_signatures{}; + } + // ... + }; + void push(T value) { this->stack.push(std::move(value)); } + // ... +}; +``` diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index aba158af..c3c9c243 100644 --- a/examples/CMakeLists.txt +++ b/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 diff --git a/examples/tutorial-create-a-sender.cpp b/examples/tutorial-create-a-sender.cpp new file mode 100644 index 00000000..b9ddb4df --- /dev/null +++ b/examples/tutorial-create-a-sender.cpp @@ -0,0 +1,118 @@ +// examples/tutorial/create-a-sender.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include +#include +#include +#include + +namespace ex = beman::execution; + +namespace { +template +class asynchronous_stack { + struct node { + node* next{}; + virtual void complete(T) = 0; + }; + std::stack stack; + node* awaiting{}; + + template + struct state: node { + using operation_state_concept = ex::operation_state_tag; + struct stop_fun { + state& self; + void operator()() noexcept { + ex::set_stopped(std::move(this->self.rcvr)); + } + }; + using stop_token_t = ex::stop_token_of_t()))>; + using callback_t = ex::stop_callback_for_t; + std::remove_cvref_t rcvr; + asynchronous_stack& self; + std::optional callback; + state(Rcvr&& r, asynchronous_stack& s): rcvr(std::forward(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->callback.emplace(ex::get_stop_token(ex::get_env(this->rcvr)), *this); + this->next = std::exchange(this->self.awaiting, 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 + static consteval auto get_completion_signatures() { + return ex::completion_signatures{}; + } + + asynchronous_stack& self; + template + auto connect(Rcvr&& rcvr) const { + static_assert(ex::operation_state>); + return state{std::forward(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::pop_sender>); +static_assert(ex::sender_in::pop_sender>); +} +// ---------------------------------------------------------------------------- + +int main() { + ex::counting_scope scope; + asynchronous_stack st; + 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{7}; + 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"; +} From ec687cf61ba1693b512b15c587b9094155cd2da1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Tue, 7 Jul 2026 05:51:53 +0100 Subject: [PATCH 2/5] address AI review feedback --- docs/create-a-sender.md | 19 ++++++++++++------- examples/tutorial-create-a-sender.cpp | 25 ++++++++++++++++++++----- 2 files changed, 32 insertions(+), 12 deletions(-) diff --git a/docs/create-a-sender.md b/docs/create-a-sender.md index 4c3ef33c..0bf6072a 100644 --- a/docs/create-a-sender.md +++ b/docs/create-a-sender.md @@ -11,7 +11,7 @@ is an asynchronous operation which either completes immediately if there is work or completes when new work becomes available. Pushing to the asynchronous stack succeeds immediately. It could be an extension to impose a limit on the stack size and make pushing also -a asynchronous operation making the two operation somewhat symmetric. +an asynchronous operation, making the two operations somewhat symmetric. The interface to the asynchronous stack could look like this: @@ -136,7 +136,7 @@ a sender: by declaring `operation_state_concept` as an alias for identified as an `ex::operation_state`. To implement the `concept` `ex::operation_state` the `state` class also needs a `start()` function which never throws (any errors would be reported via -suitable completion signature on the erceiver). +suitable completion signature on the receiver). The actual work happens when `state`s `start()` function is invoked: at that point it is checked whether there is any value @@ -377,11 +377,16 @@ class asynchronous_stack { struct state: node { // ... struct stop_fun { - state& self; + state& st; void operator()() noexcept { - state& self = this->self; - self.callback.reset(); - ex::set_stopped(std::move(self.rcvr)); + this->st.callback.reset(); + for (auto it{&this->st.stack.awaiting}; it; it = &(*it)->next) { + if (*it == &this->st) { + *it = this->st.next; + break; + } + } + ex::set_stopped(std::move(st.rcvr)); } }; using stop_token_t = ex::stop_token_of_t()))>; @@ -395,8 +400,8 @@ class asynchronous_stack { ex::set_value(std::move(rcvr), std::move(value)); } else { - this->callback.emplace(ex::get_stop_token(ex::get_env(this->rcvr)), *this); 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 { diff --git a/examples/tutorial-create-a-sender.cpp b/examples/tutorial-create-a-sender.cpp index b9ddb4df..3535fafa 100644 --- a/examples/tutorial-create-a-sender.cpp +++ b/examples/tutorial-create-a-sender.cpp @@ -1,10 +1,14 @@ // examples/tutorial/create-a-sender.cpp -*-C++-*- // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception -#include #include #include #include +#ifdef BEMAN_HAS_MODULES +import beman.execution; +#else +#include +#endif namespace ex = beman::execution; @@ -22,9 +26,20 @@ class asynchronous_stack { struct state: node { using operation_state_concept = ex::operation_state_tag; struct stop_fun { - state& self; + state& st; void operator()() noexcept { - ex::set_stopped(std::move(this->self.rcvr)); + std::cout << "request was stopped\n"; + state& s = this->st; + this->st.callback.reset(); + std::cout << "reset the stop callback\n"; + 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)); + std::cout << "set_stopped was called\n"; } }; using stop_token_t = ex::stop_token_of_t()))>; @@ -40,8 +55,8 @@ class asynchronous_stack { ex::set_value(std::move(rcvr), std::move(value)); } else { - this->callback.emplace(ex::get_stop_token(ex::get_env(this->rcvr)), *this); 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 { @@ -84,7 +99,7 @@ static_assert(ex::sender_in::pop_sender>); int main() { ex::counting_scope scope; asynchronous_stack st; - auto sender = st.pop() | ex::then([](int v){ std::cout << "got value=" << v << "\n"; }); + [[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); From 13bbb0720368843fee10bc0a892af9ecdb1d18ed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Tue, 7 Jul 2026 07:13:48 +0100 Subject: [PATCH 3/5] fixed formatting --- examples/tutorial-create-a-sender.cpp | 46 ++++++++----------- .../execution/detail/inplace_stop_source.hpp | 10 ++-- 2 files changed, 26 insertions(+), 30 deletions(-) diff --git a/examples/tutorial-create-a-sender.cpp b/examples/tutorial-create-a-sender.cpp index 3535fafa..4d277c60 100644 --- a/examples/tutorial-create-a-sender.cpp +++ b/examples/tutorial-create-a-sender.cpp @@ -16,18 +16,18 @@ namespace { template class asynchronous_stack { struct node { - node* next{}; + node* next{}; virtual void complete(T) = 0; }; std::stack stack; node* awaiting{}; template - struct state: node { + struct state : node { using operation_state_concept = ex::operation_state_tag; struct stop_fun { state& st; - void operator()() noexcept { + void operator()() noexcept { std::cout << "request was stopped\n"; state& s = this->st; this->st.callback.reset(); @@ -43,18 +43,17 @@ class asynchronous_stack { } }; using stop_token_t = ex::stop_token_of_t()))>; - using callback_t = ex::stop_callback_for_t; + using callback_t = ex::stop_callback_for_t; std::remove_cvref_t rcvr; asynchronous_stack& self; - std::optional callback; - state(Rcvr&& r, asynchronous_stack& s): rcvr(std::forward(r)), self(s) {} + std::optional callback; + state(Rcvr&& r, asynchronous_stack& s) : rcvr(std::forward(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 { + } else { this->next = std::exchange(this->self.awaiting, this); this->callback.emplace(ex::get_stop_token(ex::get_env(this->rcvr)), stop_fun{*this}); } @@ -64,7 +63,8 @@ class asynchronous_stack { ex::set_value(std::move(rcvr), std::move(value)); } }; -public: + + public: struct pop_sender { using sender_concept = ex::sender_tag; template @@ -80,11 +80,10 @@ class asynchronous_stack { } }; - void push(T value) { + void push(T value) { if (this->awaiting) { std::exchange(this->awaiting, this->awaiting->next)->complete(std::move(value)); - } - else { + } else { this->stack.push(std::move(value)); } } @@ -93,35 +92,30 @@ class asynchronous_stack { static_assert(ex::sender::pop_sender>); static_assert(ex::sender_in::pop_sender>); -} +} // namespace // ---------------------------------------------------------------------------- int main() { ex::counting_scope scope; asynchronous_stack st; - [[maybe_unused]] auto sender = st.pop() | ex::then([](int v){ std::cout << "got value=" << v << "\n"; }); + [[maybe_unused]] auto sender = st.pop() | ex::then([](int v) { std::cout << "got value=" << v << "\n"; }); - for (int value{1}; value < 4; ++ value) { + for (int value{1}; value < 4; ++value) { st.push(value); } std::cout << "pushed 1,2,3\n"; int count{7}; - 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()); + 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) { + for (int value{4}; value < 7; ++value) { st.push(value); } std::cout << "pushed 4,5,6\n"; diff --git a/include/beman/execution/detail/inplace_stop_source.hpp b/include/beman/execution/detail/inplace_stop_source.hpp index 053fa64e..4bc389ce 100644 --- a/include/beman/execution/detail/inplace_stop_source.hpp +++ b/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 { From 3137bdac42f93a4ce2c8d8b6748a3efe5afcc25a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Tue, 7 Jul 2026 07:16:15 +0100 Subject: [PATCH 4/5] fixed a trailing space in the .md file --- docs/create-a-sender.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/create-a-sender.md b/docs/create-a-sender.md index 0bf6072a..a3889721 100644 --- a/docs/create-a-sender.md +++ b/docs/create-a-sender.md @@ -391,7 +391,7 @@ class asynchronous_stack { }; using stop_token_t = ex::stop_token_of_t()))>; using callback_t = ex::stop_callback_for_t; - std::optional callback; + std::optional callback; state(Rcvr&& r, asynchronous_stack& s): rcvr(std::forward(r)), self(s) {} void start() & noexcept { if (not this->self.stack.empty()) { From 1ad75b6b9f8be2fee9c135df8138129ada542682 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dietmar=20K=C3=BChl?= Date: Wed, 8 Jul 2026 00:32:05 +0100 Subject: [PATCH 5/5] fixed a problem with destroying the object while the operation is stopped --- examples/tutorial-create-a-sender.cpp | 5 +-- include/beman/execution/detail/stop_when.hpp | 39 +++++++++++++++----- 2 files changed, 31 insertions(+), 13 deletions(-) diff --git a/examples/tutorial-create-a-sender.cpp b/examples/tutorial-create-a-sender.cpp index 4d277c60..bfadb9f8 100644 --- a/examples/tutorial-create-a-sender.cpp +++ b/examples/tutorial-create-a-sender.cpp @@ -28,10 +28,8 @@ class asynchronous_stack { struct stop_fun { state& st; void operator()() noexcept { - std::cout << "request was stopped\n"; state& s = this->st; this->st.callback.reset(); - std::cout << "reset the stop callback\n"; for (auto it{&this->st.self.awaiting}; it; it = &(*it)->next) { if (*it == &this->st) { *it = this->st.next; @@ -39,7 +37,6 @@ class asynchronous_stack { } } ex::set_stopped(std::move(s.rcvr)); - std::cout << "set_stopped was called\n"; } }; using stop_token_t = ex::stop_token_of_t()))>; @@ -105,7 +102,7 @@ int main() { } std::cout << "pushed 1,2,3\n"; - int count{7}; + 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"; diff --git a/include/beman/execution/detail/stop_when.hpp b/include/beman/execution/detail/stop_when.hpp index 77cbe9e9..2c29abb4 100644 --- a/include/beman/execution/detail/stop_when.hpp +++ b/include/beman/execution/detail/stop_when.hpp @@ -74,13 +74,20 @@ struct beman::execution::detail::stop_when_t::sender { using token2_t = decltype(::beman::execution::get_stop_token(::beman::execution::get_env(::std::declval()))); - struct cb_t { - ::beman::execution::inplace_stop_source& source; - auto operator()() const noexcept { this->source.request_stop(); } - }; struct base_state { rcvr_t rcvr; ::beman::execution::inplace_stop_source source{}; + bool run_stop{true}; + }; + struct cb_t { + base_state* st; + auto operator()() const noexcept { + this->st->run_stop = false; + this->st->source.request_stop(); + if (std::exchange(this->st->run_stop, true)) { + ::beman::execution::set_stopped(::std::move(this->st->rcvr)); + } + } }; struct env { base_state* st; @@ -103,13 +110,27 @@ struct beman::execution::detail::stop_when_t::sender { auto get_env() const noexcept -> env { return env{this->st}; } template auto set_value(A&&... a) const noexcept -> void { - ::beman::execution::set_value(::std::move(this->st->rcvr), ::std::forward(a)...); + if (this->st->run_stop) { + ::beman::execution::set_value(::std::move(this->st->rcvr), ::std::forward(a)...); + } else { + this->st->run_stop = true; + } } template auto set_error(E&& e) const noexcept -> void { - ::beman::execution::set_error(::std::move(this->st->rcvr), ::std::forward(e)); + if (this->st->run_stop) { + ::beman::execution::set_error(::std::move(this->st->rcvr), ::std::forward(e)); + } else { + this->st->run_stop = true; + } + } + auto set_stopped() const noexcept -> void { + if (this->st->run_stop) { + ::beman::execution::set_stopped(::std::move(this->st->rcvr)); + } else { + this->st->run_stop = true; + } } - auto set_stopped() const noexcept -> void { ::beman::execution::set_stopped(::std::move(this->st->rcvr)); } }; using inner_state_t = decltype(::beman::execution::connect(::std::declval(), ::std::declval())); @@ -127,9 +148,9 @@ struct beman::execution::detail::stop_when_t::sender { inner_state(::beman::execution::connect(::std::forward(s), receiver{&this->base})) {} auto start() & noexcept { - this->cb1.emplace(this->tok, cb_t{this->base.source}); + this->cb1.emplace(this->tok, cb_t{&this->base}); this->cb2.emplace(::beman::execution::get_stop_token(::beman::execution::get_env(this->base.rcvr)), - cb_t{this->base.source}); + cb_t{&this->base}); ::beman::execution::start(this->inner_state); } };