Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
44 changes: 8 additions & 36 deletions include/exec/sequence_senders.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import stdexec;
# include "../stdexec/__detail/__receivers.hpp"
# include "../stdexec/__detail/__senders.hpp"
# include "../stdexec/__detail/__stop_token.hpp"
# include "../stdexec/__detail/__subscribe.hpp"
# include "../stdexec/__detail/__transform_sender.hpp"
# include "../stdexec/__detail/__type_traits.hpp"
# include "../stdexec/__detail/__utility.hpp"
Expand Down Expand Up @@ -880,18 +881,6 @@ namespace experimental::execution
&& sequence_receiver_of<_Receiver, item_types<STDEXEC::__decay_t<_Sender>>>
&& sender_to<next_sender_of_t<_Receiver, _Sender>, __stopped_means_break<_Receiver>>;

template <class _Sequence, class _Receiver>
using __subscribe_member_result_t = decltype(__declval<_Sequence>().subscribe(
__declval<_Receiver>()));

template <class _Sequence, class _Receiver>
concept __subscribable_with_member =
__minvocable_q<__subscribe_member_result_t, _Sequence, _Receiver>;

template <class _Sequence, class _Receiver>
using __subscribe_static_member_result_t = decltype(STDEXEC_REMOVE_REFERENCE(
_Sequence)::__static_subscribe(__declval<_Sequence>(), __declval<_Receiver>()));

// Receiver wrapper used to subscribe to the child of a transparent
// sequence adaptor (see `__sequence_adaptor_traits` above): it forwards
// all completions and `set_next` calls to the wrapped receiver, but
Expand Down Expand Up @@ -939,10 +928,6 @@ namespace experimental::execution
_Env __env_;
};

template <class _Sequence, class _Receiver>
concept __subscribable_with_static_member =
__minvocable_q<__subscribe_static_member_result_t, _Sequence, _Receiver>;

struct subscribe_t

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i would expect this to become a simple alias for STDEXEC::__subscribe_::__subscribe_t. why not?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the rest of subscribe_t uses set_next / next_sender_of_t, __stopped_means_break, item_types / sequence_sender_in and the adaptor bits from #2294, and those all live in exec right now. so making it an alias means moving most of the sequence sender basics into stdexec/__detail too. i kept this PR to just the hook lookup to keep it small, but happy to move all of that here if that's what you'd like. or would you rather have that as its own PR first?

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that's fine, i'm just making sure i'm understanding the PR correctly.

{
private:
Expand Down Expand Up @@ -1030,21 +1015,12 @@ namespace experimental::execution
return __declfn<__result_t,
__nothrow_subscribe && __nothrow_child_env && __nothrow_tfx_seq>();
}
else if constexpr (__subscribable_with_static_member<__tfx_seq_t, _Receiver>)
else if constexpr (STDEXEC::__subscribable<__tfx_seq_t, _Receiver>)
{
using __result_t = decltype(STDEXEC_REMOVE_REFERENCE(
__tfx_seq_t)::__static_subscribe(__declval<__tfx_seq_t>(), __declval<_Receiver>()));
using __result_t = STDEXEC::__subscribe_result_t<__tfx_seq_t, _Receiver>;
__check_operation_state<__result_t>();
constexpr bool __nothrow_subscribe = noexcept(STDEXEC_REMOVE_REFERENCE(
__tfx_seq_t)::__static_subscribe(__declval<__tfx_seq_t>(), __declval<_Receiver>()));
return __declfn<__result_t, __nothrow_subscribe && __nothrow_tfx_seq>();
}
else if constexpr (__subscribable_with_member<__tfx_seq_t, _Receiver>)
{
using __result_t = decltype(__declval<__tfx_seq_t>().subscribe(__declval<_Receiver>()));
__check_operation_state<__result_t>();
constexpr bool __nothrow_subscribe = noexcept(
__declval<__tfx_seq_t>().subscribe(__declval<_Receiver>()));
constexpr bool __nothrow_subscribe =
__nothrow_callable<decltype(STDEXEC::__subscribe), __tfx_seq_t, _Receiver>;
return __declfn<__result_t, __nothrow_subscribe && __nothrow_tfx_seq>();
}
else if constexpr (__is_debug_env<env_of_t<_Receiver>>)
Expand Down Expand Up @@ -1103,14 +1079,10 @@ namespace experimental::execution
__xform_t{}(static_cast<decltype(__env)&&>(__env),
STDEXEC::__forward_like<__tfx_seq_t>(__data))});
}
else if constexpr (__subscribable_with_static_member<__tfx_seq_t, _Receiver>)
{ // NOLINT(bugprone-branch-clone)
return __tfx_seq.__static_subscribe(static_cast<__tfx_seq_t&&>(__tfx_seq),
static_cast<_Receiver&&>(__rcvr));
}
else if constexpr (__subscribable_with_member<__tfx_seq_t, _Receiver>)
else if constexpr (STDEXEC::__subscribable<__tfx_seq_t, _Receiver>)
{
return static_cast<__tfx_seq_t&&>(__tfx_seq).subscribe(static_cast<_Receiver&&>(__rcvr));
return STDEXEC::__subscribe(static_cast<__tfx_seq_t&&>(__tfx_seq),
static_cast<_Receiver&&>(__rcvr));
}
else if constexpr (enable_sequence_sender<STDEXEC::__decay_t<__tfx_seq_t>>)
{
Expand Down
91 changes: 91 additions & 0 deletions include/stdexec/__detail/__subscribe.hpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
/*
* Copyright (c) 2026 NVIDIA Corporation
*
* Licensed under the Apache License Version 2.0 with LLVM Exceptions
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* https://llvm.org/LICENSE.txt
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#pragma once

#include "__config.hpp"

#if STDEXEC_USE_MODULES() && !defined(STDEXEC_IN_MODULE_PURVIEW)

import stdexec;

#else

# include "__execution_fwd.hpp"

# include "__concepts.hpp"
# include "__type_traits.hpp"
# include "__utility.hpp"

# include "__prologue.hpp"

// NOLINTNEXTLINE(bugprone-std-namespace-modification, modernize-concat-nested-namespaces)
namespace STDEXEC
{
namespace __subscribe_
{
template <class _Sender, class _Receiver>
concept __has_static_memfn = requires(_Sender && (*__sndr)(), _Receiver && (*__rcvr)()) {
STDEXEC_REMOVE_REFERENCE(_Sender)::__static_subscribe(__sndr(), __rcvr());
};

template <class _Sender, class _Receiver>
concept __has_memfn = requires(_Sender && (*__sndr)(), _Receiver && (*__rcvr)()) {
__sndr().subscribe(__rcvr());
};

// The customization point behind exec::subscribe. A sender hooks it with a
// static __static_subscribe(sndr, rcvr) member or a subscribe(rcvr) member
// function, the static one taking precedence. Senders defined in stdexec can
// use this to behave as sequence senders without depending on exec.
struct __subscribe_t
{
template <class _Sender, class _Receiver>
requires __has_static_memfn<_Sender, _Receiver>
STDEXEC_ATTRIBUTE(always_inline)
constexpr auto operator()(_Sender&& __sndr, _Receiver __rcvr) const
noexcept(noexcept(__sndr.__static_subscribe(__declval<_Sender>(), __declval<_Receiver>())))
-> decltype(__sndr.__static_subscribe(__declval<_Sender>(), __declval<_Receiver>()))
{
return __sndr.__static_subscribe(static_cast<_Sender&&>(__sndr),
static_cast<_Receiver&&>(__rcvr));
}

template <class _Sender, class _Receiver>
requires(!__has_static_memfn<_Sender, _Receiver>) && __has_memfn<_Sender, _Receiver>
STDEXEC_ATTRIBUTE(always_inline)
constexpr auto operator()(_Sender&& __sndr, _Receiver __rcvr) const
noexcept(noexcept(__declval<_Sender>().subscribe(__declval<_Receiver>())))
-> decltype(__declval<_Sender>().subscribe(__declval<_Receiver>()))
{
return static_cast<_Sender&&>(__sndr).subscribe(static_cast<_Receiver&&>(__rcvr));
}
};
} // namespace __subscribe_

STDEXEC_MODULE_EXPORT_AUTHORING
inline constexpr __subscribe_::__subscribe_t __subscribe{};

STDEXEC_MODULE_EXPORT_AUTHORING
template <class _Sender, class _Receiver>
using __subscribe_result_t = __call_result_t<__subscribe_::__subscribe_t, _Sender, _Receiver>;

STDEXEC_MODULE_EXPORT_AUTHORING
template <class _Sender, class _Receiver>
concept __subscribable = __callable<__subscribe_::__subscribe_t, _Sender, _Receiver>;
} // namespace STDEXEC

# include "__epilogue.hpp"
#endif // !STDEXEC_USE_MODULES() || defined(STDEXEC_IN_MODULE_PURVIEW)
1 change: 1 addition & 0 deletions include/stdexec/execution.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ import stdexec;
# include "__detail/__stopped_as_error.hpp"
# include "__detail/__stopped_as_optional.hpp"
# include "__detail/__submit.hpp"
# include "__detail/__subscribe.hpp"
# include "__detail/__sync_wait.hpp"
# include "__detail/__task.hpp"
# include "__detail/__task_scheduler.hpp"
Expand Down
40 changes: 40 additions & 0 deletions test/exec/test_sequence_senders.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -151,4 +151,44 @@ namespace
STATIC_REQUIRE(sequence_sender_to<seq_sender_t, next_receiver_t>);
STATIC_REQUIRE(sequence_sender_to<some_sequence_sender_of<set_value_t(int)>, next_receiver_t>);
}

struct tagged_operation
{
int tag;

void start() & noexcept {}
};

template <__completion_signature... _Sigs>
struct static_subscribe_sequence_of : some_sequence_sender_of<_Sigs...>
{
template <class Self, receiver R>
static auto __static_subscribe(Self&&, R&&) noexcept -> tagged_operation
{
return {1};
}

template <receiver R>
auto subscribe(R&&) const -> tagged_operation
{
return {2};
}
};

TEST_CASE("sequence_senders - __subscribe hooks", "[sequence_senders]")
{
using next_receiver_t = next_receiver<set_value_t(int), set_stopped_t()>;
using member_seq_t = some_sequence_sender_of<set_value_t(int), set_stopped_t()>;
using static_seq_t = static_subscribe_sequence_of<set_value_t(int), set_stopped_t()>;

STATIC_REQUIRE(__subscribable<member_seq_t, next_receiver_t>);
STATIC_REQUIRE(__same_as<__subscribe_result_t<member_seq_t, next_receiver_t>, nop_operation>);
STATIC_REQUIRE_FALSE(__subscribable<int, next_receiver_t>);

// the static hook wins over the member function
STATIC_REQUIRE(__subscribable<static_seq_t, next_receiver_t>);
STATIC_REQUIRE(noexcept(STDEXEC::__subscribe(static_seq_t{}, next_receiver_t{})));
CHECK(STDEXEC::__subscribe(static_seq_t{}, next_receiver_t{}).tag == 1);
CHECK(exec::subscribe(static_seq_t{}, next_receiver_t{}).tag == 1);
}
} // namespace