Commit 7bcd14c5 authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook Github Bot

Implement tail call optimization

Summary: This adds an optimization for cases where recursive .then() is used to implement a loop. Instead of creating a chain of Cores that fulfil each other, we steal a callback from previous core if possible.

Reviewed By: yfeldblum

Differential Revision: D10392101

fbshipit-source-id: 008c644df7472f5e280b24a41c62ee7199954abc
parent 6fc4b643
...@@ -266,8 +266,16 @@ void FutureBase<T>::raise(exception_wrapper exception) { ...@@ -266,8 +266,16 @@ void FutureBase<T>::raise(exception_wrapper exception) {
template <class T> template <class T>
template <class F> template <class F>
void FutureBase<T>::setCallback_(F&& func) { void FutureBase<T>::setCallback_(F&& func) {
setCallback_(std::forward<F>(func), RequestContext::saveContext());
}
template <class T>
template <class F>
void FutureBase<T>::setCallback_(
F&& func,
std::shared_ptr<folly::RequestContext> context) {
throwIfContinued(); throwIfContinued();
getCore().setCallback(std::forward<F>(func)); getCore().setCallback(std::forward<F>(func), std::move(context));
} }
template <class T> template <class T>
...@@ -415,9 +423,17 @@ FutureBase<T>::thenImplementation(F&& func, R) { ...@@ -415,9 +423,17 @@ FutureBase<T>::thenImplementation(F&& func, R) {
auto statePromise = state.stealPromise(); auto statePromise = state.stealPromise();
auto tf3 = auto tf3 =
chainExecutor(statePromise.core_->getExecutor(), *std::move(tf2)); chainExecutor(statePromise.core_->getExecutor(), *std::move(tf2));
tf3.setCallback_([p2 = std::move(statePromise)](Try<B>&& b) mutable { if (std::is_same<T, B>::value && statePromise.getCore().hasCallback()) {
p2.setTry(std::move(b)); tf3.core_->setExecutor(statePromise.core_->getExecutor());
}); auto callbackAndContext = statePromise.getCore().stealCallback();
tf3.setCallback_(
std::move(callbackAndContext.first),
std::move(callbackAndContext.second));
} else {
tf3.setCallback_([p2 = std::move(statePromise)](Try<B>&& b) mutable {
p2.setTry(std::move(b));
});
}
} }
} }
}); });
...@@ -2086,13 +2102,18 @@ SemiFuture<T>& SemiFuture<T>::wait() & { ...@@ -2086,13 +2102,18 @@ SemiFuture<T>& SemiFuture<T>::wait() & {
if (auto deferredExecutor = getDeferredExecutor()) { if (auto deferredExecutor = getDeferredExecutor()) {
// Make sure that the last callback in the future chain will be run on the // Make sure that the last callback in the future chain will be run on the
// WaitExecutor. // WaitExecutor.
setCallback_([](auto&&) {}); Promise<T> promise;
auto ret = promise.getSemiFuture();
setCallback_(
[p = std::move(promise)](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 (!isReady()) { while (!ret.isReady()) {
waitExecutor->drive(); waitExecutor->drive();
} }
waitExecutor->detach(); waitExecutor->detach();
this->detach();
*this = std::move(ret);
} else { } else {
futures::detail::waitImpl(*this); futures::detail::waitImpl(*this);
} }
...@@ -2109,16 +2130,21 @@ SemiFuture<T>& SemiFuture<T>::wait(Duration dur) & { ...@@ -2109,16 +2130,21 @@ SemiFuture<T>& SemiFuture<T>::wait(Duration dur) & {
if (auto deferredExecutor = getDeferredExecutor()) { if (auto deferredExecutor = getDeferredExecutor()) {
// Make sure that the last callback in the future chain will be run on the // Make sure that the last callback in the future chain will be run on the
// WaitExecutor. // WaitExecutor.
setCallback_([](auto&&) {}); Promise<T> promise;
auto ret = promise.getSemiFuture();
setCallback_(
[p = std::move(promise)](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());
while (!isReady()) { while (!ret.isReady()) {
if (!waitExecutor->driveUntil(deadline)) { if (!waitExecutor->driveUntil(deadline)) {
break; break;
} }
} }
waitExecutor->detach(); waitExecutor->detach();
this->detach();
*this = std::move(ret);
} else { } else {
futures::detail::waitImpl(*this, dur); futures::detail::waitImpl(*this, dur);
} }
......
...@@ -258,6 +258,9 @@ class FutureBase { ...@@ -258,6 +258,9 @@ class FutureBase {
template <class F> template <class F>
void setCallback_(F&& func); void setCallback_(F&& func);
template <class F>
void setCallback_(F&& func, std::shared_ptr<folly::RequestContext> context);
/// Provides a threadsafe back-channel so the consumer's thread can send an /// Provides a threadsafe back-channel so the consumer's thread can send an
/// interrupt-object to the producer's thread. /// interrupt-object to the producer's thread.
/// ///
...@@ -2037,14 +2040,18 @@ class FutureAwaitable { ...@@ -2037,14 +2040,18 @@ class FutureAwaitable {
} }
T await_resume() { T await_resume() {
return std::move(future_).value(); return std::move(result_).value();
} }
void await_suspend(std::experimental::coroutine_handle<> h) { void await_suspend(std::experimental::coroutine_handle<> h) {
future_.setCallback_([h](Try<T>&&) mutable { h.resume(); }); future_.setCallback_([this, h](Try<T>&& result) mutable {
result_ = std::move(result);
h.resume();
});
} }
private: private:
folly::Try<T> result_;
folly::Future<T> future_; folly::Future<T> future_;
}; };
......
...@@ -112,17 +112,19 @@ static_assert(sizeof(SpinLock) == 1, "missized"); ...@@ -112,17 +112,19 @@ static_assert(sizeof(SpinLock) == 1, "missized");
/// The FSM to manage the primary producer-to-consumer info-flow has these /// The FSM to manage the primary producer-to-consumer info-flow has these
/// allowed (atomic) transitions: /// allowed (atomic) transitions:
/// ///
/// +-------------------------------------------------------------+ /// +----------------------------------------------------------------+
/// | ---> OnlyResult ----- | /// | ---> OnlyResult ----- |
/// | / \ | /// | / \ |
/// | (setResult()) (setCallback()) | /// | (setResult()) (setCallback()) |
/// | / \ | /// | / \ |
/// | Start -----> ------> Done | /// | ---> Start ---> ------> Done |
/// | \ / | /// | \ \ / |
/// | (setCallback()) (setResult()) | /// | \ (setCallback()) (setResult()) |
/// | \ / | /// | \ \ / |
/// | ---> OnlyCallback --- | /// | \ ---> OnlyCallback --- |
/// +-------------------------------------------------------------+ /// | \ / |
/// | <-- (stealCallback()) - |
/// +----------------------------------------------------------------+
/// ///
/// States and the corresponding producer-to-consumer data status & ownership: /// States and the corresponding producer-to-consumer data status & ownership:
/// ///
...@@ -141,6 +143,8 @@ static_assert(sizeof(SpinLock) == 1, "missized"); ...@@ -141,6 +143,8 @@ static_assert(sizeof(SpinLock) == 1, "missized");
/// point forward only the producer thread can safely access that callback /// point forward only the producer thread can safely access that callback
/// (see `setResult()` and `doCallback()` where the producer thread can both /// (see `setResult()` and `doCallback()` where the producer thread can both
/// read and modify the callback). /// read and modify the callback).
/// Alternatively producer thread can also steal the callback (see
/// `stealCallback()`).
/// - Done: callback can be safely accessed only within `doCallback()`, which /// - Done: callback can be safely accessed only within `doCallback()`, which
/// gets called on exactly one thread exactly once just after the transition /// gets called on exactly one thread exactly once just after the transition
/// to Done. The future object will have determined whether that callback /// to Done. The future object will have determined whether that callback
...@@ -183,6 +187,9 @@ class Core final { ...@@ -183,6 +187,9 @@ class Core final {
"void futures are not supported. Use Unit instead."); "void futures are not supported. Use Unit instead.");
public: public:
using Result = Try<T>;
using Callback = folly::Function<void(Result&&)>;
/// State will be Start /// State will be Start
static Core* make() { static Core* make() {
return new Core(); return new Core();
...@@ -272,12 +279,12 @@ class Core final { ...@@ -272,12 +279,12 @@ class Core final {
/// and might also synchronously execute that callback (e.g., if there is no /// and might also synchronously execute that callback (e.g., if there is no
/// executor or if the executor is inline). /// executor or if the executor is inline).
template <typename F> template <typename F>
void setCallback(F&& func) { void setCallback(F&& func, std::shared_ptr<folly::RequestContext> context) {
DCHECK(!hasCallback()); DCHECK(!hasCallback());
// construct callback_ first; if that fails, context_ will not leak // construct callback_ first; if that fails, context_ will not leak
::new (&callback_) Callback(std::forward<F>(func)); ::new (&callback_) Callback(std::forward<F>(func));
::new (&context_) Context(RequestContext::saveContext()); ::new (&context_) Context(std::move(context));
auto state = state_.load(std::memory_order_acquire); auto state = state_.load(std::memory_order_acquire);
while (true) { while (true) {
...@@ -304,6 +311,24 @@ class Core final { ...@@ -304,6 +311,24 @@ class Core final {
} }
} }
/// Call only from producer thread.
/// Call only once - else undefined behavior.
///
/// See FSM graph for allowed transitions.
///
/// Call only if hasCallback() == true && hasResult() == false
/// This can not be called concurrently with setResult().
///
/// Extracts callback from the Core and transitions it back into Start state.
std::pair<Callback, std::shared_ptr<folly::RequestContext>> stealCallback() {
DCHECK(state_.load(std::memory_order_relaxed) == State::OnlyCallback);
auto ret = std::make_pair(std::move(callback_), std::move(context_));
context_.~Context();
callback_.~Callback();
state_.store(State::Start, std::memory_order_relaxed);
return ret;
}
/// Call only from producer thread. /// Call only from producer thread.
/// Call only once - else undefined behavior. /// Call only once - else undefined behavior.
/// ///
...@@ -569,8 +594,6 @@ class Core final { ...@@ -569,8 +594,6 @@ class Core final {
} }
} }
using Result = Try<T>;
using Callback = folly::Function<void(Result&&)>;
using Context = std::shared_ptr<RequestContext>; using Context = std::shared_ptr<RequestContext>;
union { union {
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include <folly/Memory.h> #include <folly/Memory.h>
#include <folly/Unit.h> #include <folly/Unit.h>
#include <folly/dynamic.h> #include <folly/dynamic.h>
#include <folly/executors/ManualExecutor.h>
#include <folly/portability/GTest.h> #include <folly/portability/GTest.h>
#include <folly/synchronization/Baton.h> #include <folly/synchronization/Baton.h>
...@@ -1601,3 +1602,23 @@ TEST(Future, makePromiseContract) { ...@@ -1601,3 +1602,23 @@ TEST(Future, makePromiseContract) {
ASSERT_TRUE(c.second.isReady()); ASSERT_TRUE(c.second.isReady());
EXPECT_EQ(4, std::move(c.second).get()); EXPECT_EQ(4, std::move(c.second).get());
} }
Future<int> call(int depth) {
return makeFuture(depth == 0 ? 42 : 0);
}
Future<int> recursion(Executor* executor, int depth) {
return call(depth).then(executor, [=](int result) {
if (result) {
return folly::makeFuture(result);
}
return recursion(executor, depth - 1);
});
}
TEST(Future, ThenRecursion) {
ManualExecutor executor;
EXPECT_EQ(42, recursion(&executor, 100000).getVia(&executor));
}
...@@ -444,7 +444,6 @@ TEST(SemiFuture, MakeFutureFromSemiFutureReturnSemiFuture) { ...@@ -444,7 +444,6 @@ TEST(SemiFuture, MakeFutureFromSemiFutureReturnSemiFuture) {
p.setValue(42); p.setValue(42);
e.loopOnce(); e.loopOnce();
e.loopOnce(); e.loopOnce();
e.loopOnce();
EXPECT_TRUE(future.isReady()); EXPECT_TRUE(future.isReady());
ASSERT_EQ(future.value(), 42); ASSERT_EQ(future.value(), 42);
ASSERT_EQ(result, 42); ASSERT_EQ(result, 42);
......
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