1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
#pragma once
#include <library/cpp/threading/future/future.h>
#include <coroutine>
template <typename... Args>
struct std::coroutine_traits<NThreading::TFuture<void>, Args...> {
struct promise_type {
NThreading::TFuture<void> get_return_object() noexcept {
return NThreading::TFuture<void>(State_);
}
struct TFinalSuspend {
bool await_ready() noexcept { return false; }
void await_resume() noexcept { /* never called */ }
void await_suspend(std::coroutine_handle<promise_type> self) noexcept {
auto state = std::move(self.promise().State_);
// We must destroy the coroutine before running callbacks
// This will make sure argument copies are destroyed before the caller is resumed
self.destroy();
state->RunCallbacks();
}
};
std::suspend_never initial_suspend() noexcept { return {}; }
TFinalSuspend final_suspend() noexcept { return {}; }
void unhandled_exception() {
bool success = State_->TrySetException(std::current_exception(), /* deferCallbacks */ true);
Y_ASSERT(success && "value already set");
}
void return_void() {
bool success = State_->TrySetValue(/* deferCallbacks */ true);
Y_ASSERT(success && "value already set");
}
private:
TIntrusivePtr<NThreading::NImpl::TFutureState<void>> State_{new NThreading::NImpl::TFutureState<void>()};
};
};
template <typename T, typename... Args>
struct std::coroutine_traits<NThreading::TFuture<T>, Args...> {
struct promise_type {
NThreading::TFuture<T> get_return_object() noexcept {
return NThreading::TFuture<T>(State_);
}
struct TFinalSuspend {
bool await_ready() noexcept { return false; }
void await_resume() noexcept { /* never called */ }
void await_suspend(std::coroutine_handle<promise_type> self) noexcept {
auto state = std::move(self.promise().State_);
// We must destroy the coroutine before running callbacks
// This will make sure argument copies are destroyed before the caller is resumed
self.destroy();
state->RunCallbacks();
}
};
std::suspend_never initial_suspend() noexcept { return {}; }
TFinalSuspend final_suspend() noexcept { return {}; }
void unhandled_exception() {
bool success = State_->TrySetException(std::current_exception(), /* deferCallbacks */ true);
Y_ASSERT(success && "value already set");
}
void return_value(auto&& val) {
bool success = State_->TrySetValue(std::forward<decltype(val)>(val), /* deferCallbacks */ true);
Y_ASSERT(success && "value already set");
}
private:
TIntrusivePtr<NThreading::NImpl::TFutureState<T>> State_{new NThreading::NImpl::TFutureState<T>()};
};
};
namespace NThreading {
template <typename T, bool Extracting = false>
struct TFutureAwaitable {
NThreading::TFuture<T> Future;
TFutureAwaitable(const NThreading::TFuture<T>& future) noexcept requires (!Extracting)
: Future{future}
{
}
TFutureAwaitable(NThreading::TFuture<T>&& future) noexcept
: Future{std::move(future)}
{
}
bool await_ready() const noexcept {
return Future.IsReady();
}
void await_suspend(auto h) noexcept {
/*
* This library assumes that resume never throws an exception.
* This assumption is made due to the fact that the users of these library in most cases do not need to write their own coroutine handlers,
* and all coroutine handlers provided by the library do not throw exception from resume.
*
* WARNING: do not change subscribe to apply or something other here, creating an extra future state degrades performance.
*/
Future.NoexceptSubscribe(
[h](auto) mutable noexcept {
h();
}
);
}
decltype(auto) await_resume() {
if constexpr (Extracting && !std::is_same_v<T, void>) { // Future<void> has only GetValue()
return Future.ExtractValue();
} else {
return Future.GetValue();
}
}
};
template <typename T>
using TExtractingFutureAwaitable = TFutureAwaitable<T, true>;
} // namespace NThreading
template <typename T>
auto operator co_await(const NThreading::TFuture<T>& future) noexcept {
return NThreading::TFutureAwaitable{future};
}
template <typename T>
auto operator co_await(NThreading::TFuture<T>&& future) noexcept {
// Not TExtractongFutureAwaitable, because TFuture works like std::shared_future.
// auto value = co_await GetCachedFuture();
// If GetCachedFuture stores a future in some cache and returns its copies,
// then subsequent uses of co_await will return a moved-from value.
return NThreading::TFutureAwaitable{std::move(future)};
}
namespace NThreading {
template <typename T>
auto AsAwaitable(const NThreading::TFuture<T>& fut) noexcept {
return TFutureAwaitable(fut);
}
template <typename T>
auto AsExtractingAwaitable(NThreading::TFuture<T>&& fut) noexcept {
return TExtractingFutureAwaitable<T>(std::move(fut));
}
} // namespace NThreading
|
