Commit bc08e06d authored by Lee Howes's avatar Lee Howes Committed by Facebook Github Bot

Always passing executor into callbacks.

Summary: Propagate the executor parameter through all future callbacks to make it available at any point in the set of continuations.

Reviewed By: yfeldblum

Differential Revision: D15225264

fbshipit-source-id: 6433c9ccf8f3992fc16ebb1ea0171cc634fc3951
parent 1185bdd6
...@@ -332,22 +332,27 @@ namespace detail_msvc_15_7_workaround { ...@@ -332,22 +332,27 @@ namespace detail_msvc_15_7_workaround {
template <typename R, std::size_t S> template <typename R, std::size_t S>
using IfArgsSizeIs = std::enable_if_t<R::Arg::ArgsSize::value == S, int>; using IfArgsSizeIs = std::enable_if_t<R::Arg::ArgsSize::value == S, int>;
template <typename R, typename State, typename T, IfArgsSizeIs<R, 0> = 0> template <typename R, typename State, typename T, IfArgsSizeIs<R, 0> = 0>
decltype(auto) invoke(R, State& state, Try<T>& /* t */) { decltype(auto)
invoke(R, State& state, Executor::KeepAlive<>&&, Try<T>&& /* t */) {
return state.invoke(); return state.invoke();
} }
template <typename R, typename State, typename T, IfArgsSizeIs<R, 1> = 0> template <typename R, typename State, typename T, IfArgsSizeIs<R, 2> = 0>
decltype(auto) invoke(R, State& state, Try<T>& t) { decltype(auto) invoke(R, State& state, Executor::KeepAlive<>&& ka, Try<T>&& t) {
using Arg0 = typename R::Arg::ArgList::FirstArg; using Arg1 = typename R::Arg::ArgList::Tail::FirstArg;
return state.invoke(t.template get<R::Arg::isTry(), Arg0>()); return state.invoke(
std::move(ka), std::move(t).template get<R::Arg::isTry(), Arg1>());
} }
template <typename R, typename State, typename T, IfArgsSizeIs<R, 0> = 0> template <typename R, typename State, typename T, IfArgsSizeIs<R, 0> = 0>
decltype(auto) tryInvoke(R, State& state, Try<T>& /* t */) { decltype(auto)
tryInvoke(R, State& state, Executor::KeepAlive<>&&, Try<T>&& /* t */) {
return state.tryInvoke(); return state.tryInvoke();
} }
template <typename R, typename State, typename T, IfArgsSizeIs<R, 1> = 0> template <typename R, typename State, typename T, IfArgsSizeIs<R, 2> = 0>
decltype(auto) tryInvoke(R, State& state, Try<T>& t) { decltype(auto)
using Arg0 = typename R::Arg::ArgList::FirstArg; tryInvoke(R, State& state, Executor::KeepAlive<>&& ka, Try<T>&& t) {
return state.tryInvoke(t.template get<R::Arg::isTry(), Arg0>()); using Arg1 = typename R::Arg::ArgList::Tail::FirstArg;
return state.tryInvoke(
std::move(ka), std::move(t).template get<R::Arg::isTry(), Arg1>());
} }
} // namespace detail_msvc_15_7_workaround } // namespace detail_msvc_15_7_workaround
...@@ -359,8 +364,7 @@ template <class T> ...@@ -359,8 +364,7 @@ template <class T>
template <typename F, typename R> template <typename F, typename R>
typename std::enable_if<!R::ReturnsFuture::value, typename R::Return>::type typename std::enable_if<!R::ReturnsFuture::value, typename R::Return>::type
FutureBase<T>::thenImplementation(F&& func, R) { FutureBase<T>::thenImplementation(F&& func, R) {
static_assert( static_assert(R::Arg::ArgsSize::value == 2, "Then must take two arguments");
R::Arg::ArgsSize::value <= 1, "Then must take zero/one argument");
typedef typename R::ReturnsFuture::Inner B; typedef typename R::ReturnsFuture::Inner B;
Promise<B> p; Promise<B> p;
...@@ -401,14 +405,15 @@ FutureBase<T>::thenImplementation(F&& func, R) { ...@@ -401,14 +405,15 @@ FutureBase<T>::thenImplementation(F&& func, R) {
in some circumstances, but I think it should be explicit not implicit in some circumstances, but I think it should be explicit not implicit
in the destruction of the Future used to create it. in the destruction of the Future used to create it.
*/ */
this->setCallback_( this->setCallback_([state = futures::detail::makeCoreCallbackState(
[state = futures::detail::makeCoreCallbackState( std::move(p), std::forward<F>(func))](
std::move(p), std::forward<F>(func))](Try<T>&& t) mutable { Executor::KeepAlive<>&& ka, Try<T>&& t) mutable {
if (!R::Arg::isTry() && t.hasException()) { if (!R::Arg::isTry() && t.hasException()) {
state.setException(std::move(t.exception())); state.setException(std::move(t.exception()));
} else { } else {
state.setTry(makeTryWith([&] { state.setTry(makeTryWith([&] {
return detail_msvc_15_7_workaround::invoke(R{}, state, t); return detail_msvc_15_7_workaround::invoke(
R{}, state, std::move(ka), std::move(t));
})); }));
} }
}); });
...@@ -436,8 +441,7 @@ template <class T> ...@@ -436,8 +441,7 @@ template <class T>
template <typename F, typename R> template <typename F, typename R>
typename std::enable_if<R::ReturnsFuture::value, typename R::Return>::type typename std::enable_if<R::ReturnsFuture::value, typename R::Return>::type
FutureBase<T>::thenImplementation(F&& func, R) { FutureBase<T>::thenImplementation(F&& func, R) {
static_assert( static_assert(R::Arg::ArgsSize::value == 2, "Then must take two arguments");
R::Arg::ArgsSize::value <= 1, "Then must take zero/one argument");
typedef typename R::ReturnsFuture::Inner B; typedef typename R::ReturnsFuture::Inner B;
Promise<B> p; Promise<B> p;
...@@ -446,24 +450,25 @@ FutureBase<T>::thenImplementation(F&& func, R) { ...@@ -446,24 +450,25 @@ FutureBase<T>::thenImplementation(F&& func, R) {
// grab the Future now before we lose our handle on the Promise // grab the Future now before we lose our handle on the Promise
auto sf = p.getSemiFuture(); auto sf = p.getSemiFuture();
auto e = getKeepAliveToken(this->getExecutor()); auto e = getKeepAliveToken(this->getExecutor());
sf.setExecutor(e); sf.setExecutor(std::move(e));
auto f = Future<B>(sf.core_); auto f = Future<B>(sf.core_);
sf.core_ = nullptr; sf.core_ = nullptr;
this->setCallback_([state = futures::detail::makeCoreCallbackState( this->setCallback_([state = futures::detail::makeCoreCallbackState(
std::move(p), std::forward<F>(func)), std::move(p), std::forward<F>(func))](
e = std::move(e)](Try<T>&& t) mutable { Executor::KeepAlive<>&& ka, Try<T>&& t) mutable {
if (!R::Arg::isTry() && t.hasException()) { if (!R::Arg::isTry() && t.hasException()) {
state.setException(std::move(t.exception())); state.setException(std::move(t.exception()));
} else { } else {
// Ensure that if function returned a SemiFuture we correctly chain // Ensure that if function returned a SemiFuture we correctly chain
// potential deferral. // potential deferral.
auto tf2 = detail_msvc_15_7_workaround::tryInvoke(R{}, state, t); auto tf2 = detail_msvc_15_7_workaround::tryInvoke(
R{}, state, ka.copy(), std::move(t));
if (tf2.hasException()) { if (tf2.hasException()) {
state.setException(std::move(tf2.exception())); state.setException(std::move(tf2.exception()));
} else { } else {
auto statePromise = state.stealPromise(); auto statePromise = state.stealPromise();
auto tf3 = chainExecutor(std::move(e), *std::move(tf2)); auto tf3 = chainExecutor(std::move(ka), *std::move(tf2));
std::exchange(statePromise.core_, nullptr) std::exchange(statePromise.core_, nullptr)
->setProxy(std::exchange(tf3.core_, nullptr)); ->setProxy(std::exchange(tf3.core_, nullptr));
} }
...@@ -497,12 +502,12 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && { ...@@ -497,12 +502,12 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && {
auto ctx = std::make_shared<Context>(std::move(e)); auto ctx = std::make_shared<Context>(std::move(e));
auto f = [ctx](Try<T>&& t) { auto f = [ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
if (!ctx->token.exchange(true, std::memory_order_relaxed)) { if (!ctx->token.exchange(true, std::memory_order_relaxed)) {
ctx->promise.setTry(std::move(t)); ctx->promise.setTry(std::move(t));
} }
}; };
using R = futures::detail::callableResult<T, decltype(f)>; using R = futures::detail::tryExecutorCallableResult<T, decltype(f)>;
ctx->thisFuture = this->thenImplementation(std::move(f), R{}); ctx->thisFuture = this->thenImplementation(std::move(f), R{});
// Properly propagate interrupt values through futures chained after within() // Properly propagate interrupt values through futures chained after within()
...@@ -961,10 +966,30 @@ SemiFuture<T>::deferExTry(F&& func) && { ...@@ -961,10 +966,30 @@ SemiFuture<T>::deferExTry(F&& func) && {
deferredExecutor = newDeferredExecutor.get(); deferredExecutor = newDeferredExecutor.get();
this->setExecutor(std::move(newDeferredExecutor)); this->setExecutor(std::move(newDeferredExecutor));
} }
return std::move(*this).defer(
[deferredExecutor, f = std::forward<F>(func)](Try<T>&& t) mutable { auto sf =
return f(deferredExecutor->getExecutor(), std::move(t)); Future<T>(this->core_)
}); .thenExTry([func = std::forward<F>(func)](
folly::Executor::KeepAlive<>&& keepAlive,
folly::Try<T>&& val) mutable {
// Extract the raw executor from the deferred and pass to the
// continuation
Executor* thenExDeferredExecutor = keepAlive.get();
assert(
dynamic_cast<DeferredExecutor*>(thenExDeferredExecutor) !=
nullptr);
auto innerExecutorKA = getKeepAliveToken(
static_cast<DeferredExecutor*>(thenExDeferredExecutor)
->getExecutor());
return std::forward<F>(func)(
std::move(innerExecutorKA), std::forward<decltype(val)>(val));
})
.semi();
this->core_ = nullptr;
// Carry deferred executor through chain as constructor from Future will
// nullify it
sf.setExecutor(deferredExecutor);
return sf;
} }
template <class T> template <class T>
...@@ -1117,10 +1142,10 @@ Future<T> Future<T>::via(Executor::KeepAlive<> executor) & { ...@@ -1117,10 +1142,10 @@ Future<T> Future<T>::via(Executor::KeepAlive<> executor) & {
this->throwIfInvalid(); this->throwIfInvalid();
Promise<T> p; Promise<T> p;
auto sf = p.getSemiFuture(); auto sf = p.getSemiFuture();
auto func = [p = std::move(p)](Try<T>&& t) mutable { auto func = [p = std::move(p)](Executor::KeepAlive<>&&, Try<T>&& t) mutable {
p.setTry(std::move(t)); p.setTry(std::move(t));
}; };
using R = futures::detail::callableResult<T, decltype(func)>; using R = futures::detail::tryExecutorCallableResult<T, decltype(func)>;
this->thenImplementation(std::move(func), R{}); this->thenImplementation(std::move(func), R{});
// Construct future from semifuture manually because this may not have // Construct future from semifuture manually because this may not have
// an executor set due to legacy code. This means we can bypass the executor // an executor set due to legacy code. This means we can bypass the executor
...@@ -1152,10 +1177,12 @@ template <class T> ...@@ -1152,10 +1177,12 @@ template <class T>
template <typename F> template <typename F>
Future<typename futures::detail::tryCallableResult<T, F>::value_type> Future<typename futures::detail::tryCallableResult<T, F>::value_type>
Future<T>::thenTry(F&& func) && { Future<T>::thenTry(F&& func) && {
auto lambdaFunc = [f = std::forward<F>(func)](folly::Try<T>&& t) mutable { auto lambdaFunc = [f = std::forward<F>(func)](
folly::Executor::KeepAlive<>&&,
folly::Try<T>&& t) mutable {
return std::forward<F>(f)(std::move(t)); return std::forward<F>(f)(std::move(t));
}; };
using R = futures::detail::tryCallableResult<T, decltype(lambdaFunc)>; using R = futures::detail::tryExecutorCallableResult<T, decltype(lambdaFunc)>;
return this->thenImplementation(std::move(lambdaFunc), R{}); return this->thenImplementation(std::move(lambdaFunc), R{});
} }
...@@ -1163,15 +1190,13 @@ template <class T> ...@@ -1163,15 +1190,13 @@ template <class T>
template <typename F> template <typename F>
Future<typename futures::detail::tryExecutorCallableResult<T, F>::value_type> Future<typename futures::detail::tryExecutorCallableResult<T, F>::value_type>
Future<T>::thenExTry(F&& func) && { Future<T>::thenExTry(F&& func) && {
auto ka = getKeepAliveToken(this->getExecutor()); auto lambdaFunc = [f = std::forward<F>(func)](
Executor::KeepAlive<>&& ka, folly::Try<T>&& t) mutable {
// Enforce that executor cannot be null // Enforce that executor cannot be null
DCHECK(ka); DCHECK(ka);
return std::forward<F>(f)(std::move(ka), std::move(t));
auto lambdaFunc = [f = std::forward<F>(func),
exec = std::move(ka)](folly::Try<T>&& t) mutable {
return std::forward<F>(f)(exec, std::move(t));
}; };
using R = futures::detail::tryCallableResult<T, decltype(lambdaFunc)>; using R = futures::detail::tryExecutorCallableResult<T, decltype(lambdaFunc)>;
return this->thenImplementation(std::move(lambdaFunc), R{}); return this->thenImplementation(std::move(lambdaFunc), R{});
} }
...@@ -1179,10 +1204,11 @@ template <class T> ...@@ -1179,10 +1204,11 @@ template <class T>
template <typename F> template <typename F>
Future<typename futures::detail::valueCallableResult<T, F>::value_type> Future<typename futures::detail::valueCallableResult<T, F>::value_type>
Future<T>::thenValue(F&& func) && { Future<T>::thenValue(F&& func) && {
auto lambdaFunc = [f = std::forward<F>(func)](folly::Try<T>&& t) mutable { auto lambdaFunc = [f = std::forward<F>(func)](
Executor::KeepAlive<>&&, folly::Try<T>&& t) mutable {
return futures::detail::wrapInvoke(std::move(t), std::forward<F>(f)); return futures::detail::wrapInvoke(std::move(t), std::forward<F>(f));
}; };
using R = futures::detail::tryCallableResult<T, decltype(lambdaFunc)>; using R = futures::detail::tryExecutorCallableResult<T, decltype(lambdaFunc)>;
return this->thenImplementation(std::move(lambdaFunc), R{}); return this->thenImplementation(std::move(lambdaFunc), R{});
} }
...@@ -1190,19 +1216,18 @@ template <class T> ...@@ -1190,19 +1216,18 @@ template <class T>
template <typename F> template <typename F>
Future<typename futures::detail::valueExecutorCallableResult<T, F>::value_type> Future<typename futures::detail::valueExecutorCallableResult<T, F>::value_type>
Future<T>::thenExValue(F&& func) && { Future<T>::thenExValue(F&& func) && {
auto ka = getKeepAliveToken(this->getExecutor()); auto lambdaFunc = [f = std::forward<F>(func)](
Executor::KeepAlive<>&& ka, folly::Try<T>&& t) mutable {
// Enforce that executor cannot be null // Enforce that executor cannot be null
DCHECK(ka); DCHECK(ka);
auto lambdaFunc = [f = std::forward<F>(func),
exec = std::move(ka)](folly::Try<T>&& t) mutable {
return std::forward<F>(f)( return std::forward<F>(f)(
exec, std::move(ka),
t.template get< t.template get<
false, false,
typename futures::detail::valueExecutorCallableResult<T, F>:: typename futures::detail::valueExecutorCallableResult<T, F>::
ValueArg>()); ValueArg>());
}; };
using R = futures::detail::tryCallableResult<T, decltype(lambdaFunc)>; using R = futures::detail::tryExecutorCallableResult<T, decltype(lambdaFunc)>;
return this->thenImplementation(std::move(lambdaFunc), R{}); return this->thenImplementation(std::move(lambdaFunc), R{});
} }
...@@ -1219,14 +1244,17 @@ Future<T>::thenError(tag_t<ExceptionType>, F&& func) && { ...@@ -1219,14 +1244,17 @@ Future<T>::thenError(tag_t<ExceptionType>, F&& func) && {
this->setCallback_( this->setCallback_(
[state = futures::detail::makeCoreCallbackState( [state = futures::detail::makeCoreCallbackState(
std::move(p), std::forward<F>(func))](Try<T>&& t) mutable { std::move(p), std::forward<F>(func))](
Executor::KeepAlive<>&&, Try<T>&& t) mutable {
if (auto ex = t.template tryGetExceptionObject< if (auto ex = t.template tryGetExceptionObject<
std::remove_reference_t<ExceptionType>>()) { std::remove_reference_t<ExceptionType>>()) {
auto tf2 = state.tryInvoke(std::move(*ex)); auto tf2 = state.tryInvoke(std::move(*ex));
if (tf2.hasException()) { if (tf2.hasException()) {
state.setException(std::move(tf2.exception())); state.setException(std::move(tf2.exception()));
} else { } else {
tf2->setCallback_([p = state.stealPromise()](Try<T>&& t3) mutable { tf2->setCallback_(
[p = state.stealPromise()](
Executor::KeepAlive<>&&, Try<T>&& t3) mutable {
p.setTry(std::move(t3)); p.setTry(std::move(t3));
}); });
} }
...@@ -1252,7 +1280,7 @@ Future<T>::thenError(tag_t<ExceptionType>, F&& func) && { ...@@ -1252,7 +1280,7 @@ Future<T>::thenError(tag_t<ExceptionType>, F&& func) && {
this->setCallback_([state = futures::detail::makeCoreCallbackState( this->setCallback_([state = futures::detail::makeCoreCallbackState(
std::move(p), std::forward<F>(func))]( std::move(p), std::forward<F>(func))](
Try<T>&& t) mutable { Executor::KeepAlive<>&&, Try<T>&& t) mutable {
if (auto ex = t.template tryGetExceptionObject< if (auto ex = t.template tryGetExceptionObject<
std::remove_reference_t<ExceptionType>>()) { std::remove_reference_t<ExceptionType>>()) {
state.setTry(makeTryWith([&] { return state.invoke(std::move(*ex)); })); state.setTry(makeTryWith([&] { return state.invoke(std::move(*ex)); }));
...@@ -1275,15 +1303,16 @@ Future<T>::thenError(F&& func) && { ...@@ -1275,15 +1303,16 @@ Future<T>::thenError(F&& func) && {
Promise<T> p; Promise<T> p;
auto sf = p.getSemiFuture(); auto sf = p.getSemiFuture();
this->setCallback_( this->setCallback_([state = futures::detail::makeCoreCallbackState(
[state = futures::detail::makeCoreCallbackState( std::move(p), std::forward<F>(func))](
std::move(p), std::forward<F>(func))](Try<T> t) mutable { Executor::KeepAlive<>&&, Try<T> t) mutable {
if (t.hasException()) { if (t.hasException()) {
auto tf2 = state.tryInvoke(std::move(t.exception())); auto tf2 = state.tryInvoke(std::move(t.exception()));
if (tf2.hasException()) { if (tf2.hasException()) {
state.setException(std::move(tf2.exception())); state.setException(std::move(tf2.exception()));
} else { } else {
tf2->setCallback_([p = state.stealPromise()](Try<T>&& t3) mutable { tf2->setCallback_([p = state.stealPromise()](
Executor::KeepAlive<>&&, Try<T>&& t3) mutable {
p.setTry(std::move(t3)); p.setTry(std::move(t3));
}); });
} }
...@@ -1306,12 +1335,12 @@ Future<T>::thenError(F&& func) && { ...@@ -1306,12 +1335,12 @@ Future<T>::thenError(F&& func) && {
Promise<T> p; Promise<T> p;
auto sf = p.getSemiFuture(); auto sf = p.getSemiFuture();
this->setCallback_( this->setCallback_([state = futures::detail::makeCoreCallbackState(
[state = futures::detail::makeCoreCallbackState( std::move(p), std::forward<F>(func))](
std::move(p), std::forward<F>(func))](Try<T>&& t) mutable { Executor::KeepAlive<>&&, Try<T>&& t) mutable {
if (t.hasException()) { if (t.hasException()) {
state.setTry(makeTryWith( state.setTry(
[&] { return state.invoke(std::move(t.exception())); })); makeTryWith([&] { return state.invoke(std::move(t.exception())); }));
} else { } else {
state.setTry(std::move(t)); state.setTry(std::move(t));
} }
...@@ -1505,7 +1534,7 @@ collectAllSemiFuture(Fs&&... fs) { ...@@ -1505,7 +1534,7 @@ collectAllSemiFuture(Fs&&... fs) {
auto ctx = std::make_shared<Context>(); auto ctx = std::make_shared<Context>();
futures::detail::foreach( futures::detail::foreach(
[&](auto i, auto&& f) { [&](auto i, auto&& f) {
f.setCallback_([i, ctx](auto&& t) { f.setCallback_([i, ctx](auto&&, auto&& t) {
std::get<i.value>(ctx->results) = std::move(t); std::get<i.value>(ctx->results) = std::move(t);
}); });
}, },
...@@ -1554,8 +1583,9 @@ collectAllSemiFuture(InputIterator first, InputIterator last) { ...@@ -1554,8 +1583,9 @@ collectAllSemiFuture(InputIterator first, InputIterator last) {
auto ctx = std::make_shared<Context>(size_t(std::distance(first, last))); auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)));
for (size_t i = 0; first != last; ++first, ++i) { for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_( first->setCallback_([i, ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
[i, ctx](Try<T>&& t) { ctx->results[i] = std::move(t); }); ctx->results[i] = std::move(t);
});
} }
auto future = ctx->p.getSemiFuture(); auto future = ctx->p.getSemiFuture();
...@@ -1614,7 +1644,7 @@ collectSemiFuture(InputIterator first, InputIterator last) { ...@@ -1614,7 +1644,7 @@ collectSemiFuture(InputIterator first, InputIterator last) {
auto ctx = std::make_shared<Context>(std::distance(first, last)); auto ctx = std::make_shared<Context>(std::distance(first, last));
for (size_t i = 0; first != last; ++first, ++i) { for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<T>&& t) { first->setCallback_([i, ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
if (t.hasException()) { if (t.hasException()) {
if (!ctx->threw.exchange(true, std::memory_order_relaxed)) { if (!ctx->threw.exchange(true, std::memory_order_relaxed)) {
ctx->p.setException(std::move(t.exception())); ctx->p.setException(std::move(t.exception()));
...@@ -1669,7 +1699,7 @@ collectSemiFuture(Fs&&... fs) { ...@@ -1669,7 +1699,7 @@ collectSemiFuture(Fs&&... fs) {
auto ctx = std::make_shared<Context>(); auto ctx = std::make_shared<Context>();
futures::detail::foreach( futures::detail::foreach(
[&](auto i, auto&& f) { [&](auto i, auto&& f) {
f.setCallback_([i, ctx](auto&& t) { f.setCallback_([i, ctx](Executor::KeepAlive<>&&, auto&& t) {
if (t.hasException()) { if (t.hasException()) {
if (!ctx->threw.exchange(true, std::memory_order_relaxed)) { if (!ctx->threw.exchange(true, std::memory_order_relaxed)) {
ctx->p.setException(std::move(t.exception())); ctx->p.setException(std::move(t.exception()));
...@@ -1735,7 +1765,7 @@ collectAnySemiFuture(InputIterator first, InputIterator last) { ...@@ -1735,7 +1765,7 @@ collectAnySemiFuture(InputIterator first, InputIterator last) {
auto ctx = std::make_shared<Context>(); auto ctx = std::make_shared<Context>();
for (size_t i = 0; first != last; ++first, ++i) { for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<T>&& t) { first->setCallback_([i, ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
if (!ctx->done.exchange(true, std::memory_order_relaxed)) { if (!ctx->done.exchange(true, std::memory_order_relaxed)) {
ctx->p.setValue(std::make_pair(i, std::move(t))); ctx->p.setValue(std::make_pair(i, std::move(t)));
} }
...@@ -1777,7 +1807,7 @@ collectAnyWithoutException(InputIterator first, InputIterator last) { ...@@ -1777,7 +1807,7 @@ collectAnyWithoutException(InputIterator first, InputIterator last) {
auto ctx = std::make_shared<Context>(size_t(std::distance(first, last))); auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)));
for (size_t i = 0; first != last; ++first, ++i) { for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<T>&& t) { first->setCallback_([i, ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
if (!t.hasException() && if (!t.hasException() &&
!ctx->done.exchange(true, std::memory_order_relaxed)) { !ctx->done.exchange(true, std::memory_order_relaxed)) {
ctx->p.setValue(std::make_pair(i, std::move(t.value()))); ctx->p.setValue(std::make_pair(i, std::move(t.value())));
...@@ -1840,7 +1870,7 @@ collectN(InputIterator first, InputIterator last, size_t n) { ...@@ -1840,7 +1870,7 @@ collectN(InputIterator first, InputIterator last, size_t n) {
// vector // vector
auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)), n); auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)), n);
for (size_t i = 0; first != last; ++first, ++i) { for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<T>&& t) { first->setCallback_([i, ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
// relaxed because this guards control but does not guard data // relaxed because this guards control but does not guard data
auto const c = 1 + ctx->completed.fetch_add(1, std::memory_order_relaxed); auto const c = 1 + ctx->completed.fetch_add(1, std::memory_order_relaxed);
if (c > ctx->min) { if (c > ctx->min) {
...@@ -1954,7 +1984,8 @@ window(Executor::KeepAlive<> executor, Collection input, F func, size_t n) { ...@@ -1954,7 +1984,8 @@ window(Executor::KeepAlive<> executor, Collection input, F func, size_t n) {
[&] { return ctx->func(std::move(ctx->input[i])); }) [&] { return ctx->func(std::move(ctx->input[i])); })
.via(ctx->executor.get()); .via(ctx->executor.get());
fut.setCallback_([ctx = std::move(ctx), i](Try<Result>&& t) mutable { fut.setCallback_([ctx = std::move(ctx), i](
Executor::KeepAlive<>&&, Try<Result>&& t) mutable {
ctx->promises[i].setTry(std::move(t)); ctx->promises[i].setTry(std::move(t));
// Chain another future onto this one // Chain another future onto this one
spawn(std::move(ctx)); spawn(std::move(ctx));
...@@ -2036,14 +2067,16 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) { ...@@ -2036,14 +2067,16 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
} }
void operator()(Promise<T>&& p, Future<T>&& f) const { void operator()(Promise<T>&& p, Future<T>&& f) const {
f.setCallback_( f.setCallback_(
[p = std::move(p)](Try<T>&& t) mutable { p.setTry(std::move(t)); }); [p = std::move(p)](Executor::KeepAlive<>&&, Try<T>&& t) mutable {
p.setTry(std::move(t));
});
} }
}; };
auto ctx = std::make_shared<Context>( auto ctx = std::make_shared<Context>(
std::move(initial), std::move(func), std::distance(first, last)); std::move(initial), std::move(func), std::distance(first, last));
for (size_t i = 0; first != last; ++first, ++i) { for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<ItT>&& t) { first->setCallback_([i, ctx](Executor::KeepAlive<>&&, Try<ItT>&& t) {
(void)i; (void)i;
// Futures can be completed in any order, simultaneously. // Futures can be completed in any order, simultaneously.
// To make this non-blocking, we create a new Future chain in // To make this non-blocking, we create a new Future chain in
...@@ -2057,12 +2090,14 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) { ...@@ -2057,12 +2090,14 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
f = std::exchange(ctx->memo_, std::move(f)); f = std::exchange(ctx->memo_, std::move(f));
if (++ctx->numThens_ == ctx->numFutures_) { if (++ctx->numThens_ == ctx->numFutures_) {
// After reducing the value of the last Future, fulfill the Promise // After reducing the value of the last Future, fulfill the Promise
ctx->memo_.setCallback_( ctx->memo_.setCallback_([ctx](Executor::KeepAlive<>&&, Try<T>&& t2) {
[ctx](Try<T>&& t2) { ctx->promise_.setValue(std::move(t2)); }); ctx->promise_.setValue(std::move(t2));
});
} }
} }
f.setCallback_( f.setCallback_(
[ctx, mp = std::move(p), mt = std::move(t)](Try<T>&& v) mutable { [ctx, mp = std::move(p), mt = std::move(t)](
Executor::KeepAlive<>&&, Try<T>&& v) mutable {
if (v.hasValue()) { if (v.hasValue()) {
try { try {
Fulfill{}( Fulfill{}(
...@@ -2132,7 +2167,8 @@ void waitImpl(FutureType& f) { ...@@ -2132,7 +2167,8 @@ void waitImpl(FutureType& f) {
Promise<T> promise; Promise<T> promise;
auto ret = convertFuture(promise.getSemiFuture(), f); auto ret = convertFuture(promise.getSemiFuture(), f);
auto baton = std::make_shared<FutureBatonType>(); auto baton = std::make_shared<FutureBatonType>();
f.setCallback_([baton, promise = std::move(promise)](Try<T>&& t) mutable { f.setCallback_([baton, promise = std::move(promise)](
Executor::KeepAlive<>&&, Try<T>&& t) mutable {
promise.setTry(std::move(t)); promise.setTry(std::move(t));
baton->post(); baton->post();
}); });
...@@ -2166,7 +2202,8 @@ void waitImpl(FutureType& f, Duration dur) { ...@@ -2166,7 +2202,8 @@ void waitImpl(FutureType& f, Duration dur) {
Promise<T> promise; Promise<T> promise;
auto ret = convertFuture(promise.getSemiFuture(), f); auto ret = convertFuture(promise.getSemiFuture(), f);
auto baton = std::make_shared<FutureBatonType>(); auto baton = std::make_shared<FutureBatonType>();
f.setCallback_([baton, promise = std::move(promise)](Try<T>&& t) mutable { f.setCallback_([baton, promise = std::move(promise)](
Executor::KeepAlive<>&&, Try<T>&& t) mutable {
promise.setTry(std::move(t)); promise.setTry(std::move(t));
baton->post(); baton->post();
}); });
...@@ -2229,7 +2266,9 @@ SemiFuture<T>& SemiFuture<T>::wait() & { ...@@ -2229,7 +2266,9 @@ SemiFuture<T>& SemiFuture<T>::wait() & {
Promise<T> promise; Promise<T> promise;
auto ret = promise.getSemiFuture(); auto ret = promise.getSemiFuture();
setCallback_( setCallback_(
[p = std::move(promise)](auto&& r) mutable { p.setTry(std::move(r)); }); [p = std::move(promise)](Executor::KeepAlive<>&&, auto&& r) mutable {
p.setTry(std::move(r));
});
auto waitExecutor = futures::detail::WaitExecutor::create(); auto waitExecutor = futures::detail::WaitExecutor::create();
deferredExecutor->setExecutor(waitExecutor.copy()); deferredExecutor->setExecutor(waitExecutor.copy());
while (!ret.isReady()) { while (!ret.isReady()) {
...@@ -2257,7 +2296,9 @@ SemiFuture<T>& SemiFuture<T>::wait(Duration dur) & { ...@@ -2257,7 +2296,9 @@ SemiFuture<T>& SemiFuture<T>::wait(Duration dur) & {
Promise<T> promise; Promise<T> promise;
auto ret = promise.getSemiFuture(); auto ret = promise.getSemiFuture();
setCallback_( setCallback_(
[p = std::move(promise)](auto&& r) mutable { p.setTry(std::move(r)); }); [p = std::move(promise)](Executor::KeepAlive<>&&, auto&& r) mutable {
p.setTry(std::move(r));
});
auto waitExecutor = futures::detail::WaitExecutor::create(); auto waitExecutor = futures::detail::WaitExecutor::create();
auto deadline = futures::detail::WaitExecutor::Clock::now() + dur; auto deadline = futures::detail::WaitExecutor::Clock::now() + dur;
deferredExecutor->setExecutor(waitExecutor.copy()); deferredExecutor->setExecutor(waitExecutor.copy());
......
...@@ -125,6 +125,20 @@ struct callableResult { ...@@ -125,6 +125,20 @@ struct callableResult {
typedef Future<typename ReturnsFuture::Inner> Return; typedef Future<typename ReturnsFuture::Inner> Return;
}; };
template <typename T, typename F>
struct executorCallableResult {
typedef typename std::conditional<
is_invocable<F, Executor::KeepAlive<>&&>::value,
detail::argResult<false, F, Executor::KeepAlive<>&&>,
typename std::conditional<
is_invocable<F, Executor::KeepAlive<>&&, T&&>::value,
detail::argResult<false, F, Executor::KeepAlive<>&&, T&&>,
detail::argResult<true, F, Executor::KeepAlive<>&&, Try<T>&&>>::
type>::type Arg;
typedef isFutureOrSemiFuture<typename Arg::Result> ReturnsFuture;
typedef Future<typename ReturnsFuture::Inner> Return;
};
template < template <
typename T, typename T,
typename F, typename F,
...@@ -141,8 +155,7 @@ template < ...@@ -141,8 +155,7 @@ template <
typename F, typename F,
typename = std::enable_if_t<is_invocable<F, Executor*, Try<T>&&>::value>> typename = std::enable_if_t<is_invocable<F, Executor*, Try<T>&&>::value>>
struct tryExecutorCallableResult { struct tryExecutorCallableResult {
typedef detail::argResult<true, F, const Executor::KeepAlive<>&, Try<T>&&> typedef detail::argResult<true, F, Executor::KeepAlive<>&&, Try<T>&&> Arg;
Arg;
typedef isFutureOrSemiFuture<typename Arg::Result> ReturnsFuture; typedef isFutureOrSemiFuture<typename Arg::Result> ReturnsFuture;
typedef typename ReturnsFuture::Inner value_type; typedef typename ReturnsFuture::Inner value_type;
typedef Future<value_type> Return; typedef Future<value_type> Return;
...@@ -159,7 +172,7 @@ struct valueCallableResult { ...@@ -159,7 +172,7 @@ struct valueCallableResult {
template <typename T, typename F> template <typename T, typename F>
struct valueExecutorCallableResult { struct valueExecutorCallableResult {
typedef detail::argResult<false, F, const Executor::KeepAlive<>&, T&&> Arg; typedef detail::argResult<false, F, Executor::KeepAlive<>&&, T&&> Arg;
typedef isFutureOrSemiFuture<typename Arg::Result> ReturnsFuture; typedef isFutureOrSemiFuture<typename Arg::Result> ReturnsFuture;
typedef typename ReturnsFuture::Inner value_type; typedef typename ReturnsFuture::Inner value_type;
typedef typename Arg::ArgList::Tail::FirstArg ValueArg; typedef typename Arg::ArgList::Tail::FirstArg ValueArg;
...@@ -203,6 +216,28 @@ struct Extract<R (&)(Args...)> { ...@@ -203,6 +216,28 @@ struct Extract<R (&)(Args...)> {
class DeferredExecutor; class DeferredExecutor;
template <class T, class F>
auto makeExecutorLambda(
F&& func,
typename std::enable_if<is_invocable<F>::value, int>::type = 0) {
return
[func = std::forward<F>(func)](Executor::KeepAlive<>&&, auto&&) mutable {
return std::forward<F>(func)();
};
}
template <class T, class F>
auto makeExecutorLambda(
F&& func,
typename std::enable_if<!is_invocable<F>::value, int>::type = 0) {
using R = futures::detail::callableResult<T, F&&>;
return [func = std::forward<F>(func)](
Executor::KeepAlive<>&&,
typename R::Arg::ArgList::FirstArg&& param) mutable {
return std::forward<F>(func)(std::forward<decltype(param)>(param));
};
}
} // namespace detail } // namespace detail
} // namespace futures } // namespace futures
......
...@@ -1173,16 +1173,22 @@ class Future : private futures::detail::FutureBase<T> { ...@@ -1173,16 +1173,22 @@ class Future : private futures::detail::FutureBase<T> {
auto then(Executor::KeepAlive<> x, Arg&& arg) && { auto then(Executor::KeepAlive<> x, Arg&& arg) && {
auto oldX = getKeepAliveToken(this->getExecutor()); auto oldX = getKeepAliveToken(this->getExecutor());
this->setExecutor(std::move(x)); this->setExecutor(std::move(x));
// TODO(T29171940): thenImplementation here is ambiguous // TODO(T29171940): thenImplementation here is ambiguous
// as then used to be but that is better than keeping then in the public // as then used to be but that is better than keeping then in the public
// API. // API.
using R = futures::detail::callableResult<T, Arg&&>; auto lambdaFunc =
futures::detail::makeExecutorLambda<T>(std::forward<Arg>(arg));
using R = futures::detail::executorCallableResult<T, decltype(lambdaFunc)>;
return std::move(*this) return std::move(*this)
.thenImplementation(std::forward<Arg>(arg), R{}) .thenImplementation(std::move(lambdaFunc), R{})
.via(std::move(oldX)); .via(std::move(oldX));
} }
template <typename R, typename... Args>
auto then(Executor::KeepAlive<>&& x, R (&func)(Args...)) && {
return std::move(*this).then(std::move(x), &func);
}
/// When this Future has completed, execute func which is a function that /// When this Future has completed, execute func which is a function that
/// can be called with `Try<T>&&` (often a lambda with parameter type /// can be called with `Try<T>&&` (often a lambda with parameter type
/// `auto&&` or `auto`). /// `auto&&` or `auto`).
...@@ -1910,7 +1916,8 @@ class FutureAwaitable { ...@@ -1910,7 +1916,8 @@ class FutureAwaitable {
// Make sure the future object doesn't get destroyed until setCallback_ // Make sure the future object doesn't get destroyed until setCallback_
// returns. // returns.
auto future = std::move(future_); auto future = std::move(future_);
future.setCallback_([this, h](Try<T>&& result) mutable { future.setCallback_(
[this, h](Executor::KeepAlive<>&&, Try<T>&& result) mutable {
result_ = std::move(result); result_ = std::move(result);
h.resume(); h.resume();
}); });
......
...@@ -199,7 +199,7 @@ class Core final { ...@@ -199,7 +199,7 @@ class Core final {
public: public:
using Result = Try<T>; using Result = Try<T>;
using Callback = folly::Function<void(Result&&)>; using Callback = folly::Function<void(Executor::KeepAlive<>&&, Result&&)>;
/// State will be Start /// State will be Start
static Core* make() { static Core* make() {
...@@ -587,7 +587,7 @@ class Core final { ...@@ -587,7 +587,7 @@ class Core final {
auto cr = std::move(core_ref); auto cr = std::move(core_ref);
Core* const core = cr.getCore(); Core* const core = cr.getCore();
RequestContextScopeGuard rctx(std::move(core->context_)); RequestContextScopeGuard rctx(std::move(core->context_));
core->callback_(std::move(core->result_)); core->callback_(std::move(keepAlive), std::move(core->result_));
}); });
} catch (const std::exception& e) { } catch (const std::exception& e) {
ew = exception_wrapper(std::current_exception(), e); ew = exception_wrapper(std::current_exception(), e);
...@@ -597,7 +597,7 @@ class Core final { ...@@ -597,7 +597,7 @@ class Core final {
if (ew) { if (ew) {
RequestContextScopeGuard rctx(std::move(context_)); RequestContextScopeGuard rctx(std::move(context_));
result_ = Try<T>(std::move(ew)); result_ = Try<T>(std::move(ew));
callback_(std::move(result_)); callback_(Executor::KeepAlive<>{}, std::move(result_));
} }
} else { } else {
attached_.fetch_add(1, std::memory_order_relaxed); attached_.fetch_add(1, std::memory_order_relaxed);
...@@ -607,7 +607,7 @@ class Core final { ...@@ -607,7 +607,7 @@ class Core final {
detachOne(); detachOne();
}; };
RequestContextScopeGuard rctx(std::move(context_)); RequestContextScopeGuard rctx(std::move(context_));
callback_(std::move(result_)); callback_(Executor::KeepAlive<>{}, std::move(result_));
} }
} }
......
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