Skip to content

Commit 69df6dc

Browse files
janondrusekericnieblerispeterslehecka
committed
task<> scheduler affinity (#290)
This commit implements scheduler affinity -- aka "sticky" scheduling -- in `unifex::task<>`. The idea is that it is impossible for a child operation to cause the current coroutine to resume on the wrong execution context. * `task<>`-based coroutines track and propagate the current scheduler * `at_coroutine_exit` remembers current scheduler from when the cleanup action is scheduled * `schedule` always returns an instance of `sender_for<schedule, the_sender>`, which is also a `scheduler_provider` * scheduler affinity when co_await-ing senders in a `task<>`-returning coroutine * scheduler affinity when co_await-ing awaitables in a `task<>`-returning coroutine * awaitables and senders that are `blocking_kind::always_inline` don't get a thunk * More senders and awaitables support compile-time blocking queries * `co_await schedule(sched)` is magic in a `task<>`-returning coroutine: it changes execution context and schedules a cleanup action to transition back to the original scheduler Move implementation of special co_await behavior of scheduler senders out of task.hpp Hoist untyped RAII containers for coroutine_handle<> out of task<> and its awaiter (#329) While looking at the binary size impact of adopting coroutines with `unifex::task<>`, I noticed that a number of operations on `coroutine_handle<T>` are expressed in `unifex::task<>` as if they depend on `T` when they don't. The consequence is extra code. This diff creates a `coro_holder` class that uniquely owns a `coroutine_handle<>` and makes `unifex::task<>` inherit from it. We technically lose some type safety, but it's still correct by construction. This change saves about 1.5 kilobytes in one of our apps. Similar to the above, I noticed binary duplication due to false template parameter dependencies in `unifex::task<>`'s awaiter type. This diff hoists a non-type-specific RAII container for a `coroutine_handle<>` that stores the handle as a `std::uintptr_t` so that `task`'s awaiter can use the low bit as a dirty flag. This change saves another ~1.5 kilobytes in one of our apps. Fix scheduler affinity (#405) * Fix scheduler affinity We have been storing a `task<>`'s scheduler as an `any_scheduler_ref`, which has proven to be a source of use-after-free bugs. This change switches all the `any_scheduler_ref`s to `any_scheduler`s, fixing the lifetime issues. Make task<>'s thunk-on-resume unstoppable (#495) * Make task<>'s thunk-on-resume unstoppable When awaiting an async Sender that swallows done signals (such as let_done(never_sender{}, just)), the user-level code looks like it swallows done signals: ``` // never cancels co_await let_done(never_sender{}, just); ``` However, `task<>`'s Scheduler affinity implementation transforms the above code into this: ``` co_await typed_via(let_done(never_sender{}, just), <current scheduler>); ``` The `schedule()` operation inside the injected `typed_via` can emit done if the current stop token has had stop requested, leading to very non-obvious cancellation behaviour that can't be worked around. This diff introduces a pair of regression tests that capture the above scenario, and the analogous scenario of awaiting an async Awaitable that completes with done. The next diff will fix these failing tests. * Change task<>'s thunk-on-resume to be unstoppable This diff fixes the broken tests in the previous diff. Respect blocking_kind in `let_value()` (#381) * `let_value()` would always assume `blocking_kind::maybe`, which results in potentially unnecessary reschedule on resumption * replicate `blocking_kind` customization from `finally()` fix `variant_sender` blocking kind (#474) add `unifex::v2::async_scope` (#463) * simpler than `unifex::v1::async_scope` (`nest()` and `join()`) * does not support cancellation fixing linter error (#414) move deduction guide to namespace scope for gcc-10 in scheduler concept, check copy_constructability after requiring call to schedule() work around gcc-10 bugs avoid warning about missing braces in initializer back out change to awaiter_type_t Co-authored-by: Eric Niebler <eniebler@boost.org> Co-authored-by: Ian Petersen <ispeters@gmail.com> Co-authored-by: Ondrej Lehecka <lehecka@fb.com>
1 parent 63ce7c6 commit 69df6dc

39 files changed

+1547
-304
lines changed

include/unifex/allocate.hpp

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include <unifex/sender_concepts.hpp>
2323
#include <unifex/tag_invoke.hpp>
2424
#include <unifex/bind_back.hpp>
25+
#include <unifex/blocking.hpp>
2526

2627
#include <memory>
2728
#include <type_traits>
@@ -105,6 +106,10 @@ namespace _alloc {
105106
static_cast<Self&&>(s).sender_, (Receiver &&) r};
106107
}
107108

109+
friend constexpr auto tag_invoke(tag_t<unifex::blocking>, const type& self) noexcept {
110+
return blocking(self.sender_);
111+
}
112+
108113
Sender sender_;
109114
};
110115
} // namespace _alloc

include/unifex/any_scheduler.hpp

Lines changed: 48 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,10 @@ struct _with<CPOs...>::any_scheduler {
198198
return _sender{this};
199199
}
200200

201+
type_index type() const noexcept {
202+
return _get_type_index(impl_);
203+
}
204+
201205
friend _equal_to_fn;
202206
friend bool operator==(const any_scheduler& left, const any_scheduler& right) noexcept {
203207
return _equal_to(left.impl_, right);
@@ -210,15 +214,44 @@ struct _with<CPOs...>::any_scheduler {
210214
any_scheduler_impl<CPOs...> impl_;
211215
};
212216

213-
template <typename... ReceiverCPOs>
214-
using any_scheduler_ref_impl = any_ref_t<_schedule_and_connect<ReceiverCPOs...>>;
217+
template <typename... CPOs>
218+
using any_scheduler_ref_impl =
219+
any_ref_t<
220+
_schedule_and_connect<CPOs...>,
221+
_get_type_index,
222+
overload<bool(const this_&, const any_scheduler_ref<CPOs...>&) noexcept>(_equal_to)>;
223+
224+
#if defined(__GLIBCXX__)
225+
template <typename>
226+
inline constexpr bool _is_tuple = false;
227+
228+
template <typename... Ts>
229+
inline constexpr bool _is_tuple<std::tuple<Ts...>> = true;
230+
231+
template <typename... Ts>
232+
inline constexpr bool _is_tuple<std::tuple<Ts...> const> = true;
233+
#endif
215234

216235
template <typename... CPOs>
217236
struct _with<CPOs...>::any_scheduler_ref {
237+
#if !defined(__GLIBCXX__)
218238
template (typename Scheduler)
219-
(requires (!same_as<const Scheduler, const any_scheduler_ref>) AND scheduler<Scheduler>)
239+
(requires (!same_as<const Scheduler, const any_scheduler_ref>) AND
240+
scheduler<Scheduler>)
220241
/* implicit */ any_scheduler_ref(Scheduler& sched) noexcept
221242
: impl_(sched) {}
243+
#else
244+
// Under-constrained implicit tuple converting constructor from a
245+
// single argument doesn't exclude instances of the tuple type
246+
// itself, so it is considered for copy/move constructors, leading
247+
// to constraint recursion with the any_scheduler_ref constructor
248+
// below.
249+
template (typename Scheduler)
250+
(requires (!same_as<const Scheduler, const any_scheduler_ref>) AND
251+
(!_is_tuple<Scheduler>) AND scheduler<Scheduler>)
252+
/* implicit */ any_scheduler_ref(Scheduler& sched) noexcept
253+
: impl_(sched) {}
254+
#endif
222255

223256
struct _sender {
224257
template <template <class...> class Variant, template <class...> class Tuple>
@@ -257,14 +290,26 @@ struct _with<CPOs...>::any_scheduler_ref {
257290
return _sender{this};
258291
}
259292

293+
type_index type() const noexcept {
294+
return _get_type_index(impl_);
295+
}
296+
297+
// Shallow equality comparison by default, for regularity:
260298
friend bool operator==(const any_scheduler_ref& left, const any_scheduler_ref& right) noexcept {
261299
return left.impl_ == right.impl_;
262300
}
263301
friend bool operator!=(const any_scheduler_ref& left, const any_scheduler_ref& right) noexcept {
264302
return !(left == right);
265303
}
266304

305+
// Deep equality comparison:
306+
friend _equal_to_fn;
307+
bool equal_to(const any_scheduler_ref& that) const noexcept {
308+
return _equal_to(impl_, that);
309+
}
310+
267311
private:
312+
268313
any_scheduler_ref_impl<CPOs...> impl_;
269314
};
270315

include/unifex/async_trace.hpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ namespace _async_trace {
103103
return operation<Receiver>{(Receiver &&) r};
104104
}
105105

106-
friend blocking_kind tag_invoke(tag_t<blocking>, const sender&) noexcept {
106+
friend auto tag_invoke(tag_t<blocking>, const sender&) noexcept {
107107
return blocking_kind::always_inline;
108108
}
109109
};

include/unifex/at_coroutine_exit.hpp

Lines changed: 38 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,13 @@
1919
#include <unifex/tag_invoke.hpp>
2020
#include <unifex/await_transform.hpp>
2121
#include <unifex/continuations.hpp>
22+
#include <unifex/scheduler_concepts.hpp>
2223
#include <unifex/stop_token_concepts.hpp>
2324
#include <unifex/unstoppable_token.hpp>
2425
#include <unifex/get_stop_token.hpp>
26+
#include <unifex/inline_scheduler.hpp>
27+
#include <unifex/any_scheduler.hpp>
28+
#include <unifex/blocking.hpp>
2529

2630
#if UNIFEX_NO_COROUTINES
2731
# error "Coroutine support is required to use this header"
@@ -115,11 +119,19 @@ struct _cleanup_promise_base {
115119
}
116120
#endif
117121

118-
friend unstoppable_token tag_invoke(tag_t<get_stop_token>, const _cleanup_promise_base&) noexcept {
122+
friend unstoppable_token
123+
tag_invoke(tag_t<get_stop_token>, const _cleanup_promise_base&) noexcept {
119124
return unstoppable_token{};
120125
}
121126

127+
friend any_scheduler
128+
tag_invoke(tag_t<get_scheduler>, const _cleanup_promise_base& p) noexcept {
129+
return p.sched_;
130+
}
131+
132+
inline static constexpr inline_scheduler _default_scheduler{};
122133
continuation_handle<> continuation_{};
134+
any_scheduler sched_{_default_scheduler};
123135
bool isUnhandledDone_{false};
124136
};
125137

@@ -145,6 +157,15 @@ struct _die_on_done_rec {
145157
UNIFEX_ASSERT(!"A cleanup action tried to cancel. Calling terminate...");
146158
std::terminate();
147159
}
160+
161+
template(typename CPO)
162+
(requires is_receiver_query_cpo_v<CPO> AND
163+
is_callable_v<CPO, const Receiver&>)
164+
friend auto tag_invoke(CPO cpo, const type& p)
165+
noexcept(is_nothrow_callable_v<CPO, const Receiver&>)
166+
-> callable_result_t<CPO, const Receiver&> {
167+
return cpo(p.rec_);
168+
}
148169
};
149170
};
150171

@@ -177,7 +198,7 @@ struct _die_on_done {
177198
_die_on_done_rec_t<Receiver>{(Receiver&&) rec});
178199
}
179200

180-
Sender sender_;
201+
UNIFEX_NO_UNIQUE_ADDRESS Sender sender_;
181202
};
182203
};
183204

@@ -229,7 +250,7 @@ struct _cleanup_promise : _cleanup_promise_base {
229250
return unifex::await_transform(*this, _die_on_done_fn{}((Value&&) value));
230251
}
231252

232-
std::tuple<Ts&...> args_;
253+
UNIFEX_NO_UNIQUE_ADDRESS std::tuple<Ts&...> args_;
233254
};
234255

235256
template <typename... Ts>
@@ -251,14 +272,24 @@ struct [[nodiscard]] _cleanup_task {
251272
}
252273

253274
template <typename Promise>
254-
bool await_suspend(coro::coroutine_handle<Promise> parent) noexcept {
275+
bool await_suspend_impl_(Promise& parent) noexcept {
255276
continuation_.promise().continuation_ =
256-
exchange_continuation(parent.promise(), continuation_);
277+
exchange_continuation(parent, continuation_);
278+
continuation_.promise().sched_ = get_scheduler(parent);
257279
return false;
258280
}
259281

282+
template <typename Promise>
283+
bool await_suspend(coro::coroutine_handle<Promise> parent) noexcept {
284+
return await_suspend_impl_(parent.promise());
285+
}
286+
260287
std::tuple<Ts&...> await_resume() noexcept {
261-
return std::exchange(continuation_, {}).promise().args_;
288+
return std::move(std::exchange(continuation_, {}).promise().args_);
289+
}
290+
291+
friend constexpr auto tag_invoke(tag_t<blocking>, const _cleanup_task&) noexcept {
292+
return blocking_kind::always_inline;
262293
}
263294

264295
private:
@@ -275,7 +306,7 @@ namespace _at_coroutine_exit {
275306
public:
276307
template (typename Action, typename... Ts)
277308
(requires callable<std::decay_t<Action>, std::decay_t<Ts>...>)
278-
_cleanup_task<Ts...> operator()(Action&& action, Ts&&... ts) const {
309+
_cleanup_task<std::decay_t<Ts>...> operator()(Action&& action, Ts&&... ts) const {
279310
return _fn::at_coroutine_exit((Action&&) action, (Ts&&) ts...);
280311
}
281312
} at_coroutine_exit{};

include/unifex/await_transform.hpp

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -85,13 +85,13 @@ struct _awaitable_base<Promise, Value>::type {
8585
struct _rec {
8686
public:
8787
explicit _rec(_expected<Value>* result, coro::coroutine_handle<Promise> continuation) noexcept
88-
: result_(result)
89-
, continuation_(continuation)
88+
: result_(result)
89+
, continuation_(continuation)
9090
{}
9191

9292
_rec(_rec&& r) noexcept
93-
: result_(std::exchange(r.result_, nullptr))
94-
, continuation_(std::exchange(r.continuation_, nullptr))
93+
: result_(std::exchange(r.result_, nullptr))
94+
, continuation_(std::exchange(r.continuation_, nullptr))
9595
{}
9696

9797
template(class... Us)
@@ -185,13 +185,16 @@ struct _awaitable<Promise, Sender>::type
185185
template <typename Promise, typename Sender>
186186
using _as_awaitable = typename _awaitable<Promise, Sender>::type;
187187

188-
inline const struct _fn {
188+
struct _fn {
189189
// Call custom implementation if present.
190190
template(typename Promise, typename Value)
191191
(requires tag_invocable<_fn, Promise&, Value>)
192192
auto operator()(Promise& promise, Value&& value) const
193193
noexcept(is_nothrow_tag_invocable_v<_fn, Promise&, Value>)
194194
-> tag_invoke_result_t<_fn, Promise&, Value> {
195+
static_assert(detail::_awaitable<tag_invoke_result_t<_fn, Promise&, Value>>,
196+
"The return type of a customization of unifex::await_transform() "
197+
"must satisfy the awaitable concept.");
195198
return unifex::tag_invoke(_fn{}, promise, (Value&&)value);
196199
}
197200

@@ -218,7 +221,7 @@ inline const struct _fn {
218221
return (Value&&) value;
219222
}
220223
}
221-
} await_transform {};
224+
};
222225

223226
} // namespace _await_tfx
224227

@@ -231,7 +234,7 @@ inline const struct _fn {
231234
//
232235
// Coroutine promise_types can implement their .await_transform() methods to
233236
// forward to this customisation point to enable use of type customisations.
234-
using _await_tfx::await_transform;
237+
inline constexpr _await_tfx::_fn await_transform {};
235238

236239
} // namespace unifex
237240

include/unifex/blocking.hpp

Lines changed: 67 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,11 @@
2222

2323
namespace unifex {
2424

25-
enum class blocking_kind {
25+
namespace _block {
26+
enum class _enum {
2627
// No guarantees about the timing and context on which the receiver will
2728
// be called.
28-
maybe,
29+
maybe = 0,
2930

3031
// Always completes asynchronously.
3132
// Guarantees that the receiver will not be called on the current thread
@@ -44,8 +45,46 @@ enum class blocking_kind {
4445
always_inline
4546
};
4647

47-
namespace _blocking {
48-
inline const struct _fn {
48+
struct blocking_kind {
49+
template <_enum Kind>
50+
using constant = std::integral_constant<_enum, Kind>;
51+
52+
blocking_kind() = default;
53+
54+
constexpr blocking_kind(_enum kind) noexcept
55+
: value(kind)
56+
{}
57+
58+
template <_enum Kind>
59+
constexpr blocking_kind(constant<Kind>) noexcept
60+
: value(Kind)
61+
{}
62+
63+
constexpr operator _enum() const noexcept {
64+
return value;
65+
}
66+
67+
constexpr _enum operator()() const noexcept {
68+
return value;
69+
}
70+
71+
friend constexpr bool operator==(blocking_kind a, blocking_kind b) noexcept {
72+
return a.value == b.value;
73+
}
74+
75+
friend constexpr bool operator!=(blocking_kind a, blocking_kind b) noexcept {
76+
return a.value != b.value;
77+
}
78+
79+
static constexpr constant<_enum::maybe> maybe {};
80+
static constexpr constant<_enum::never> never {};
81+
static constexpr constant<_enum::always> always {};
82+
static constexpr constant<_enum::always_inline> always_inline {};
83+
84+
_enum value{};
85+
};
86+
87+
struct _fn {
4988
template(typename Sender)
5089
(requires tag_invocable<_fn, const Sender&>)
5190
constexpr auto operator()(const Sender& s) const
@@ -55,12 +94,32 @@ inline const struct _fn {
5594
}
5695
template(typename Sender)
5796
(requires (!tag_invocable<_fn, const Sender&>))
58-
constexpr blocking_kind operator()(const Sender&) const noexcept {
97+
constexpr auto operator()(const Sender&) const noexcept {
98+
return blocking_kind::maybe;
99+
}
100+
};
101+
102+
namespace _cfn {
103+
template <_enum Kind>
104+
static constexpr auto _kind(blocking_kind::constant<Kind> kind) noexcept {
105+
return kind;
106+
}
107+
static constexpr auto _kind(blocking_kind) noexcept {
59108
return blocking_kind::maybe;
60109
}
61-
} blocking{};
62-
} // namespace _blocking
63-
using _blocking::blocking;
110+
111+
template <typename T>
112+
constexpr auto cblocking() noexcept {
113+
using blocking_t = remove_cvref_t<decltype(_fn{}(UNIFEX_DECLVAL(T&)))>;
114+
return _cfn::_kind(blocking_t{});
115+
}
116+
}
117+
118+
} // namespace _block
119+
120+
inline constexpr _block::_fn blocking {};
121+
using _block::_cfn::cblocking;
122+
using _block::blocking_kind;
64123

65124
} // namespace unifex
66125

0 commit comments

Comments
 (0)