| // Copyright 2018 The Fuchsia Authors. All rights reserved. | |
| // Use of this source code is governed by a BSD-style license that can be | |
| // found in the LICENSE file. | |
| #ifndef SRC_MODULAR_LIB_ASYNC_CPP_FUTURE_H_ | |
| #define SRC_MODULAR_LIB_ASYNC_CPP_FUTURE_H_ | |
| #include <lib/async/cpp/task.h> | |
| #include <lib/async/default.h> | |
| #include <lib/fit/function.h> | |
| #include <lib/syslog/cpp/macros.h> | |
| #include <memory> | |
| #include <string> | |
| #include <type_traits> | |
| #include <utility> | |
| #include <vector> | |
| #include "src/lib/fxl/functional/apply.h" | |
| #include "src/lib/fxl/macros.h" | |
| #include "src/lib/fxl/memory/ref_counted.h" | |
| #include "src/lib/fxl/memory/ref_ptr.h" | |
| #include "src/lib/fxl/memory/weak_ptr.h" | |
| namespace modular { | |
| // # Futures | |
| // | |
| // A *future* is an object representing the eventual value of an asynchronous | |
| // operation. They are a useful complement to callback functions or lambdas | |
| // because they are _composable_: asynchronous operations can be sequentially | |
| // executed, and an async operation's result can be passed to another async | |
| // operation, like a Unix pipeline. | |
| // | |
| // To use a future: | |
| // | |
| // 1. A *producer*, typically an async operation, creates a Future with | |
| // Future<ResultType>::Create(). | |
| // 2. The producer starts its async operation (e.g. a network request or disk | |
| // read). | |
| // 3. The producer synchronously returns the Future to a *consumer*. | |
| // 4. A consumer attaches a *callback* lambda to the Future using Then(). (The | |
| // callback can be attached to the future any time after the future is | |
| // created, before or after the async operation is finished.) | |
| // 5. Some time later, when the producer's async operation is finished, the | |
| // producer *completes* the future with a *result* using Complete(result). | |
| // |result| is 0 to N movable or copyable values, e.g. Complete(), | |
| // Complete(value), Complete(value1, value2, ...). | |
| // 6. The consumer's callback is invoked after the future is completed, with the | |
| // completed result passed as zero or more parameters to the callback. | |
| // | |
| // The following example shows a simple use case: | |
| // | |
| // Producer: | |
| // | |
| // FuturePtr<Bytes> MakeNetworkRequest(NetworkRequest& request) { | |
| // auto f = Future<Bytes>::Create("NetworkRequest"); // a "trace_name" that's | |
| // // logged when things go | |
| // // wrong | |
| // auto network_request_callback = [f] (Bytes bytes) { | |
| // f->Complete(bytes); | |
| // }; | |
| // PerformAsyncNetworkRequest(request, network_request_callback); | |
| // return f; | |
| // } | |
| // | |
| // Client: | |
| // | |
| // FuturePtr<Bytes> f = MakeNetworkRequest(); | |
| // f->Then([] (Bytes bytes) { | |
| // ProcessBytes(bytes); | |
| // }); | |
| // | |
| // ## Chaining and sequencing futures | |
| // | |
| // Futures can be composed; this is commonly called "chaining". Methods that | |
| // attach callbacks, such as Then(), will return another Future: the returned | |
| // Future is completed once the previous callback has finished executing. For | |
| // example: | |
| // | |
| // Client: | |
| // | |
| // ShowProgressSpinner(true); | |
| // FuturePtr<Bytes> f = MakeNetworkRequest(request); | |
| // f->AsyncMap([] (Bytes zipped_bytes) { | |
| // // Use AsyncMap() if your callback wants to return another Future. | |
| // FuturePtr<Bytes> f = UnzipInBackground(zipped_bytes); | |
| // return f; | |
| // })->Map([] (Bytes unzipped_bytes) { | |
| // // Use Map() if your callback returns a non-future: the callback's return | |
| // // value will be wrapped up into the returned Future. | |
| // JPEGImage image = DecodeImageSynchronously(unzipped_bytes); | |
| // return image; | |
| // })->AsyncMap([] (JPEGImage image) { | |
| // FuturePtr<> f = UpdateUIAsynchronously(image); | |
| // return f; | |
| // })->Then([] { | |
| // // Use Then() if your callback returns void. Note that Then() still returns | |
| // // a Future<>, which will be completed only when your callback finishes | |
| // // executing. | |
| // ShowProgressSpinner(false); | |
| // }); | |
| // | |
| // ## Memory Management & Ownership | |
| // | |
| // FuturePtr is a fxl::RefPtr, which is a smart pointer similar to | |
| // std::shared_ptr that holds a reference count (refcount). When you call | |
| // Future::Create(), you are expected to maintain a reference to it. When the | |
| // Future is deleted, its result and callbacks are also deleted. | |
| // | |
| // Each method documents how it affects the future's refcount. To summarize: | |
| // | |
| // * Calling Then() on a future does not affect its refcount. This applies to | |
| // all methods that returns a chained future, such as AsyncMap() and Map(). | |
| // * However, calling Then() on a future will cause that future to maintain a | |
| // reference to the returned chained future. So, you do not need to maintain a | |
| // reference to the returned future. | |
| // * Calling Complete() does not affect the future's refcount. | |
| // * Unlike Complete(), the closure returned by Completer() _does own_ the | |
| // future, so you do not need to maintain a reference to the future after | |
| // calling Completer(). (You do need to maintain a reference to the closure, | |
| // however.) | |
| // * Wait(futures) returns a future that every future in |futures| owns, so you | |
| // do not need to maintain a reference to the returned future. The callback | |
| // attached to each future in |futures| will also keep a reference to | |
| // themselves, so that if a future that is Wait()ed on otherwise goes out of | |
| // scope, the future itself is kept alive. | |
| // | |
| // See each method's documentation for more details on memory management. | |
| // | |
| // ## Use Weak*() variants to cancel callback chains | |
| // | |
| // "Weak" variants exist for all chain/sequence methods (WeakThen(), | |
| // WeakConstThen(), WeakMap() and WeakAsyncMap()). These are almost identical | |
| // to their non-weak counterparts but take an fxl::WeakPtr<T> as a first | |
| // argument. If, at callback invocation time, the WeakPtr is invalid, execution | |
| // will halt and no future further down the chain will be executed. | |
| // | |
| // Example: | |
| // | |
| // FuturePtr<> f = MakeFuture(); | |
| // f->WeakThen(weak_ptr_factory.GetWeakPtr(), [] { | |
| // FX_LOGS(INFO) << "This won't execute"; | |
| // })->Then([] { | |
| // FX_LOGS(INFO) << "Neither will this"; | |
| // }); | |
| // weak_ptr_factory.InvalidateWeakPtrs(); | |
| // f->Complete(); | |
| // | |
| // ## Use Wait() to synchronize on multiple futures | |
| // | |
| // If multiple futures are running, use the Wait() function to create a | |
| // Future that completes when all the futures passed to it are completed: | |
| // | |
| // FuturePtr<Bytes> f1 = MakeNetworkRequest(request1); | |
| // FuturePtr<Bytes> f2 = MakeNetworkRequest(request2); | |
| // FuturePtr<Bytes> f3 = MakeNetworkRequest(request3); | |
| // Wait<Future<>>("Network requests", {f1, f2, f3})->Then([] { | |
| // AllNetworkRequestsAreComplete(); | |
| // }); | |
| // | |
| // See the Wait() function documentation for more details. | |
| // | |
| // ## Use Completer() to integrate with functions requiring callbacks | |
| // | |
| // Use the Completer() method to integrate with existing code that uses callback | |
| // functions. Completer() returns a fit::function<void(Result)> that, when | |
| // called, calls Complete() on the future. Re-visiting the first example: | |
| // | |
| // Without Completer(): | |
| // | |
| // FuturePtr<Bytes> MakeNetworkRequest(NetworkRequest& request) { | |
| // auto f = Future<Bytes>::Create("NetworkRequest"); | |
| // auto network_request_callback = [f] (Bytes bytes) { | |
| // f->Complete(bytes); | |
| // }; | |
| // PerformAsyncNetworkRequest(request, network_request_callback); | |
| // return f; | |
| // } | |
| // | |
| // With Completer(): | |
| // | |
| // FuturePtr<Bytes> MakeNetworkRequest(NetworkRequest& request) { | |
| // auto f = Future<Bytes>::Create("NetworkRequest"); | |
| // PerformAsyncNetworkRequest(request, f->Completer()); | |
| // return f; | |
| // } | |
| // | |
| // ## Use error values to propagate errors back to consumers | |
| // | |
| // If the future can fail, use a value (or multiple values) that's capable of | |
| // storing both the error and a successful result to propagate the return value | |
| // back to the consumer. For example: | |
| // | |
| // FuturePtr<std::error_code, Bytes> f = MakeNetworkRequest(); | |
| // f->Then([] (std::error_code error, Bytes bytes) { | |
| // if (error) { | |
| // // handle error | |
| // } else { | |
| // // network request was successful | |
| // ProcessBytes(bytes); | |
| // } | |
| // }); | |
| // | |
| // ## fuchsia::modular::Future vs other Futures/Promises | |
| // | |
| // If you are familiar with Futures & Promises in other languages, this Future | |
| // class is intentionally different from others, to better integrate with | |
| // Fuchsia coding patterns: | |
| // | |
| // * NOT THREADSAFE. This will change in the future when thread safety is | |
| // required, but there are no use cases yet. (YAGNI!) | |
| // * Support for multiple result values via variadic template parameters. This | |
| // is required for smooth integration with the fuchsia::modular::Operation | |
| // class. | |
| // * Only a single callback can be set via Then(), since move semantics are used | |
| // so heavily in Fuchsia code. (If multiple callbacks were supported and the | |
| // result is moved, how does one move the result from one callback to | |
| // another?) | |
| // * Multiple callbacks can be set if the callback lambda takes the result via | |
| // const&. In this case, use ConstThen() to attach each callback, rather | |
| // than Then(). ConstThen() calls can also be chained, like Then(). | |
| // * There are no success/error callbacks and control flows: all callbacks are | |
| // "success callbacks". | |
| // * The traditional reason for error callbacks is to convert exceptions into | |
| // error values, but Google C++ style doesn't use exceptions, so error | |
| // callbacks aren't needed. | |
| // * There's some argument that using separate success/error control flow | |
| // paths is beneficial. However, in practice, a lot of client code using | |
| // this pattern don't attach error callbacks, only success callbacks, so | |
| // errors often go unchecked. | |
| // * If error values need to be propagated back to the client, use a dedicated | |
| // error type. (Note that many built-in types may have values that can be | |
| // interpreted as errors, e.g. nullptr, or 0 or -1.) This also forces a | |
| // consumer to inspect the type and check for errors. | |
| // * No cancellation/progress support. Adding cancellation/progress: | |
| // * adds more complexity to the futures implementation, | |
| // * can be implemented on top of a core futures implementation for the few | |
| // cases where they're required, and | |
| // * typically requires extra cancel/progress callbacks, which adds more | |
| // control flow paths. | |
| // * see fxl::CancelableCallback if you need cancellation. | |
| // * No execution contexts (yet). | |
| // * There needs to be a comprehensive story about runloops etc first. | |
| template <typename... Result> | |
| class Future; | |
| template <typename... Result> | |
| using FuturePtr = fxl::RefPtr<Future<Result...>>; | |
| namespace internal { | |
| enum class FutureStatus { | |
| kAwaiting, // not completed | |
| kCompleted, // value available, not yet moved into callback | |
| kConsumed // value moved into callback | |
| }; | |
| // type_traits functions, ported from C++17. | |
| template <typename From, typename To> | |
| constexpr bool is_convertible_v = std::is_convertible<From, To>::value; | |
| template <class T> | |
| constexpr bool is_void_v = std::is_void<T>::value; | |
| template <typename... Result> | |
| class DefaultResultsFuture { | |
| public: | |
| using type = Future<std::vector<std::tuple<Result...>>>; | |
| }; | |
| template <typename Result> | |
| class DefaultResultsFuture<Result> { | |
| public: | |
| using type = Future<std::vector<Result>>; | |
| }; | |
| template <> | |
| class DefaultResultsFuture<> { | |
| public: | |
| using type = Future<>; | |
| }; | |
| template <typename... Result> | |
| using DefaultResultsFuture_t = typename DefaultResultsFuture<Result...>::type; | |
| template <typename ResultsFuture> | |
| class ResultCollector { | |
| public: | |
| // ResultsFuture = Future<std::vector<ElementType>> | |
| using ElementType = | |
| typename std::tuple_element_t<0, typename ResultsFuture::result_tuple_type>::value_type; | |
| ResultCollector(size_t reserved_count) : results_(reserved_count) {} | |
| bool IsComplete() const { return finished_count_ == results_.size(); } | |
| template <typename... Result> | |
| void AssignResult(size_t result_index, Result&&... result) { | |
| results_[result_index] = std::make_unique<ElementType>(std::forward<Result>(result)...); | |
| finished_count_++; | |
| } | |
| void Complete(ResultsFuture* future) { | |
| std::vector<ElementType> final_results; | |
| final_results.reserve(results_.size()); | |
| for (auto& result : results_) { | |
| final_results.push_back(std::move(*result)); | |
| } | |
| future->Complete(std::move(final_results)); | |
| } | |
| private: | |
| size_t finished_count_ = 0; | |
| // Use unique ptrs initially so that we can reserve even if |ElementType| is | |
| // not default-constructible. | |
| // | |
| // TODO(rosswang): Consider adding a specialization for default-constructible | |
| // types. | |
| std::vector<std::unique_ptr<ElementType>> results_; | |
| }; | |
| template <> | |
| class ResultCollector<Future<>> { | |
| public: | |
| ResultCollector(size_t reserved_count); | |
| bool IsComplete() const; | |
| // The template on this allows us to use Future<> collectors to swallow | |
| // unneeded results of futures we're waiting on. | |
| template <typename... Result> | |
| void AssignResult(size_t, Result... result) { | |
| finished_count_++; | |
| } | |
| void Complete(Future<>* future) const; | |
| private: | |
| size_t finished_count_ = 0; | |
| size_t reserved_count_; | |
| }; | |
| } // namespace internal | |
| template <typename... Result> | |
| class Future : public fxl::RefCountedThreadSafe<Future<Result...>> { | |
| public: | |
| using result_tuple_type = std::tuple<Result...>; | |
| // Creates a FuturePtr<Result...>. |trace_name| is used solely for debugging | |
| // purposes, and is logged when something goes wrong (e.g. Complete() is | |
| // called twice.) | |
| static FuturePtr<Result...> Create(const std::string& trace_name) { | |
| auto f = fxl::AdoptRef(new Future<Result...>); | |
| f->trace_name_ = std::move(trace_name); | |
| return f; | |
| } | |
| // Creates a FuturePtr<Result...> that's already completed. For example: | |
| // | |
| // FuturePtr<int> f = Future<int>::CreateCompleted("MyFuture", 5); | |
| // f->Then([] (int i) { | |
| // // this lambda executes immediately | |
| // assert(i == 5); | |
| // }); | |
| static FuturePtr<Result...> CreateCompleted(const std::string& trace_name, Result&&... result) { | |
| auto f = Create(std::move(trace_name)); | |
| f->Complete(std::forward<Result>(result)...); | |
| return f; | |
| } | |
| // Completes a future with |result|. This causes any callbacks registered | |
| // with Then(), ConstThen(), etc to be invoked with |result| passed to them | |
| // as a parameter. | |
| // | |
| // Calling Complete() does not affect this future's refcount. This is because: | |
| // | |
| // 1. Any callbacks that are registered are called immediately and | |
| // synchronously, so the future's lifetime does not need to be extended | |
| // before callbacks are invoked. | |
| // 2. Then() correctly handles cases where the future may be deleted by their | |
| // callbacks. | |
| // 3. There is no danger of the future being deleted before Complete() is | |
| // called, because if Complete() is called, the code that calls Complete() | |
| // must have a reference to the future. | |
| void Complete(Result&&... result) { | |
| CompleteWithTuple(std::forward_as_tuple(std::forward<Result>(result)...)); | |
| } | |
| // Returns a fit::function<void(Result)> that, when called, calls Complete() | |
| // on this future. For example: | |
| // | |
| // FuturePtr<Bytes> MakeNetworkRequest(NetworkRequest& request) { | |
| // auto f = Future<Bytes>::Create(); | |
| // PerformAsyncNetworkRequest(request, f->Completer()); | |
| // return f; | |
| // } | |
| // | |
| // The returned closure will maintain a reference to the future, so that the | |
| // closure can call Complete() on it correctly later. In other words, calling | |
| // Completer() will increase this future's refcount, and you do not need to | |
| // maintain a reference to it. After the closure is called, the future's | |
| // refcount will drop by 1. This enables you to write code like | |
| // | |
| // { | |
| // auto f = Future<>::Create(); | |
| // CallAsyncMethod(f->Completer()); | |
| // // f will now go out of scope, but f->Completer() owns it, so it's | |
| // // still kept alive. | |
| // } | |
| fit::function<void(Result...)> Completer() { | |
| return [shared_this = FuturePtr<Result...>(this)](Result&&... result) { | |
| shared_this->Complete(std::forward<Result>(result)...); | |
| }; | |
| } | |
| // Attaches a |callback| that is invoked when the future is completed with | |
| // Complete(), and returns a Future that is complete once |callback| has | |
| // finished executing. | |
| // | |
| // * The callback is invoked immediately (synchronously); it is not scheduled | |
| // on the event loop. | |
| // * The callback is invoked on the same thread as the code that calls | |
| // Complete(). | |
| // * Only one callback can be attached: any callback that was previously | |
| // attached with Then() is discarded. | |
| // * |callback| is called after callbacks attached with ConstThen(). | |
| // * It is safe for |callback| to delete the future that Then() is invoked on. | |
| // If this occurs, any chained futures returned by Then(), Map() etc will be | |
| // de-referenced by this future and not be completed, even if a reference to | |
| // the chained future is maintained elsewhere. | |
| // * It is also safe for |callback| to delete the chained future that Then() | |
| // returns. | |
| // * The future returned by Then() will be owned by this future, so you do not | |
| // need to maintain a reference to it. | |
| // | |
| // The type of this function looks complex, but is basically: | |
| // | |
| // FuturePtr<> Then(fit::function<void(Result...)> callback); | |
| template <typename Callback, typename = typename std::enable_if_t< | |
| internal::is_void_v<std::result_of_t<Callback(Result...)>>>> | |
| FuturePtr<> Then(Callback callback) { | |
| return SubfutureCreate(Future<>::Create(trace_name_ + "(Then)"), | |
| SubfutureVoidCallback<Result...>(std::move(callback)), | |
| SubfutureCompleter<>(), [] { return true; }); | |
| } | |
| // Equivalent to Then(), but guards execution of |callback| with a WeakPtr. | |
| // If, at the time |callback| is to be executed, |weak_ptr| has been | |
| // invalidated, |callback| is not run, nor is the next Future in the chain | |
| // completed. | |
| template <typename Callback, typename T, | |
| typename = typename std::enable_if_t< | |
| internal::is_void_v<std::result_of_t<Callback(Result...)>>>> | |
| FuturePtr<> WeakThen(fxl::WeakPtr<T> weak_ptr, Callback callback) { | |
| return SubfutureCreate(Future<>::Create(trace_name_ + "(WeakThen)"), | |
| SubfutureVoidCallback<Result...>(std::move(callback)), | |
| SubfutureCompleter<>(), [weak_ptr] { return !!weak_ptr; }); | |
| } | |
| // Similar to Then(), except that: | |
| // | |
| // * |const_callback| must take in the completed result via a const&, | |
| // * multiple callbacks can be attached, | |
| // * |const_callback| is called _before_ the Then() callback. | |
| FuturePtr<> ConstThen(fit::function<void(const Result&...)> const_callback) { | |
| FuturePtr<> subfuture = Future<>::Create(trace_name_ + "(ConstThen)"); | |
| AddConstCallback(SubfutureCallback<const Result&...>( | |
| subfuture, SubfutureVoidCallback<const Result&...>(std::move(const_callback)), | |
| SubfutureCompleter<>(), [] { return true; })); | |
| return subfuture; | |
| } | |
| // See WeakThen(). | |
| template <typename T> | |
| FuturePtr<> WeakConstThen(fxl::WeakPtr<T> weak_ptr, | |
| fit::function<void(const Result&...)> const_callback) { | |
| FuturePtr<> subfuture = Future<>::Create(trace_name_ + "(WeakConstThen)"); | |
| AddConstCallback(SubfutureCallback<const Result&...>( | |
| subfuture, SubfutureVoidCallback<const Result&...>(std::move(const_callback)), | |
| SubfutureCompleter<>(), [weak_ptr] { return !!weak_ptr; })); | |
| return subfuture; | |
| } | |
| // Attaches a |callback| that is invoked when this future is completed with | |
| // Complete(). |callback| must return another future: when the returned future | |
| // completes, the future returned by AsyncMap() will complete. For example: | |
| // | |
| // ShowProgressSpinner(true); | |
| // FuturePtr<Bytes> f = MakeNetworkRequest(request); | |
| // f->AsyncMap([] (Bytes zipped_bytes) { | |
| // FuturePtr<Bytes> f = UnzipInBackground(zipped_bytes); | |
| // return f; | |
| // })->AsyncMap([] (Bytes unzipped_bytes) { | |
| // FuturePtr<JPEGImage> f = DecodeImageInBackground(unzipped_bytes); | |
| // return f; | |
| // })->AsyncMap([] (JPEGImage image) { | |
| // FuturePtr<> f = UpdateUIAsynchronously(image); | |
| // return f; | |
| // })->Then([] { | |
| // ShowProgressSpinner(false); | |
| // }); | |
| // | |
| // The type of this method looks terrifying, but is basically: | |
| // | |
| // FuturePtr<CallbackResult> | |
| // AsyncMap(fit::function<FuturePtr<CallbackResult>(Result...)> callback); | |
| template <typename Callback, typename AsyncMapResult = std::result_of_t<Callback(Result...)>, | |
| typename MapResult = typename AsyncMapResult::element_type::result_tuple_type, | |
| typename = typename std::enable_if_t< | |
| internal::is_convertible_v<FuturePtr<MapResult>, AsyncMapResult>>> | |
| AsyncMapResult AsyncMap(Callback callback) { | |
| return SubfutureCreate(AsyncMapResult::element_type::Create(trace_name_ + "(AsyncMap)"), | |
| std::move(callback), SubfutureAsyncMapCompleter<AsyncMapResult>(), | |
| [] { return true; }); | |
| } | |
| template <typename Callback, typename T, | |
| typename AsyncMapResult = std::result_of_t<Callback(Result...)>, | |
| typename MapResult = typename AsyncMapResult::element_type::result_tuple_type, | |
| typename = typename std::enable_if_t< | |
| internal::is_convertible_v<FuturePtr<MapResult>, AsyncMapResult>>> | |
| AsyncMapResult WeakAsyncMap(fxl::WeakPtr<T> weak_ptr, Callback callback) { | |
| return SubfutureCreate(AsyncMapResult::element_type::Create(trace_name_ + "(WeakAsyncMap)"), | |
| std::move(callback), SubfutureAsyncMapCompleter<AsyncMapResult>(), | |
| [weak_ptr] { return !!weak_ptr; }); | |
| } | |
| // Attaches a |callback| that is invoked when this future is completed with | |
| // Complete(). The returned future is completed with |callback|'s return | |
| // value, when |callback| finishes executing. Returned tuples are flattened | |
| // into variadic futures. | |
| // | |
| // To return a future that produces a tuple (uncommon), wrap the map result | |
| // in another tuple (or use |AsyncMap|). | |
| // | |
| // That is: | |
| // Callback return type | Returned future | |
| // ----------------------------------+---------------- | |
| // T | FuturePtr<T> | |
| // std::tuple<T, U, ...> | FuturePtr<T, U, ...> | |
| // std::tuple<std::tuple<T, U, ...>> | FuturePtr<std::tuple<T, U, ...>> | |
| template <typename Callback, typename MapResult = std::result_of_t<Callback(Result...)>> | |
| auto Map(Callback callback) { | |
| return Map(std::move(callback), Tag<MapResult>{}); | |
| } | |
| template <typename Callback, typename T, | |
| typename MapResult = std::result_of_t<Callback(Result...)>> | |
| auto WeakMap(fxl::WeakPtr<T> weak_ptr, Callback callback) { | |
| return WeakMap(weak_ptr, std::move(callback), Tag<MapResult>{}); | |
| } | |
| const std::string& trace_name() const { return trace_name_; } | |
| private: | |
| template <typename... Args> | |
| friend class Future; | |
| // This is a utility class used as a template parameter to determine function | |
| // overloading; see | |
| // <https://www.boost.org/community/generic_programming.html#tag_dispatching> | |
| // for more info. | |
| // | |
| // It is used in the |Map| overloads to "specialize" for |std::tuple| and | |
| // capture its parameter pack. | |
| template <typename T> | |
| class Tag {}; | |
| template <typename ResultsFuture, typename... Args> | |
| friend fxl::RefPtr<ResultsFuture> Wait(const std::string& trace_name, | |
| const std::vector<FuturePtr<Args...>>& futures); | |
| template <typename ResultsFuture, typename TimeoutCallback, typename... Args> | |
| friend fxl::RefPtr<ResultsFuture> WaitWithTimeout(const std::string& trace_name, | |
| async_dispatcher_t* dispatcher, | |
| zx::duration timeout, | |
| TimeoutCallback on_timeout, | |
| const std::vector<FuturePtr<Args...>>& futures); | |
| FRIEND_REF_COUNTED_THREAD_SAFE(Future); | |
| Future() : result_{}, weak_factory_(this) {} | |
| void SetCallback(fit::function<void(Result...)> callback) { | |
| callback_ = std::move(callback); | |
| MaybeInvokeCallbacks(); | |
| } | |
| void SetCallbackWithTuple(fit::function<void(std::tuple<Result...>)> callback) { | |
| SetCallback([callback = std::move(callback)](Result&&... result) { | |
| callback(std::forward_as_tuple(std::forward<Result>(result)...)); | |
| }); | |
| } | |
| void AddConstCallback(fit::function<void(const Result&...)> callback) { | |
| if (!callback) { | |
| return; | |
| } | |
| // It's impossible to add a const callback after a future is completed | |
| // *and* it has a callback: the completed value will be moved into the | |
| // callback and won't be available for a ConstThen(). | |
| if (status_ == internal::FutureStatus::kConsumed) { | |
| FX_LOGS(FATAL) << "Future@" << static_cast<void*>(this) | |
| << (trace_name_.length() ? "(" + trace_name_ + ")" : "") | |
| << ": Cannot add a const callback after completed result is " | |
| "already moved into Then() callback."; | |
| } | |
| const_callbacks_.emplace_back(std::move(callback)); | |
| MaybeInvokeCallbacks(); | |
| } | |
| void CompleteWithTuple(std::tuple<Result...>&& result) { | |
| FX_DCHECK(status_ == internal::FutureStatus::kAwaiting) | |
| << "Future@" << static_cast<void*>(this) | |
| << (trace_name_.length() ? "(" + trace_name_ + ")" : "") << ": Complete() called twice."; | |
| result_ = std::forward<std::tuple<Result...>>(result); | |
| status_ = internal::FutureStatus::kCompleted; | |
| MaybeInvokeCallbacks(); | |
| } | |
| void MaybeInvokeCallbacks() { | |
| if (status_ == internal::FutureStatus::kAwaiting) { | |
| return; | |
| } | |
| if (const_callbacks_.size()) { | |
| // Move |const_callbacks_| to a local variable. MaybeInvokeCallbacks() | |
| // can be called multiple times if the client only uses ConstThen() or | |
| // WeakConstThen() to fetch the completed values. This prevents calling | |
| // these callbacks multiple times by moving them out of the members | |
| // scope. | |
| auto local_const_callbacks = std::move(const_callbacks_); | |
| for (auto& const_callback : local_const_callbacks) { | |
| fxl::Apply(const_callback, result_); | |
| } | |
| } | |
| if (callback_) { | |
| auto callback = std::move(callback_); | |
| status_ = internal::FutureStatus::kConsumed; | |
| fxl::Apply(callback, std::move(result_)); | |
| } | |
| } | |
| // The "subfuture" methods below are private helper functions designed to be | |
| // used with the futures that are returned by the public API (Then(), Map(), | |
| // etc); those returned futures are named "subfutures" in the code, which is | |
| // why the methods are named likewise. | |
| // A convenience method to call this->SetCallback() with the lambda returned | |
| // by SubfutureCallback(). | |
| template <typename Subfuture, typename SubfutureCompleter, typename Callback, typename Guard> | |
| Subfuture SubfutureCreate(Subfuture subfuture, Callback&& callback, | |
| SubfutureCompleter&& subfuture_completer, Guard&& guard) { | |
| SetCallback(SubfutureCallback<Result...>(subfuture, callback, subfuture_completer, guard)); | |
| return subfuture; | |
| } | |
| // Returns a lambda that: | |
| // | |
| // 1. calls |guard| before calling |callback|, and only calls |callback| if | |
| // |guard| returns true; | |
| // 2. will not call |subfuture_completer| if either |this| or |subfuture| are | |
| // destroyed by |callback|. | |
| template <typename... CoercedResult, typename Subfuture, typename SubfutureCompleter, | |
| typename Callback, typename Guard> | |
| auto SubfutureCallback(Subfuture subfuture, Callback&& callback, | |
| SubfutureCompleter&& subfuture_completer, Guard&& guard) { | |
| return [this, subfuture, callback = std::move(callback), subfuture_completer, | |
| guard](CoercedResult&&... result) mutable { | |
| if (!guard()) | |
| return; | |
| auto weak_future = weak_factory_.GetWeakPtr(); | |
| auto weak_subfuture = subfuture->weak_factory_.GetWeakPtr(); | |
| auto subfuture_result = callback(std::forward<CoercedResult>(result)...); | |
| // |callback| above may delete this future or the returned subfuture when | |
| // it finishes executing, so check if |weak_future| and |weak_subfuture| | |
| // are still valid before attempting to complete the subfuture. | |
| if (weak_future && weak_subfuture) | |
| subfuture_completer(subfuture, std::move(subfuture_result)); | |
| }; | |
| } | |
| // Returns a lambda that calls |callback|, and returns an empty tuple. The | |
| // consistent return type enables generic programming techniques to be | |
| // applied to |callback| since the return type is consistent (it's always a | |
| // |std::tuple<T...>|). | |
| template <typename... CoercedResult, typename Callback> | |
| auto SubfutureVoidCallback(Callback&& callback) { | |
| return [callback = std::move(callback)](CoercedResult&&... result) mutable { | |
| callback(std::forward<CoercedResult>(result)...); | |
| return std::make_tuple(); | |
| }; | |
| } | |
| // Returns a lambda that, when called with a subfuture and a std::tuple, will | |
| // complete the subfuture with the values from the tuple elements. This method | |
| // is designed to be used with SubfutureAsyncMapCompleter(), which will do | |
| // the same thing but can be passed Futures for the std::tuple values. | |
| // Together, this enables generic programming techniques to be applied to the | |
| // returned lambda, since the lambda presents a consistent API for callers. | |
| template <typename... SubfutureResult> | |
| auto SubfutureCompleter() { | |
| return [](FuturePtr<SubfutureResult...> subfuture, | |
| std::tuple<SubfutureResult...> subfuture_result) { | |
| subfuture->CompleteWithTuple(std::move(subfuture_result)); | |
| }; | |
| } | |
| // See the documentation for SubfutureCompleter() above. | |
| template <typename AsyncMapResult> | |
| auto SubfutureAsyncMapCompleter() { | |
| return [](AsyncMapResult subfuture, std::tuple<AsyncMapResult> subfuture_result) { | |
| std::get<0>(subfuture_result) | |
| ->SetCallbackWithTuple( | |
| [subfuture](std::tuple<typename AsyncMapResult::element_type::result_tuple_type> | |
| transformed_result) { | |
| subfuture->CompleteWithTuple(std::move(std::get<0>(transformed_result))); | |
| }); | |
| }; | |
| } | |
| // The following overloads enable |Map| and |WeakMap| to flatten out functions that map to | |
| // |std::tuple|. | |
| template <typename Callback, typename MapResult> | |
| FuturePtr<MapResult> Map(Callback callback, Tag<MapResult>) { | |
| // Directly passing |callback| like this ends up relying on an implicit | |
| // |std::tuple| memberwise constructor, which should be fine. It will | |
| // convert from |MapResult| to |std::tuple<MapResult>| implicitly. | |
| return Map(std::move(callback), Tag<std::tuple<MapResult>>{}); | |
| } | |
| template <typename Callback, typename... MapResult> | |
| FuturePtr<MapResult...> Map(Callback callback, Tag<std::tuple<MapResult...>>) { | |
| return SubfutureCreate(Future<MapResult...>::Create(trace_name_ + "(Map)"), std::move(callback), | |
| SubfutureCompleter<MapResult...>(), [] { return true; }); | |
| } | |
| template <typename Callback, typename T, typename MapResult> | |
| FuturePtr<MapResult> WeakMap(fxl::WeakPtr<T> weak_ptr, Callback callback, Tag<MapResult>) { | |
| // Directly passing |callback| like this ends up relying on an implicit | |
| // |std::tuple| memberwise constructor, which should be fine. It will | |
| // convert from |MapResult| to |std::tuple<MapResult>| implicitly. | |
| return WeakMap(weak_ptr, std::move(callback), Tag<std::tuple<MapResult>>{}); | |
| } | |
| template <typename Callback, typename T, typename... MapResult> | |
| FuturePtr<MapResult...> WeakMap(fxl::WeakPtr<T> weak_ptr, Callback callback, | |
| Tag<std::tuple<MapResult...>>) { | |
| return SubfutureCreate(Future<MapResult...>::Create(trace_name_ + "(WeakMap)"), | |
| std::move(callback), SubfutureCompleter<MapResult...>(), | |
| [weak_ptr] { return !!weak_ptr; }); | |
| } | |
| std::string trace_name_; | |
| internal::FutureStatus status_ = internal::FutureStatus::kAwaiting; | |
| std::tuple<Result...> result_; | |
| // The callback attached to this future. | |
| fit::function<void(Result...)> callback_; | |
| // Callbacks that have attached with the Const*() methods, such as | |
| // ConstThen(). | |
| std::vector<fit::function<void(const Result&...)>> const_callbacks_; | |
| // Keep this last in the list of members. (See WeakPtrFactory documentation | |
| // for more info.) | |
| fxl::WeakPtrFactory<Future<Result...>> weak_factory_; | |
| FXL_DISALLOW_COPY_ASSIGN_AND_MOVE(Future); | |
| // For unit tests only. | |
| friend class FutureTest; | |
| // For unit tests only. | |
| const std::tuple<Result...>& get() const { | |
| FX_DCHECK(status_ != internal::FutureStatus::kAwaiting) | |
| << trace_name_ << ": get() called on unset future"; | |
| return result_; | |
| } | |
| }; | |
| // Returns a Future that completes when every future in |futures| is complete. | |
| // The future returned by Wait() will be kept alive until every future in | |
| // |futures| either completes or is destroyed. If any future in |futures| is | |
| // destroyed prior to completing, the returned future will never complete. The | |
| // order of the results corresponds to the order of the given futures, | |
| // regardless of their completion order. | |
| // | |
| // The default type of the resulting future depends on the types of the | |
| // component futures. Void futures produce a void future. Monadic futures | |
| // produce a future of a flat vector. Polyadic futures produce a future of a | |
| // vector of tuples. That is: | |
| // | |
| // Components | Result | |
| // -------------------+-------------------------------------------- | |
| // FuturePtr<> | FuturePtr<> | |
| // FuturePtr<T> | FuturePtr<std::vector<T>> | |
| // FuturePtr<T, U...> | FuturePtr<std::vector<std::tuple<T, U...>>> | |
| // | |
| // These defaults may be overridden by specifying the template argument for | |
| // Wait as the type of future desired. All cases may produce a Future<> or a | |
| // Future<std::vector<std::tuple<...>>>. | |
| // | |
| // Example usage: | |
| // | |
| // FuturePtr<Bytes> f1 = MakeNetworkRequest(request1); | |
| // FuturePtr<Bytes> f2 = MakeNetworkRequest(request2); | |
| // FuturePtr<Bytes> f3 = MakeNetworkRequest(request3); | |
| // std::vector<FuturePtr<Bytes>> requests{f1, f2, f3}; | |
| // Wait("NetworkRequests", requests)->Then([]( | |
| // std::vector<Bytes> bytes_vector) { | |
| // Bytes f1_bytes = bytes_vector[0]; | |
| // Bytes f2_bytes = bytes_vector[1]; | |
| // Bytes f3_bytes = bytes_vector[2]; | |
| // }); | |
| // | |
| // This is similar to Promise.All() in JavaScript, or Join() in Rust. | |
| template <typename ResultsFuture, typename... Result> | |
| fxl::RefPtr<ResultsFuture> Wait(const std::string& trace_name, | |
| const std::vector<FuturePtr<Result...>>& futures) { | |
| if (futures.empty()) { | |
| auto immediate = ResultsFuture::Create(trace_name + "(Completed)"); | |
| immediate->CompleteWithTuple({}); | |
| return immediate; | |
| } | |
| auto results = std::make_shared<internal::ResultCollector<ResultsFuture>>(futures.size()); | |
| fxl::RefPtr<ResultsFuture> all_futures_completed = | |
| ResultsFuture::Create(trace_name + "(WillWait)"); | |
| for (size_t i = 0; i < futures.size(); i++) { | |
| const auto& future = futures[i]; | |
| future->SetCallback([i, all_futures_completed, results](Result&&... result) { | |
| results->AssignResult(i, std::forward<Result>(result)...); | |
| if (results->IsComplete()) { | |
| results->Complete(all_futures_completed.get()); | |
| } | |
| }); | |
| } | |
| return all_futures_completed; | |
| } | |
| // Like |Wait|, but gives up after a timeout. After the timeout, |on_timeout| is | |
| // invoked with a diagnostic error string containing the trace names of the | |
| // futures that have not completed. | |
| // | |
| // This maintains a reference to the returned |Future| until all component | |
| // futures have been completed or destroyed, or until the timeout has elapsed, | |
| // whichever happens first. However, |on_timeout| will be invoked on timeout if | |
| // any future has not completed even if any or all futures have been destroyed. | |
| template <typename ResultsFuture, typename TimeoutCallback, typename... Result> | |
| fxl::RefPtr<ResultsFuture> WaitWithTimeout( | |
| const std::string& trace_name, async_dispatcher_t* dispatcher, zx::duration timeout, | |
| TimeoutCallback on_timeout /* void (const std::string&) */, | |
| const std::vector<FuturePtr<Result...>>& futures) { | |
| auto all_futures_completed = Wait<ResultsFuture>(trace_name, futures); | |
| if (all_futures_completed->status_ != internal::FutureStatus::kAwaiting) { | |
| return all_futures_completed; | |
| } | |
| auto all_trace_names = std::make_shared<std::vector<std::unique_ptr<std::string>>>(); | |
| all_trace_names->reserve(futures.size()); | |
| for (const auto& future : futures) { | |
| // There's no point in waiting on completed futures. Furthermore if we tried | |
| // that we'd have to put this before |Wait| since |Wait| consumes the | |
| // results, but then if all futures are already completed this is all just | |
| // wasted effort. | |
| if (future->status_ == internal::FutureStatus::kAwaiting) { | |
| size_t i = all_trace_names->size(); | |
| all_trace_names->push_back(std::make_unique<std::string>(future->trace_name_)); | |
| future->AddConstCallback([i, all_trace_names](const Result&...) { | |
| if (!all_trace_names->empty()) { | |
| (*all_trace_names)[i] = nullptr; | |
| } | |
| }); | |
| } | |
| } | |
| // Return a proxy so that we can cancel result forwarding in the case of a | |
| // timeout. This could with more difficulty be done within |Wait|, but this | |
| // way allows us to reuse the logic more easily. | |
| fxl::RefPtr<ResultsFuture> all_proxy = | |
| ResultsFuture::Create(trace_name + "(WillWaitWithTimeout)"); | |
| all_futures_completed->SetCallback(all_proxy->Completer()); | |
| // TODO(rosswang): Factor this into dump and cancel functions that can be | |
| // called at other times. | |
| async::PostDelayedTask( | |
| dispatcher, | |
| [all_trace_names = std::move(all_trace_names), on_timeout = std::move(on_timeout), | |
| all_futures_completed = all_futures_completed->weak_factory_.GetWeakPtr()] { | |
| std::ostringstream msg; | |
| for (const auto& trace_name : *all_trace_names) { | |
| if (trace_name) { | |
| msg << "\n\t" << *trace_name; | |
| } | |
| } | |
| if (!msg.str().empty()) { | |
| on_timeout("Wait timed out. Still waiting for futures:" + msg.str()); | |
| if (all_futures_completed) { | |
| // cancel results forwarding (possibly releasing all_proxy) | |
| all_futures_completed->SetCallback(nullptr); | |
| } | |
| // Possibly release the component futures. Once this task completes | |
| // and goes out of scope, the last reference to the Wait future (also | |
| // holding onto the component futures) should be released as well. | |
| all_trace_names->clear(); | |
| } | |
| }, | |
| timeout); | |
| return all_proxy; | |
| } | |
| // |WaitWithTimeout| on the thread defaut dispatcher. | |
| template <typename ResultsFuture, typename TimeoutCallback, typename... Result> | |
| fxl::RefPtr<ResultsFuture> WaitWithTimeout( | |
| const std::string& trace_name, zx::duration timeout, | |
| TimeoutCallback on_timeout /* void (const std::string&) */, | |
| const std::vector<FuturePtr<Result...>>& futures) { | |
| return WaitWithTimeout<ResultsFuture>(trace_name, async_get_default_dispatcher(), timeout, | |
| std::move(on_timeout), futures); | |
| } | |
| // These overloads allow us to effectively default the first template parameter, | |
| // |ResultsFuture| (since the others are intended to be inferred). | |
| // TODO(rosswang): If the overload combinatoric explosion gets too heavy, we can | |
| // use a dummy struct parameter instead to encapsulate that template parameter. | |
| template <typename... Result> | |
| auto Wait(const std::string& trace_name, const std::vector<FuturePtr<Result...>>& futures) { | |
| return Wait<internal::DefaultResultsFuture_t<Result...>>(trace_name, futures); | |
| } | |
| template <typename TimeoutCallback, typename... Result> | |
| auto WaitWithTimeout(const std::string& trace_name, async_dispatcher_t* dispatcher, | |
| zx::duration timeout, TimeoutCallback on_timeout, | |
| const std::vector<FuturePtr<Result...>>& futures) { | |
| return WaitWithTimeout<internal::DefaultResultsFuture_t<Result...>>( | |
| trace_name, dispatcher, timeout, std::move(on_timeout), futures); | |
| } | |
| template <typename TimeoutCallback, typename... Result> | |
| auto WaitWithTimeout(const std::string& trace_name, zx::duration timeout, | |
| TimeoutCallback on_timeout /* void (const std::string&) */, | |
| const std::vector<FuturePtr<Result...>>& futures) { | |
| return WaitWithTimeout<internal::DefaultResultsFuture_t<Result...>>( | |
| trace_name, async_get_default_dispatcher(), timeout, std::move(on_timeout), futures); | |
| } | |
| // We need to provide the initializer list overloads or template deduction fails | |
| // for the above overloads if given an initializer list. | |
| // TODO(rosswang): Add a potentially heterogeneous variadic template instead, | |
| // and prefer it over initializer lists outside of tests. | |
| template <typename ResultsFuture, typename... Result> | |
| auto Wait(const std::string& trace_name, std::initializer_list<FuturePtr<Result...>> futures) { | |
| return Wait<ResultsFuture>(trace_name, std::vector<FuturePtr<Result...>>(futures)); | |
| } | |
| template <typename ResultsFuture, typename TimeoutCallback, typename... Result> | |
| fxl::RefPtr<ResultsFuture> WaitWithTimeout(const std::string& trace_name, | |
| async_dispatcher_t* dispatcher, zx::duration timeout, | |
| TimeoutCallback on_timeout, | |
| std::initializer_list<FuturePtr<Result...>> futures) { | |
| return WaitWithTimeout<ResultsFuture>(trace_name, dispatcher, timeout, std::move(on_timeout), | |
| std::vector<FuturePtr<Result...>>(futures)); | |
| } | |
| template <typename ResultsFuture, typename TimeoutCallback, typename... Result> | |
| fxl::RefPtr<ResultsFuture> WaitWithTimeout( | |
| const std::string& trace_name, zx::duration timeout, | |
| TimeoutCallback on_timeout /* void (const std::string&) */, | |
| std::initializer_list<FuturePtr<Result...>> futures) { | |
| return WaitWithTimeout<ResultsFuture>(trace_name, async_get_default_dispatcher(), timeout, | |
| std::move(on_timeout), | |
| std::vector<FuturePtr<Result...>>(futures)); | |
| } | |
| template <typename... Result> | |
| auto Wait(const std::string& trace_name, std::initializer_list<FuturePtr<Result...>> futures) { | |
| return Wait(trace_name, std::vector<FuturePtr<Result...>>(futures)); | |
| } | |
| template <typename TimeoutCallback, typename... Result> | |
| auto WaitWithTimeout(const std::string& trace_name, async_dispatcher_t* dispatcher, | |
| zx::duration timeout, TimeoutCallback on_timeout, | |
| std::initializer_list<FuturePtr<Result...>> futures) { | |
| return WaitWithTimeout(trace_name, dispatcher, timeout, std::move(on_timeout), | |
| std::vector<FuturePtr<Result...>>(futures)); | |
| } | |
| template <typename TimeoutCallback, typename... Result> | |
| auto WaitWithTimeout(const std::string& trace_name, zx::duration timeout, | |
| TimeoutCallback on_timeout /* void (const std::string&) */, | |
| std::initializer_list<FuturePtr<Result...>> futures) { | |
| return WaitWithTimeout(trace_name, async_get_default_dispatcher(), timeout, std::move(on_timeout), | |
| std::vector<FuturePtr<Result...>>(futures)); | |
| } | |
| } // namespace modular | |
| #endif // SRC_MODULAR_LIB_ASYNC_CPP_FUTURE_H_ |