forked from facebook/watchman
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Future.h
460 lines (402 loc) · 15 KB
/
Future.h
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
/* Copyright 2017-present Facebook, Inc.
* Licensed under the Apache License, Version 2.0 */
#pragma once
#include <condition_variable>
#include <functional>
#include <memory>
#include <mutex>
#include <vector>
#include "Result.h"
namespace watchman {
// An executor is a very light abstraction over something that can run
// some function. We use this to perform the WaitableResult callback
// in some other context (eg: ThreadPool). In practice, this is used
// to move the execution context for a .then call to a worker thread.
struct Executor {
virtual ~Executor() = default;
virtual void run(std::function<void()>&& func) = 0;
};
// WaitableResult<T> encapsulates a Result<T> and allows waiting and notifying
// and interested party. You are not expected to create an instance of this
// class directly; it is used as the shared state between the Promise and
// Future classes.
template <typename T>
class WaitableResult : public std::enable_shared_from_this<WaitableResult<T>> {
static_assert(!std::is_same<T, void>::value, "use Unit instead of void");
public:
WaitableResult() = default;
explicit WaitableResult(Result<T>&& t) : result_(std::move(t)) {}
WaitableResult(const WaitableResult&) = delete;
WaitableResult(WaitableResult&&) = delete;
// Assign to the underlying Result<T>.
// After assigning, dispatch any associated callback and notify any waiters
template <typename U>
void assign(U&& value) {
std::unique_lock<std::mutex> lock(mutex_);
result_ = Result<T>(std::forward<U>(value));
maybeCallback(std::move(lock));
}
// Get a reference to the enclosed result
Result<T>& result() {
return result_;
}
// Wait until the result is no longer empty
void wait() const {
std::unique_lock<std::mutex> lock(mutex_);
condition_.wait(lock, [&] { return !result_.empty(); });
}
// Wait for up to the specified duration.
// Returns true as soon as the result is no longer empty.
// Returns false if the result is still empty after duration has
// passed.
template <class Rep, class Period>
bool wait_for(const std::chrono::duration<Rep, Period>& duration) {
std::unique_lock<std::mutex> lock(mutex_);
return condition_.wait_for(
lock, duration, [&] { return !result_.empty(); });
}
// Associate a callback with the result.
// This is intended for use by internal plumbing and a casual
// user of a Future should not call this method; instead,
// you want to use Future::then().
//
// The callback will be dispatched when the assign() method
// is called. If the assign() method was called prior to
// setCallback(), it will be called by setCallback().
//
// It is possible for the callback to fire twice for the
// same Future if setExecutor() was used to assign an
// executor that runs immediately, and if the callback
// throws an exception. If this happens, callback will be
// called a second time with the Result containing the
// exception that it called previously. Throwing an
// exception in that context will bubble up to the caller
// of assign().
//
// If you stick to Future::then(), you don't need to
// worry about this.
template <typename Func>
void setCallback(Func&& func) {
std::unique_lock<std::mutex> lock(mutex_);
callback_ = std::forward<Func>(func);
if (!result_.empty()) {
maybeCallback(std::move(lock));
}
}
// Change the executor associated with the future
void setExecutor(Executor* executor) {
std::unique_lock<std::mutex> lock(mutex_);
executor_ = executor;
}
private:
Result<T> result_;
mutable std::condition_variable condition_;
mutable std::mutex mutex_;
std::function<void(Result<T>&&)> callback_;
Executor* executor_{nullptr};
// If a callback is set, call it.
// Then notify any waiters that the result is available.
void maybeCallback(std::unique_lock<std::mutex>&& lock) {
if (result_.empty()) {
return;
}
if (callback_) {
// Ensure that we are kept alive while we dispatch the callback
auto scope_guard = this->shared_from_this();
// Steal the callback
std::function<void(Result<T> &&)> func;
std::swap(func, callback_);
// For safety, ensure that we are unlocked while calling the callback.
// While the intent is that the executor run the callback in a
// different thread context, it may choose to run something
// immediately.
lock.unlock();
if (executor_) {
try {
// Unfortunately, have to make a copy of func here in
// order to have sane exception handling
executor_->run([scope_guard, this, func] {
func(std::move(result_));
});
condition_.notify_all();
return;
} catch (const std::exception& exc) {
// We get here if executor_->run() threw an exception.
// This is most likely to happen if the thread pool is
// full, but we can't make any assumptions of the nature
// of the exception; it may have simply run the callback
// immediately, and we're just seeing the exception from
// the callback here.
// We're really only capturing this so that we can
// propagate thread pool errors through the exception
// chain.
// Replace the current value with the exception, and we'll
// dispatch the exception to the callback below.
result_ = Result<T>(std::current_exception());
}
}
// We don't catch and propagate exceptions that this function
// throws (like we do in the executor case above), because
// the callback is supposed to manage exceptions and do
// the right thing for itself.
func(std::move(result_));
}
condition_.notify_all();
}
};
template <typename T>
class Promise;
template <typename T>
class Future;
// A little bit of helper glue for unwrapping Future<Future<T>>
template <typename T>
struct isFuture : std::false_type {
using Inner = typename Unit::Lift<T>::type;
};
template <typename T>
struct isFuture<Future<T>> : std::true_type {
using Inner = T;
};
// Extracts the return type of a functor call
template <typename F, typename... Args>
using resultOf = decltype(std::declval<F>()(std::declval<Args>()...));
// The Future is the client side of the Promise/Future pair.
// A Promise can return a single instance of a Future. The two
// are linked by a shared WaitableResult object.
// A Future is only safe for access from a single thread at a time.
template <typename T>
class Future {
static_assert(!std::is_same<T, void>::value, "use Unit instead of void");
public:
using value_type = T;
Future() noexcept = default;
// Moveable
Future(Future&&) noexcept = default;
Future& operator=(Future&&) noexcept = default;
// Not copyable
Future(const Future&) = delete;
Future& operator=(const Future&) = delete;
// makeFuture() uses this to build ready Future instances
explicit Future(std::shared_ptr<WaitableResult<T>> state) : state_(state) {}
// Block until the associated Promise is fulfilled
void wait() const {
if (!state_) {
throw std::logic_error("Future has no shared state");
}
state_->wait();
}
// Wait for up to the specified duration for the associated Promise to
// be fulfilled.
// Returns true as soon as the Promise is fulfilled.
// Returns false if the Promise was not fulfilled within the specified
// duration.
template <class Rep, class Period>
bool wait_for(const std::chrono::duration<Rep, Period>& duration) const {
if (!state_) {
throw std::logic_error("Future has no shared state");
}
return state_->wait_for(duration);
}
// Returns true if the associated Promise has been fulfilled.
bool isReady() const {
return this->wait_for(std::chrono::milliseconds(0));
}
// Waits for the Promise to be fulfilled, then returns a reference
// to the value in the promise.
// If the Result holds an error this will cause the error to be
// thrown.
T& get() {
return result().value();
}
// As get() above, but returns a const reference.
const T& get() const {
return result().value();
}
// Waits for the Promise to be fulfilled, then returns a reference
// to the Result in the promise.
Result<T>& result() {
wait();
return state_->result();
}
// Waits for the Promise to be fulfilled, then returns a const reference
// to the Result in the promise.
const Result<T>& result() const {
wait();
return state_->result();
}
// Chain a future together with some action to happen once
// it is ready.
// f.then([](Result<T>&& result) { return something; })
// This handles the case where something is not a Future<>
template <typename Func>
typename std::enable_if<
!isFuture<resultOf<Func, Result<T>&&>>::value,
Future<typename isFuture<resultOf<Func, Result<T>&&>>::Inner>>::type
then(Func&& func) {
using Ret = typename isFuture<resultOf<Func, Result<T>&&>>::Inner;
struct thenState {
Promise<Ret> promise;
Func func;
thenState(Func&& func) : func(std::forward<Func>(func)) {}
};
auto state = std::make_shared<thenState>(std::forward<Func>(func));
auto result = state->promise.getFuture();
state_->setCallback([state](Result<T>&& result) {
state->promise.setResult(makeResultWith(
[&]() mutable { return state->func(std::move(result)); }));
});
return result;
}
// Chain a future together with some action to happen once
// it is ready.
// f.then([](Result<T>&& result) { return something; })
// This handles the case where something is a Future<> and
// unwraps it so that the result of .then is Future<> rather
// than Future<Future<>>.
template <typename Func>
typename std::enable_if<
isFuture<resultOf<Func, Result<T>&&>>::value,
Future<typename isFuture<resultOf<Func, Result<T>&&>>::Inner>>::type
then(Func&& func) {
using Ret = resultOf<Func, Result<T>&&>;
using InnerRet = typename isFuture<Ret>::Inner;
struct thenState {
Promise<InnerRet> promise;
Func func;
thenState(Func&& func) : func(std::forward<Func>(func)) {}
};
auto state = std::make_shared<thenState>(std::forward<Func>(func));
auto result = state->promise.getFuture();
state_->setCallback([state](Result<T>&& res) {
try {
auto future = state->func(std::move(res));
future.setCallback([state](Result<InnerRet>&& res2) {
state->promise.setResult(std::move(res2));
});
} catch (const std::exception& exc) {
state->promise.setException(std::current_exception());
}
});
return result;
}
// Exposes setCallback for .then when unwrapping Future<Future<>>.
// Since Future<Future<T>> is a different class from Future<T>,
// we have to make this public.
// You probably want to use .then() and not this directly.
template <typename Func>
void setCallback(Func&& func) {
state_->setCallback(std::forward<Func>(func));
}
// Returns a future with its execution context switched to the provided
// executor. A subsequent then() call will be dispatched by that
// executor. Note that, depending on timing, this sequence:
// makeFuture().via(exec).then(A).then(B)
// will execute A in the context of exec, but may execute B either in
// exec or the current context. If you need to ensure the execution
// context for a then() call, you must precede it with a via() call
// and supply the appropriate executor:
// makeFuture().via(exec).then(A).via(exec).then(B)
Future<T> via(Executor* executor) && {
state_->setExecutor(executor);
return std::move(*this);
}
private:
std::shared_ptr<WaitableResult<T>> state_;
};
// The Promise is the server side of the Promise/Future pair.
template <typename T>
class Promise {
static_assert(!std::is_same<T, void>::value, "use Unit instead of void");
public:
// Default construct to an un-fulfilled Promise
Promise() : state_(std::make_shared<WaitableResult<T>>()) {}
// Moveable
Promise(Promise&&) noexcept = default;
Promise& operator=(Promise&&) noexcept = default;
// Copyable
Promise(const Promise&) = delete;
Promise& operator=(const Promise&) = delete;
// Fulfills the promise with a value of type T.
// Causes any waiters/callbacks associated with the Promise to be
// notified/dispatched.
// It is an error fulfill the same promise multiple times.
void setValue(T&& value) {
setResult(Result<T>(std::forward<T>(value)));
}
// Fulfills the promise with a Result<T>.
// Causes any waiters/callbacks associated with the Promise to be
// notified/dispatched.
// It is an error fulfill the same promise multiple times.
void setResult(Result<T>&& result) {
if (assigned_) {
throw std::logic_error("Promise already fulfilled");
}
assigned_ = true;
state_->assign(std::move(result));
}
// Fulfills the promise with an exception.
// Causes any waiters/callbacks associated with the Promise to be
// notified/dispatched.
// It is an error fulfill the same promise multiple times.
void setException(std::exception_ptr exc) {
setResult(Result<T>(exc));
}
// Returns a Future associated with this Promise. The Future
// allows a client to wait for the results.
// It is an error to call getFuture() multiple times.
Future<T> getFuture() {
if (gotFuture_) {
throw std::logic_error("Future already obtained");
}
gotFuture_ = true;
return Future<T>(state_);
}
private:
std::shared_ptr<WaitableResult<T>> state_;
bool gotFuture_{false};
bool assigned_{false};
};
// Convert a Result<T> to a Future<T> that is ready immediately
template <typename T>
Future<T> makeFuture(Result<T>&& t) {
return Future<T>(std::make_shared<WaitableResult<T>>(std::move(t)));
}
// Helper for making an already fulfilled Future from a value;
// auto-deduces the Value type.
template <typename T>
Future<typename std::decay<T>::type> makeFuture(T&& t) {
return makeFuture(Result<typename std::decay<T>::type>(std::forward<T>(t)));
}
inline Future<Unit> makeFuture() {
return makeFuture(Result<Unit>(Unit{}));
}
// Yields a Future holding a vector<Result<T>> for each of the input futures
template <typename InputIterator>
Future<std::vector<Result<
typename std::iterator_traits<InputIterator>::value_type::value_type>>>
collectAll(InputIterator first, InputIterator last) {
using T =
typename std::iterator_traits<InputIterator>::value_type::value_type;
struct CollectAll {
// Pre-allocate enough room for all results to avoid needing to
// synchronize when the callback assigns everything
CollectAll(size_t n) : results(n) {}
// The destructor triggers the assignment and fulfillment of the Promise
~CollectAll() {
p.setValue(std::move(results));
}
Promise<std::vector<Result<T>>> p;
std::vector<Result<T>> results;
};
auto state = std::make_shared<CollectAll>(std::distance(first, last));
size_t i = 0;
while (first != last) {
first->setCallback([state, i](Result<T>&& result) {
state->results[i] = std::move(result);
});
++i;
++first;
}
return state->p.getFuture();
}
}