Skip to content

Commit 010afa4

Browse files
ericnieblerericniebler
committed
unify the storage of intermediate results
1 parent 6764a66 commit 010afa4

14 files changed

Lines changed: 234 additions & 353 deletions

‎include/exec/detail/shared.hpp‎

Lines changed: 7 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
#include "../../stdexec/__detail/__optional.hpp"
2828
#include "../../stdexec/__detail/__queries.hpp"
2929
#include "../../stdexec/__detail/__receivers.hpp"
30+
#include "../../stdexec/__detail/__storage.hpp"
3031
#include "../../stdexec/__detail/__transform_completion_signatures.hpp"
3132
#include "../../stdexec/__detail/__tuple.hpp"
3233
#include "../../stdexec/__detail/__variant.hpp" // IWYU pragma: keep
@@ -77,24 +78,6 @@ namespace experimental::execution::__shared
7778
template <class _Env>
7879
using __env_t = __join_env_t<prop<get_stop_token_t, inplace_stop_token>, _Env>;
7980

80-
struct __notify_fn
81-
{
82-
template <class _Receiver, class _Tag, class... _Args>
83-
constexpr void operator()(_Receiver& __rcvr, _Tag, _Args&&... __args) const noexcept
84-
{
85-
_Tag()(static_cast<_Receiver&&>(__rcvr), static_cast<_Args&&>(__args)...);
86-
}
87-
};
88-
89-
struct __notify_visitor
90-
{
91-
template <class _Receiver, class _Tuple>
92-
constexpr void operator()(_Receiver& __rcvr, _Tuple&& __tupl) const noexcept
93-
{
94-
STDEXEC::__apply(__notify_fn(), static_cast<_Tuple&&>(__tupl), __rcvr);
95-
};
96-
};
97-
9881
////////////////////////////////////////////////////////////////////////////////////////
9982
template <class _Env, class _Variant>
10083
struct __receiver
@@ -132,14 +115,10 @@ namespace experimental::execution::__shared
132115

133116
////////////////////////////////////////////////////////////////////////////////////////
134117
template <class _CvSender, class _Env>
135-
using __result_variant_t =
136-
__transform_reduce_completion_signatures_t<__completion_signatures_of_t<_CvSender, _Env>,
137-
__mbind_front_q<__decayed_tuple, set_value_t>::__f,
138-
__mbind_front_q<__decayed_tuple, set_error_t>::__f,
139-
__tuple<set_stopped_t>,
140-
__munique<__qq<__variant>>::__f,
141-
__tuple<set_error_t, std::exception_ptr>,
142-
__tuple<set_stopped_t>>;
118+
using __result_variant_t = __mapply<__mbind_front_q<__results_storage,
119+
set_stopped_t(),
120+
set_error_t(std::exception_ptr)>,
121+
__completion_signatures_of_t<_CvSender, _Env>>;
143122

144123
////////////////////////////////////////////////////////////////////////////////////////
145124
template <class _CvChild, class _Env>
@@ -269,10 +248,7 @@ namespace experimental::execution::__shared
269248
using __cv_variant_t = __if_c<__is_split, __variant_t const &, __variant_t>;
270249

271250
__on_stop_.reset();
272-
273-
STDEXEC::__visit(__notify_visitor(),
274-
static_cast<__cv_variant_t&&>(__sh_state_->__results_),
275-
__rcvr_);
251+
static_cast<__cv_variant_t&&>(__sh_state_->__results_).__complete(__rcvr_);
276252
}
277253

278254
_Receiver __rcvr_;
@@ -353,7 +329,7 @@ namespace experimental::execution::__shared
353329
__waiters_list_t __waiters_{};
354330
inplace_stop_source __stop_source_{};
355331
__env_t<_Env> __env_;
356-
_Variant __results_{__no_init}; // Initialized to the "set_stopped" state in the ctor.
332+
_Variant __results_; // Initialized to the "set_stopped" state in the ctor.
357333
};
358334

359335
template <class _Env, class _Variant>

‎include/exec/fork_join.hpp‎

Lines changed: 4 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
#pragma once
1717

1818
#include "../stdexec/__detail/__receiver_ref.hpp"
19+
#include "../stdexec/__detail/__storage.hpp"
1920
#include "../stdexec/execution.hpp"
2021

2122
#include <exception>
@@ -27,26 +28,6 @@ namespace experimental::execution
2728

2829
struct fork_join_impl_t
2930
{
30-
struct _dematerialize_fn
31-
{
32-
struct _impl_fn
33-
{
34-
template <class Rcvr, class Tag, class... Args>
35-
STDEXEC_ATTRIBUTE(always_inline, host, device)
36-
constexpr void operator()(Rcvr& rcvr, Tag, Args const &... args) const noexcept
37-
{
38-
Tag{}(static_cast<Rcvr&&>(rcvr), args...);
39-
}
40-
};
41-
42-
template <class Rcvr, class Tuple>
43-
STDEXEC_ATTRIBUTE(always_inline, host, device)
44-
constexpr void operator()(Rcvr& rcvr, Tuple const & tupl) const noexcept
45-
{
46-
STDEXEC::__apply(_impl_fn{}, tupl, rcvr);
47-
}
48-
};
49-
5031
struct _mk_when_all_fn
5132
{
5233
template <class CacheSndr, class... Closures>
@@ -58,17 +39,7 @@ namespace experimental::execution
5839
};
5940

6041
template <class Completions>
61-
using _maybe_eptr_completion_t =
62-
STDEXEC::__if_c<STDEXEC::__nothrow_decay_copyable_results_t<Completions>::value,
63-
STDEXEC::__mset_nil,
64-
STDEXEC::__tuple<STDEXEC::set_error_t, ::std::exception_ptr>>;
65-
66-
template <class Completions>
67-
using _variant_t = STDEXEC::__mset_insert<
68-
STDEXEC::__for_each_completion_signature_t<Completions,
69-
STDEXEC::__decayed_tuple,
70-
STDEXEC::__mset>,
71-
_maybe_eptr_completion_t<Completions>>::template rebind<STDEXEC::__variant>;
42+
using _variant_t = STDEXEC::__mapply_q<STDEXEC::__results_storage, Completions>;
7243

7344
template <class Domain>
7445
struct _env_t
@@ -102,7 +73,7 @@ namespace experimental::execution
10273

10374
STDEXEC_ATTRIBUTE(host, device) void start() noexcept
10475
{
105-
STDEXEC::__visit(_dematerialize_fn{}, *_results_, _rcvr_);
76+
std::as_const(*_results_).__complete(_rcvr_);
10677
}
10778

10879
Rcvr _rcvr_;
@@ -230,7 +201,7 @@ namespace experimental::execution
230201
}
231202

232203
Rcvr _rcvr_;
233-
_variant_t<_child_completions_t> _cache_{STDEXEC::__no_init};
204+
_variant_t<_child_completions_t> _cache_;
234205
STDEXEC::__manual_lifetime<_child_opstate_t> _child_opstate_{};
235206
_fork_opstate_t _fork_opstate_;
236207
};

‎include/exec/just_from.hpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ namespace experimental::execution
165165
{
166166
// Extract the tags from the completion signatures and use them to construct the attributes.
167167
return STDEXEC::__mapply<
168-
STDEXEC::__mtransform<STDEXEC::__q1<STDEXEC::__detail::__tag_of_sig_t>,
168+
STDEXEC::__mtransform<STDEXEC::__q1<STDEXEC::__signature_tag_t>,
169169
STDEXEC::__munique<STDEXEC::__qq<_just_from::_attrs>>>,
170170
completion_signatures>();
171171
}

‎include/exec/sequence/ignore_all_values.hpp‎

Lines changed: 41 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919
#include "../../stdexec/execution.hpp"
2020

2121
// include these after execution.hpp
22+
#include "../../stdexec/__detail/__senders.hpp"
2223
#include "../../stdexec/__detail/__tuple.hpp"
23-
#include "../../stdexec/__detail/__variant.hpp"
2424
#include "../sender_for.hpp"
2525
#include "../sequence_senders.hpp"
2626

@@ -58,7 +58,7 @@ namespace experimental::execution
5858
static_cast<_Receiver&&>(__rcvr));
5959
};
6060

61-
template <class _ResultVariant>
61+
template <class _ResultsStorage>
6262
struct __result_type
6363
{
6464
template <class... _Args>
@@ -83,31 +83,28 @@ namespace experimental::execution
8383
{
8484
STDEXEC::set_value(static_cast<_Receiver&&>(__rcvr));
8585
}
86-
else if constexpr (STDEXEC::__mapply<STDEXEC::__msize, _ResultVariant>::value != 0)
86+
else if constexpr (STDEXEC::__mapply<STDEXEC::__msize, _ResultsStorage>::value != 0)
8787
{
88-
STDEXEC_ASSERT(__result_.index() != __variant_npos);
89-
STDEXEC::__visit(__visit_fn,
90-
static_cast<_ResultVariant&&>(__result_),
91-
static_cast<_Receiver&&>(__rcvr));
88+
static_cast<_ResultsStorage&&>(__result_).__complete(__rcvr);
9289
}
9390
}
9491

95-
_ResultVariant __result_{STDEXEC::__no_init};
92+
_ResultsStorage __result_;
9693
__std::atomic<int> __emplaced_{0};
9794
};
9895

99-
template <class _ItemReceiver, class _ResultVariant>
96+
template <class _ItemReceiver, class _ResultsStorage>
10097
struct __item_operation_base
10198
{
10299
STDEXEC_ATTRIBUTE(no_unique_address) _ItemReceiver __rcvr_;
103-
__result_type<_ResultVariant>* __result_;
100+
__result_type<_ResultsStorage>* __result_;
104101
};
105102

106-
template <class _ItemReceiver, class _ResultVariant>
103+
template <class _ItemReceiver, class _ResultsStorage>
107104
struct __item_receiver
108105
{
109106
using receiver_concept = STDEXEC::receiver_tag;
110-
__item_operation_base<_ItemReceiver, _ResultVariant>* __op_;
107+
__item_operation_base<_ItemReceiver, _ResultsStorage>* __op_;
111108

112109
template <class... _Args>
113110
void set_value([[maybe_unused]] _Args&&... __args) noexcept
@@ -117,7 +114,7 @@ namespace experimental::execution
117114
}
118115

119116
template <class _Error>
120-
requires __variant_emplaceable<_ResultVariant,
117+
requires __variant_emplaceable<_ResultsStorage,
121118
__decayed_tuple<set_error_t, _Error>,
122119
set_error_t,
123120
_Error>
@@ -130,7 +127,9 @@ namespace experimental::execution
130127
}
131128

132129
void set_stopped() noexcept
133-
requires __variant_emplaceable<_ResultVariant, __decayed_tuple<set_stopped_t>, set_stopped_t>
130+
requires __variant_emplaceable<_ResultsStorage,
131+
__decayed_tuple<set_stopped_t>,
132+
set_stopped_t>
134133
&& __callable<set_stopped_t, _ItemReceiver>
135134
{
136135
// stop without error
@@ -144,15 +143,15 @@ namespace experimental::execution
144143
}
145144
};
146145

147-
template <class _Sender, class _ItemReceiver, class _ResultVariant>
148-
struct __item_operation : __item_operation_base<_ItemReceiver, _ResultVariant>
146+
template <class _Sender, class _ItemReceiver, class _ResultsStorage>
147+
struct __item_operation : __item_operation_base<_ItemReceiver, _ResultsStorage>
149148
{
150-
using __base_t = __item_operation_base<_ItemReceiver, _ResultVariant>;
151-
using __item_receiver_t = __item_receiver<_ItemReceiver, _ResultVariant>;
149+
using __base_t = __item_operation_base<_ItemReceiver, _ResultsStorage>;
150+
using __item_receiver_t = __item_receiver<_ItemReceiver, _ResultsStorage>;
152151

153-
__item_operation(__result_type<_ResultVariant>* __parent,
154-
_Sender&& __sndr,
155-
_ItemReceiver __rcvr)
152+
__item_operation(__result_type<_ResultsStorage>* __parent,
153+
_Sender&& __sndr,
154+
_ItemReceiver __rcvr)
156155
noexcept(__nothrow_decay_copyable<_ItemReceiver>
157156
&& __nothrow_connectable<_Sender, __item_receiver_t>)
158157
: __base_t{static_cast<_ItemReceiver&&>(__rcvr), __parent}
@@ -167,18 +166,18 @@ namespace experimental::execution
167166
connect_result_t<_Sender, __item_receiver_t> __op_;
168167
};
169168

170-
template <class _Sender, class _ResultVariant>
169+
template <class _Sender, class _ResultsStorage>
171170
struct __item_sender
172171
{
173172
using sender_concept = STDEXEC::sender_tag;
174173
using completion_signatures = STDEXEC::completion_signatures<set_value_t(), set_stopped_t()>;
175174

176175
template <class _Self, class _Receiver>
177176
using __operation_t =
178-
__item_operation<__copy_cvref_t<_Self, _Sender>, _Receiver, _ResultVariant>;
177+
__item_operation<__copy_cvref_t<_Self, _Sender>, _Receiver, _ResultsStorage>;
179178

180179
template <class _Receiver>
181-
using __item_receiver_t = __item_receiver<_Receiver, _ResultVariant>;
180+
using __item_receiver_t = __item_receiver<_Receiver, _ResultsStorage>;
182181

183182
template <__decays_to<__item_sender> _Self,
184183
STDEXEC::receiver_of<completion_signatures> _Receiver>
@@ -192,23 +191,23 @@ namespace experimental::execution
192191
}
193192
STDEXEC_EXPLICIT_THIS_END(connect)
194193

195-
_Sender __sender_;
196-
__result_type<_ResultVariant>* __parent_;
194+
_Sender __sender_;
195+
__result_type<_ResultsStorage>* __parent_;
197196
};
198197

199-
template <class _Receiver, class _ResultVariant>
200-
struct __operation_base : __result_type<_ResultVariant>
198+
template <class _Receiver, class _ResultsStorage>
199+
struct __operation_base : __result_type<_ResultsStorage>
201200
{
202201
STDEXEC_IMMOVABLE_NO_UNIQUE_ADDRESS
203202
_Receiver __rcvr_;
204203
};
205204

206-
template <class _Receiver, class _ResultVariant>
205+
template <class _Receiver, class _ResultsStorage>
207206
struct __receiver
208207
{
209208
using receiver_concept = STDEXEC::receiver_tag;
210209

211-
constexpr explicit __receiver(__operation_base<_Receiver, _ResultVariant>* __op) noexcept
210+
constexpr explicit __receiver(__operation_base<_Receiver, _ResultsStorage>* __op) noexcept
212211
: __op_{__op}
213212
{}
214213

@@ -218,7 +217,7 @@ namespace experimental::execution
218217
template <sender _Item>
219218
[[nodiscard]]
220219
auto set_next(_Item&& __item) & noexcept(__nothrow_decay_copyable<_Item>)
221-
-> __item_sender<__decay_t<_Item>, _ResultVariant>
220+
-> __item_sender<__decay_t<_Item>, _ResultsStorage>
222221
{
223222
return {static_cast<_Item&&>(__item), __op_};
224223
}
@@ -245,32 +244,30 @@ namespace experimental::execution
245244
}
246245

247246
private:
248-
__operation_base<_Receiver, _ResultVariant>* __op_;
247+
__operation_base<_Receiver, _ResultsStorage>* __op_;
249248
};
250249

251-
template <class _Sigs>
252-
using __result_variant_ = __transform_reduce_completion_signatures_t<
253-
_Sigs,
254-
__mconst<__mlist<>>::__f,
255-
__mcompose_q<__mlist, __mbind_front_q<__decayed_tuple, set_error_t>::__f>::__f,
256-
__mlist<__tuple<set_stopped_t>>,
257-
__mconcat<__qq<__variant>>::__f>;
250+
template <class _Signature>
251+
using __is_set_value_signature_t =
252+
__mbool<__same_as<set_value_t, __signature_tag_t<_Signature>>>;
258253

254+
// Storage for the non-set_value completions
259255
template <class _Sender, class _Env>
260256
using __result_variant_t =
261-
__result_variant_<__sequence_completion_signatures_of_t<_Sender, _Env>>;
257+
__mapply<__mremove_if<__q1<__is_set_value_signature_t>, __qq<__results_storage>>,
258+
__sequence_completion_signatures_of_t<_Sender, _Env>>;
262259

263260
template <class _Sender, class _Receiver>
264261
struct __operation
265262
: __operation_base<_Receiver, __result_variant_t<_Sender, env_of_t<_Receiver>>>
266263
{
267-
using _ResultVariant = __result_variant_t<_Sender, env_of_t<_Receiver>>;
268-
using __base_type = __operation_base<_Receiver, _ResultVariant>;
269-
using __receiver_t = __receiver<_Receiver, _ResultVariant>;
264+
using __variant_t = __result_variant_t<_Sender, env_of_t<_Receiver>>;
265+
using __base_t = __operation_base<_Receiver, __variant_t>;
266+
using __receiver_t = __receiver<_Receiver, __variant_t>;
270267

271268
explicit __operation(_Sender&& __sndr, _Receiver __rcvr)
272269
noexcept(__nothrow_subscribable<_Sender, __receiver_t>)
273-
: __base_type{{}, static_cast<_Receiver&&>(__rcvr)}
270+
: __base_t{{}, static_cast<_Receiver&&>(__rcvr)}
274271
, __op_{exec::subscribe(static_cast<_Sender&&>(__sndr), __receiver_t{this})}
275272
{}
276273

‎include/exec/sequence/merge_each.hpp‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -238,13 +238,8 @@ namespace experimental::execution
238238
void start() & noexcept
239239
{
240240
// emit delayed error into the sequence
241-
STDEXEC::__visit(
242-
[this](auto&& __error) noexcept
243-
{
244-
STDEXEC::set_error(static_cast<_ErrorReceiver&&>(__rcvr_),
245-
static_cast<decltype(__error)&&>(__error));
246-
},
247-
static_cast<_ErrorStorage&&>(*__op_->__error_storage_));
241+
STDEXEC::__visit(STDEXEC::__mk_completion_fn(STDEXEC::set_error, __rcvr_),
242+
static_cast<_ErrorStorage&&>(*__op_->__error_storage_));
248243
}
249244

250245
_ErrorReceiver __rcvr_;

0 commit comments

Comments
 (0)