Skip to content

Add inplace_stoppable_base to reduce operation state sizes #2413

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ericniebler merged 17 commits into from
Nov 5, 2026
Merged

Add inplace_stoppable_base to reduce operation state sizes #2413

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
76bd202
Fix __shared stop callback invocable check
vasama Nov 5, 2026
507109d
Fix timed_thread_scheduler stop token usage
vasama Nov 5, 2026
4915dab
Fix stop_when stop callback invocable check
vasama Nov 5, 2026
e89efdb
Fix windows_thread_pool stop callback usage
vasama Nov 5, 2026
d39feb5
Fix task stop callback usage
vasama Nov 5, 2026
c6d5efb
Add inplace_stoppable_base, stoppable_base_for_t
vasama Nov 10, 2023
c68ec67
Use stoppable base in async_scope
vasama Nov 5, 2026
a1df009
Use stoppable base in io_uring context
vasama Nov 5, 2026
6e91466
Use stoppable base in when_all
vasama Nov 5, 2026
98d8594
Use stoppable base in when_any
vasama Nov 5, 2026
7a2e80b
Use stoppable base in __shared
vasama Nov 5, 2026
8fbbad2
Use stoppable base in merge_each
vasama Nov 5, 2026
18244ed
Use stoppable base in async_scope
vasama Nov 5, 2026
5474cd1
Use stoppable base in timed_thread_scheduler
vasama Nov 5, 2026
564032b
Use stoppable base in CUDA when_all
vasama Nov 5, 2026
f90ada0
Use stoppable base in CUDA ensure_started
vasama Nov 5, 2026
6688192
Use stoppable base in CUDA split
vasama Nov 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 46 additions & 31 deletions include/exec/async_scope.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,11 @@
#pragma once

#include "../stdexec/__detail/__intrusive_queue.hpp"
#include "../stdexec/__detail/__optional.hpp"
#include "../stdexec/execution.hpp"
#include "../stdexec/stop_token.hpp"
#include "completion_signatures.hpp"
#include "env.hpp"
#include "stoppable_base.hpp"

#include "../stdexec/__detail/__atomic.hpp"
#include <mutex>
Expand Down Expand Up @@ -281,16 +281,6 @@ namespace experimental::execution
__deleted
};

struct __forward_stopped
{
inplace_stop_source* __stop_source_;

void operator()() noexcept
{
__stop_source_->request_stop();
}
};

struct __subscription : __immovable
{
constexpr void __complete() noexcept
Expand Down Expand Up @@ -366,19 +356,25 @@ namespace experimental::execution
void (*__delete_)(_Ty*);
};

// Forwards stop requests on the scope's token to __stop_source_.
template <class _Completions, class _Env>
struct __future_state_base
: private inplace_stoppable_base<__future_state_base<_Completions, _Env>>
{
constexpr __future_state_base(_Env __env, __impl const * __scope)
: __forward_scope_{std::in_place,
__scope->__stop_source_.get_token(),
__forward_stopped{&__stop_source_}}
, __env_(make_env(static_cast<_Env&&>(__env),
: __env_(make_env(static_cast<_Env&&>(__env),
STDEXEC::prop{get_stop_token, __scope->__stop_source_.get_token()}))
{}
{
// Not in the initializer list: if the scope has already been stopped,
// this calls on_stop_requested(), which needs __stop_source_.
this->register_stoppable(__scope->__stop_source_.get_token());
}

~__future_state_base()
{
// The base is destroyed after __stop_source_, so stop forwarding first.
this->deregister_stoppable();

[[maybe_unused]]
std::unique_lock __guard{__mutex_};
if (__step_ == __future_step::__created)
Expand All @@ -404,14 +400,25 @@ namespace experimental::execution
STDEXEC_ASSERT(actual == __from);
}

inplace_stop_source __stop_source_;
STDEXEC::__optional<inplace_stop_callback<__forward_stopped>> __forward_scope_;
std::mutex __mutex_;
__future_step __step_ = __future_step::__created;
void __stop_forwarding_scope_() noexcept
{
this->deregister_stoppable();
}

void on_stop_requested() noexcept
{
__stop_source_.request_stop();
}

inplace_stop_source __stop_source_;
std::mutex __mutex_;
__future_step __step_ = __future_step::__created;
std::unique_ptr<__future_state_base, __dynamic_delete<__future_state_base>> __no_future_;
__completions_as_variant_t<_Completions> __data_;
__intrusive_queue<&__subscription::__next_> __subscribers_;
__env_t<_Env> __env_;

friend inplace_stoppable_base<__future_state_base>;
};

template <class _Completions, class _Env>
Expand All @@ -423,7 +430,7 @@ namespace experimental::execution
{
auto& __state = *__state_;
auto __local_subscribers = std::move(__state.__subscribers_);
__state.__forward_scope_.reset();
__state.__stop_forwarding_scope_();
if (__state.__no_future_.get() != nullptr)
{
// nobody is waiting for the results
Expand Down Expand Up @@ -530,17 +537,26 @@ namespace experimental::execution
// latter's members, and its member functions are only instantiated where
// __future::connect is instantiated, which is below.
template <class _Sender, class _Env, class _Receiver>
struct __future_opstate : __subscription
struct __future_opstate
: __subscription
, private stoppable_base_for_t<stop_token_of_t<env_of_t<_Receiver>>,
__future_opstate<_Sender, _Env, _Receiver>>
{
private:
using __forward_consumer_t =
stop_callback_for_t<stop_token_of_t<env_of_t<_Receiver>>, __forward_stopped>;
using __stop_base_t =
stoppable_base_for_t<stop_token_of_t<env_of_t<_Receiver>>, __future_opstate>;
friend __stop_base_t;

void on_stop_requested() noexcept
{
__state_->__stop_source_.request_stop();
}

constexpr void __complete_() noexcept
{
STDEXEC_TRY
{
__forward_consumer_.reset();
this->deregister_stoppable();
auto __state = std::move(__state_);
STDEXEC_ASSERT(__state != nullptr);
std::unique_lock __guard{__state->__mutex_};
Expand Down Expand Up @@ -589,8 +605,6 @@ namespace experimental::execution

STDEXEC_ATTRIBUTE(no_unique_address) _Receiver __rcvr_;
std::unique_ptr<__future_state<_Sender, _Env>> __state_;
STDEXEC_ATTRIBUTE(no_unique_address)
STDEXEC::__optional<__forward_consumer_t> __forward_consumer_;

public:
template <class _Receiver2>
Expand All @@ -601,13 +615,14 @@ namespace experimental::execution
{ static_cast<__future_opstate*>(__self)->__complete_(); }}
, __rcvr_(static_cast<_Receiver2&&>(__rcvr))
, __state_(std::move(__state))
, __forward_consumer_(std::in_place,
get_stop_token(get_env(__rcvr_)),
__forward_stopped{&__state_->__stop_source_})
{}
{
this->register_stoppable(get_stop_token(get_env(__rcvr_)));
}

constexpr ~__future_opstate() noexcept
{
// on_stop_requested() uses __state_, which is moved below.
this->deregister_stoppable();
if (__state_ != nullptr)
{
auto __raw_state = __state_.get();
Expand Down
35 changes: 18 additions & 17 deletions include/exec/detail/shared.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
#include "../../stdexec/__detail/__intrusive_slist.hpp"
#include "../../stdexec/__detail/__memory.hpp"
#include "../../stdexec/__detail/__meta.hpp"
#include "../../stdexec/__detail/__optional.hpp"
#include "../../stdexec/__detail/__queries.hpp"
#include "../../stdexec/__detail/__receivers.hpp"
#include "../../stdexec/__detail/__storage.hpp"
Expand All @@ -33,6 +32,7 @@
#include "../../stdexec/stop_token.hpp"

#include "../completion_signatures.hpp"
#include "../stoppable_base.hpp"

#include <exception>
#include <mutex>
Expand Down Expand Up @@ -152,9 +152,11 @@ namespace experimental::execution::__shared
// split-sender's `connect` fn returns. It holds a ref count to the shared
// state.
template <class _Tag, class _CvChild, class _Env, class _Receiver>
struct __local_state final : __local_state_base
struct __local_state final
: __local_state_base
, stoppable_base_for_t<stop_token_of_t<env_of_t<_Receiver>>,
__local_state<_Tag, _CvChild, _Env, _Receiver>>
{
using __stok_t = stop_token_of_t<env_of_t<_Receiver>>;
using __sh_state_t = __shared_state<_CvChild, _Env>;
using __sh_state_ptr_t = std::shared_ptr<__sh_state_t>;

Expand All @@ -181,14 +183,14 @@ namespace experimental::execution::__shared
// Scenario: there are no more split senders, this is the only operation
// state, the underlying operation has not yet been started, and the
// receiver's stop token is already in the "stop requested" state. Then
// registering the stop callback will call __local_state::operator() on
// *this synchronously. It may also be called asynchronously at any point
// after the callback is registered. Beware. We are guaranteed, however,
// that __local_state::operator() will not complete the operation or
// decrement the shared state's ref count until after *this has been added
// to the waiters list.
// registering the stop callback will call on_stop_requested() on *this
// synchronously. It may also be called asynchronously at any point after
// the callback is registered. Beware. We are guaranteed, however, that
// on_stop_requested() will not complete the operation or decrement the
// shared state's ref count until after *this has been added to the
// waiters list.
auto const __stok = STDEXEC::get_stop_token(STDEXEC::get_env(__rcvr_));
__on_stop_.emplace(__stok, *this);
this->register_stoppable(__stok);

// We haven't put __state in the waiters list yet and we are holding a ref
// count to __sh_state_, so nothing can happen to the __sh_state_ here.
Expand All @@ -207,13 +209,13 @@ namespace experimental::execution::__shared

// Otherwise, failed to add the waiter because of a stop-request.
// Complete synchronously with set_stopped().
__on_stop_.reset();
this->deregister_stoppable();
std::exchange(__sh_state_, {})->__detach();
STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_));
}

// Stop request callback:
constexpr void operator()() noexcept
void on_stop_requested() noexcept
{
// We reach here when a split/ensure_started sender has received a stop
// request from the receiver to which it is connected.
Expand Down Expand Up @@ -244,21 +246,20 @@ namespace experimental::execution::__shared
// This is called from __shared_state::__notify_waiters when the input async
// operation completes; or, if it has already completed when start is
// called, it is called from start: __notify cannot race with
// __local_state::operator(). See comment in __local_state::operator().
// on_stop_requested(). See comment in on_stop_requested().
constexpr void __notify() noexcept final
{
// The split algorithm sends by T const&. ensure_started sends by T&&.
constexpr bool __is_split = __std::same_as<split_t, _Tag>;
using __variant_t = decltype(__sh_state_->__results_);
using __cv_variant_t = __if_c<__is_split, __variant_t const &, __variant_t>;

__on_stop_.reset();
this->deregister_stoppable();
static_cast<__cv_variant_t&&>(__sh_state_->__results_).__complete(__rcvr_);
}

_Receiver __rcvr_;
__optional<stop_callback_for_t<__stok_t, __local_state&>> __on_stop_{};
__sh_state_ptr_t __sh_state_;
_Receiver __rcvr_;
__sh_state_ptr_t __sh_state_;
};

//////////////////////////////////////////////////////////////////////////////
Expand Down
72 changes: 47 additions & 25 deletions include/exec/linux/io_uring_context.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
# include "./safe_file_descriptor.hpp"

# include "../scope.hpp"
# include "../stoppable_base.hpp"

# if !__has_include(<linux/version.h>)
# error "linux/version.h not found. Do you use Linux?"
Expand Down Expand Up @@ -929,8 +930,7 @@ namespace experimental::execution
{
if (__op_->__n_ops_.fetch_sub(1, STDEXEC::__std::memory_order_relaxed) == 1)
{
__op_->__on_context_stop_.reset();
__op_->__on_receiver_stop_.reset();
__op_->__deregister_stop_callbacks_();
STDEXEC::set_stopped((static_cast<_Base&&>(*__op_)).receiver());
}
}
Expand Down Expand Up @@ -991,26 +991,31 @@ namespace experimental::execution
__base.submit_stop(__sqe);
};

struct __context_stop_tag
{};

struct __receiver_stop_tag
{};

// Stop requests on either the context's or the receiver's token start
// __stop_operation_.
template <__stoppable_task _Base>
struct __stoppable_task_facade : __impl_base<_Base, __has_submit_stop<_Base>>
struct __stoppable_task_facade
: __impl_base<_Base, __has_submit_stop<_Base>>
, private inplace_stoppable_base<__stoppable_task_facade<_Base>, __context_stop_tag>
, private stoppable_base_for_t<
STDEXEC::stop_token_of_t<STDEXEC::env_of_t<__receiver_of_t<_Base>>>,
__stoppable_task_facade<_Base>,
__receiver_stop_tag>
{
using __receiver_t = __receiver_of_t<_Base>;
using __base_t = __impl_base<_Base, __has_submit_stop<_Base>>;

struct __stop_callback
{
__stoppable_task_facade* __self_;

void operator()() noexcept
{
__self_->__stop_operation_.start();
}
};

using __on_context_stop_t = std::optional<STDEXEC::inplace_stop_callback<__stop_callback>>;
using __on_receiver_stop_t = std::optional<
STDEXEC::stop_callback_for_t<STDEXEC::stop_token_of_t<STDEXEC::env_of_t<__receiver_t>>,
__stop_callback>>;
using __context_stop_base_t =
inplace_stoppable_base<__stoppable_task_facade, __context_stop_tag>;
using __receiver_stop_base_t =
stoppable_base_for_t<STDEXEC::stop_token_of_t<STDEXEC::env_of_t<__receiver_t>>,
__stoppable_task_facade,
__receiver_stop_tag>;

template <class... _Args>
requires STDEXEC::__std::constructible_from<_Base, _Args...>
Expand Down Expand Up @@ -1048,18 +1053,17 @@ namespace experimental::execution
STDEXEC_ASSERT(prev == 0);
__context& __context_ = this->__base_.context();
__receiver_t& __receiver = this->__base_.receiver();
__on_context_stop_.emplace(__context_.get_stop_token(), __stop_callback{this});
__on_receiver_stop_.emplace(STDEXEC::get_stop_token(STDEXEC::get_env(__receiver)),
__stop_callback{this});
__context_stop_base_t::register_stoppable(__context_.get_stop_token());
__receiver_stop_base_t::register_stoppable(
STDEXEC::get_stop_token(STDEXEC::get_env(__receiver)));
this->__base_.submit(__sqe);
}

void complete(::io_uring_cqe const & __cqe) noexcept
{
if (__n_ops_.fetch_sub(1, STDEXEC::__std::memory_order_relaxed) == 1)
{
__on_context_stop_.reset();
__on_receiver_stop_.reset();
__deregister_stop_callbacks_();
__receiver_t& __receiver = this->__base_.receiver();
__context& __context_ = this->__base_.context();
auto token = STDEXEC::get_stop_token(STDEXEC::get_env(__receiver));
Expand All @@ -1074,10 +1078,28 @@ namespace experimental::execution
}
}

void __deregister_stop_callbacks_() noexcept
{
__context_stop_base_t::deregister_stoppable();
__receiver_stop_base_t::deregister_stoppable();
}

__stop_operation<__stoppable_task_facade> __stop_operation_;
STDEXEC::__std::atomic<int> __n_ops_{0};
__on_context_stop_t __on_context_stop_{};
__on_receiver_stop_t __on_receiver_stop_{};

private:
void on_stop_requested(__context_stop_tag) noexcept
{
__stop_operation_.start();
}

void on_stop_requested(__receiver_stop_tag) noexcept
{
__stop_operation_.start();
}

friend __context_stop_base_t;
friend __receiver_stop_base_t;
};

template <class _Base>
Expand Down
Loading
Loading