Commit 69f86ccc authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook Github Bot

Add via() overloads that accept Executor::KeepAlive

Summary: This enables using Executor adaptors (e.g. SerialExecutor) with futures.

Reviewed By: yfeldblum

Differential Revision: D8198893

fbshipit-source-id: 48c59b618bd2bbad4277b7f0b0f0b19347266813
parent 3ce2b217
...@@ -135,6 +135,9 @@ class Executor { ...@@ -135,6 +135,9 @@ class Executor {
static_assert( static_assert(
std::is_base_of<Executor, ExecutorT>::value, std::is_base_of<Executor, ExecutorT>::value,
"getKeepAliveToken only works for folly::Executor implementations."); "getKeepAliveToken only works for folly::Executor implementations.");
if (!executor) {
return {};
}
folly::Executor* executorPtr = executor; folly::Executor* executorPtr = executor;
if (executorPtr->keepAliveAcquire()) { if (executorPtr->keepAliveAcquire()) {
return makeKeepAlive<ExecutorT>(executor); return makeKeepAlive<ExecutorT>(executor);
......
...@@ -724,22 +724,29 @@ SemiFuture<T>& SemiFuture<T>::operator=(Future<T>&& other) noexcept { ...@@ -724,22 +724,29 @@ SemiFuture<T>& SemiFuture<T>::operator=(Future<T>&& other) noexcept {
} }
template <class T> template <class T>
Future<T> SemiFuture<T>::via(Executor* executor, int8_t priority) && { Future<T> SemiFuture<T>::via(
Executor::KeepAlive<> executor,
int8_t priority) && {
if (!executor) { if (!executor) {
throw_exception<FutureNoExecutor>(); throw_exception<FutureNoExecutor>();
} }
if (auto deferredExecutor = getDeferredExecutor()) { if (auto deferredExecutor = getDeferredExecutor()) {
deferredExecutor->setExecutor(executor); deferredExecutor->setExecutor(executor.get());
} }
auto newFuture = Future<T>(this->core_); auto newFuture = Future<T>(this->core_);
this->core_ = nullptr; this->core_ = nullptr;
newFuture.setExecutor(executor, priority); newFuture.setExecutor(std::move(executor), priority);
return newFuture; return newFuture;
} }
template <class T>
Future<T> SemiFuture<T>::via(Executor* executor, int8_t priority) && {
return std::move(*this).via(getKeepAliveToken(executor), priority);
}
template <class T> template <class T>
Future<T> SemiFuture<T>::toUnsafeFuture() && { Future<T> SemiFuture<T>::toUnsafeFuture() && {
return std::move(*this).via(&folly::InlineExecutor::instance()); return std::move(*this).via(&folly::InlineExecutor::instance());
...@@ -867,8 +874,8 @@ typename std:: ...@@ -867,8 +874,8 @@ typename std::
} }
template <class T> template <class T>
Future<T> Future<T>::via(Executor* executor, int8_t priority) && { Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) && {
this->setExecutor(executor, priority); this->setExecutor(std::move(executor), priority);
auto newFuture = Future<T>(this->core_); auto newFuture = Future<T>(this->core_);
this->core_ = nullptr; this->core_ = nullptr;
...@@ -876,7 +883,12 @@ Future<T> Future<T>::via(Executor* executor, int8_t priority) && { ...@@ -876,7 +883,12 @@ Future<T> Future<T>::via(Executor* executor, int8_t priority) && {
} }
template <class T> template <class T>
Future<T> Future<T>::via(Executor* executor, int8_t priority) & { Future<T> Future<T>::via(Executor* executor, int8_t priority) && {
return std::move(*this).via(getKeepAliveToken(executor), priority);
}
template <class T>
Future<T> Future<T>::via(Executor::KeepAlive<> executor, int8_t priority) & {
this->throwIfInvalid(); this->throwIfInvalid();
Promise<T> p; Promise<T> p;
auto sf = p.getSemiFuture(); auto sf = p.getSemiFuture();
...@@ -891,7 +903,12 @@ Future<T> Future<T>::via(Executor* executor, int8_t priority) & { ...@@ -891,7 +903,12 @@ Future<T> Future<T>::via(Executor* executor, int8_t priority) & {
// check in SemiFuture::via // check in SemiFuture::via
auto f = Future<T>(sf.core_); auto f = Future<T>(sf.core_);
sf.core_ = nullptr; sf.core_ = nullptr;
return std::move(f).via(executor, priority); return std::move(f).via(std::move(executor), priority);
}
template <class T>
Future<T> Future<T>::via(Executor* executor, int8_t priority) & {
return via(getKeepAliveToken(executor), priority);
} }
template <typename T> template <typename T>
......
...@@ -268,6 +268,12 @@ class FutureBase { ...@@ -268,6 +268,12 @@ class FutureBase {
getCore().setExecutor(x, priority); getCore().setExecutor(x, priority);
} }
void setExecutor(
Executor::KeepAlive<> x,
int8_t priority = Executor::MID_PRI) {
getCore().setExecutor(std::move(x), priority);
}
// Variant: returns a value // Variant: returns a value
// e.g. f.then([](Try<T> t){ return t.value(); }); // e.g. f.then([](Try<T> t){ return t.value(); });
template <typename F, typename R, bool isTry, typename... Args> template <typename F, typename R, bool isTry, typename... Args>
...@@ -408,6 +414,10 @@ class SemiFuture : private futures::detail::FutureBase<T> { ...@@ -408,6 +414,10 @@ class SemiFuture : private futures::detail::FutureBase<T> {
/// https://akrzemi1.wordpress.com/2014/06/02/ref-qualifiers/ /// https://akrzemi1.wordpress.com/2014/06/02/ref-qualifiers/
Future<T> via(Executor* executor, int8_t priority = Executor::MID_PRI) &&; Future<T> via(Executor* executor, int8_t priority = Executor::MID_PRI) &&;
Future<T> via(
Executor::KeepAlive<> executor,
int8_t priority = Executor::MID_PRI) &&;
/** /**
* Defer work to run on the consumer of the future. * Defer work to run on the consumer of the future.
* Function must take a Try as a parameter. * Function must take a Try as a parameter.
...@@ -689,11 +699,19 @@ class Future : private futures::detail::FutureBase<T> { ...@@ -689,11 +699,19 @@ class Future : private futures::detail::FutureBase<T> {
/// https://akrzemi1.wordpress.com/2014/06/02/ref-qualifiers/ /// https://akrzemi1.wordpress.com/2014/06/02/ref-qualifiers/
Future<T> via(Executor* executor, int8_t priority = Executor::MID_PRI) &&; Future<T> via(Executor* executor, int8_t priority = Executor::MID_PRI) &&;
Future<T> via(
Executor::KeepAlive<> executor,
int8_t priority = Executor::MID_PRI) &&;
/// This variant creates a new future, where the ref-qualifier && version /// This variant creates a new future, where the ref-qualifier && version
/// moves `this` out. This one is less efficient but avoids confusing users /// moves `this` out. This one is less efficient but avoids confusing users
/// when "return f.via(x);" fails. /// when "return f.via(x);" fails.
Future<T> via(Executor* executor, int8_t priority = Executor::MID_PRI) &; Future<T> via(Executor* executor, int8_t priority = Executor::MID_PRI) &;
Future<T> via(
Executor::KeepAlive<> executor,
int8_t priority = Executor::MID_PRI) &;
/** When this Future has completed, execute func which is a function that /** When this Future has completed, execute func which is a function that
takes one of: takes one of:
(const) Try<T>&& (const) Try<T>&&
......
...@@ -194,12 +194,18 @@ class Core final { ...@@ -194,12 +194,18 @@ class Core final {
/// Call only from Future thread, either before attaching a callback or after /// Call only from Future thread, either before attaching a callback or after
/// the callback has already been invoked, but not concurrently with anything /// the callback has already been invoked, but not concurrently with anything
/// which might trigger invocation of the callback /// which might trigger invocation of the callback
void setExecutor(Executor* x, int8_t priority = Executor::MID_PRI) { void setExecutor(
Executor::KeepAlive<> x,
int8_t priority = Executor::MID_PRI) {
DCHECK(fsm_.getState() != State::OnlyCallback); DCHECK(fsm_.getState() != State::OnlyCallback);
executor_ = x ? getKeepAliveToken(x) : Executor::KeepAlive<>(); executor_ = std::move(x);
priority_ = priority; priority_ = priority;
} }
void setExecutor(Executor* x, int8_t priority = Executor::MID_PRI) {
setExecutor(getKeepAliveToken(x), priority);
}
Executor* getExecutor() const { Executor* getExecutor() const {
return executor_.get(); return executor_.get();
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment