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

Add inline forms of future continuation operations.

Summary:
Adds inline-supporting forms of thenValue and thenTry. Carries this information in the Core through a variant of the HasCallback state. Dispatches the continuation inline rather than on the executor if that state is set.

Propagate completing executor through intermediate promise and make inline execution conditional on matching executors.

Only continue inline if the executor of the completing task matches that of the task-to-be-enqueued, to ensure that calls to via always trigger reposting and that there are no surprises where work runs on the wrong executor.

Reviewed By: yfeldblum

Differential Revision: D15292236

fbshipit-source-id: 51ddd5cfd1196ea9e9f408e2b1337ff9d4fa5b9f
parent 263ebaa1
This diff is collapsed.
......@@ -258,7 +258,7 @@ class FutureBase {
/// not worth listing all those and their fancy template signatures as
/// friends. But it's not for public consumption.
template <class F>
void setCallback_(F&& func);
void setCallback_(F&& func, InlineContinuation = InlineContinuation::forbid);
/// Provides a threadsafe back-channel so the consumer's thread can send an
/// interrupt-object to the producer's thread.
......@@ -418,13 +418,13 @@ class FutureBase {
// e.g. f.thenTry([](Try<T> t){ return t.value(); });
template <typename F, typename R>
typename std::enable_if<!R::ReturnsFuture::value, typename R::Return>::type
thenImplementation(F&& func, R);
thenImplementation(F&& func, R, InlineContinuation);
// Variant: returns a Future
// e.g. f.thenTry([](Try<T> t){ return makeFuture<T>(t); });
template <typename F, typename R>
typename std::enable_if<R::ReturnsFuture::value, typename R::Return>::type
thenImplementation(F&& func, R);
thenImplementation(F&& func, R, InlineContinuation);
template <typename E>
SemiFuture<T> withinImplementation(Duration dur, E e, Timekeeper* tk) &&;
......@@ -1103,8 +1103,17 @@ class Future : private futures::detail::FutureBase<T> {
///
/// Func shall return either another Future or a value.
///
/// thenInline will run the continuation inline with the execution of the
/// previous callback in the chain if the callback attached to the previous
/// future that triggers execution of func runs on the same executor that func
/// would be executed on.
///
/// A Future for the return type of func is returned.
///
/// Versions of these functions with Inline in the name will run the
/// continuation inline if the executor the previous task completes on matches
/// the executor the next is to be enqueued on to.
///
/// Future<string> f2 = f1.thenTry([](Try<T>&&) { return string("foo"); });
///
/// Preconditions:
......@@ -1121,6 +1130,11 @@ class Future : private futures::detail::FutureBase<T> {
F&& func) && {
return std::move(*this).thenTry(std::forward<F>(func));
}
template <typename F>
Future<typename futures::detail::tryCallableResult<T, F>::value_type>
thenInline(F&& func) && {
return std::move(*this).thenTryInline(std::forward<F>(func));
}
/// Variant where func is an member function
///
......@@ -1180,7 +1194,10 @@ class Future : private futures::detail::FutureBase<T> {
futures::detail::makeExecutorLambda<T>(std::forward<Arg>(arg));
using R = futures::detail::executorCallableResult<T, decltype(lambdaFunc)>;
return std::move(*this)
.thenImplementation(std::move(lambdaFunc), R{})
.thenImplementation(
std::move(lambdaFunc),
R{},
futures::detail::InlineContinuation::forbid)
.via(std::move(oldX));
}
......@@ -1195,6 +1212,12 @@ class Future : private futures::detail::FutureBase<T> {
///
/// Func shall return either another Future or a value.
///
/// Versions of these functions with Inline in the name will run the
/// continuation inline with the execution of the previous callback in the
/// chain if the callback attached to the previous future that triggers
/// execution of func runs on the same executor that func would be executed
/// on.
///
/// A Future for the return type of func is returned.
///
/// Future<string> f2 = std::move(f1).thenTry([](auto&& t) {
......@@ -1214,21 +1237,40 @@ class Future : private futures::detail::FutureBase<T> {
Future<typename futures::detail::tryCallableResult<T, F>::value_type> thenTry(
F&& func) &&;
template <typename F>
Future<typename futures::detail::tryCallableResult<T, F>::value_type>
thenTryInline(F&& func) &&;
template <typename F>
Future<typename futures::detail::tryExecutorCallableResult<T, F>::value_type>
thenExTry(F&& func) &&;
template <typename F>
Future<typename futures::detail::tryExecutorCallableResult<T, F>::value_type>
thenExTryInline(F&& func) &&;
template <typename R, typename... Args>
auto thenTry(R (&func)(Args...)) && {
return std::move(*this).thenTry(&func);
}
template <typename R, typename... Args>
auto thenTryInline(R (&func)(Args...)) && {
return std::move(*this).thenTryInline(&func);
}
/// When this Future has completed, execute func which is a function that
/// can be called with `T&&` (often a lambda with parameter type
/// `auto&&` or `auto`).
///
/// Func shall return either another Future or a value.
///
/// Versions of these functions with Inline in the name will run the
/// continuation inline with the execution of the previous callback in the
/// chain if the callback attached to the previous future that triggers
/// execution of func runs on the same executor that func would be executed
/// on.
///
/// A Future for the return type of func is returned.
///
/// Future<string> f2 = f1.thenValue([](auto&& v) {
......@@ -1248,16 +1290,30 @@ class Future : private futures::detail::FutureBase<T> {
Future<typename futures::detail::valueCallableResult<T, F>::value_type>
thenValue(F&& func) &&;
template <typename F>
Future<typename futures::detail::valueCallableResult<T, F>::value_type>
thenValueInline(F&& func) &&;
template <typename F>
Future<
typename futures::detail::valueExecutorCallableResult<T, F>::value_type>
thenExValue(F&& func) &&;
template <typename F>
Future<
typename futures::detail::valueExecutorCallableResult<T, F>::value_type>
thenExValueInline(F&& func) &&;
template <typename R, typename... Args>
auto thenValue(R (&func)(Args...)) && {
return std::move(*this).thenValue(&func);
}
template <typename R, typename... Args>
auto thenValueInline(R (&func)(Args...)) && {
return std::move(*this).thenValueInline(&func);
}
/// Set an error continuation for this Future where the continuation can
/// be called with a known exception type and returns a `T`, `Future<T>`, or
/// `SemiFuture<T>`.
......
......@@ -126,6 +126,12 @@ void Promise<T>::setTry(Try<T>&& t) {
core_->setResult(std::move(t));
}
template <class T>
void Promise<T>::setTry(Executor::KeepAlive<>&& ka, Try<T>&& t) {
throwIfFulfilled();
core_->setResult(std::move(ka), std::move(t));
}
template <class T>
template <class M>
void Promise<T>::setValue(M&& v) {
......
......@@ -433,6 +433,10 @@ class Promise {
return *core;
}
/// Fulfill the Promise with the specified Try (value or exception) and
/// propagate the completing executor.
void setTry(Executor::KeepAlive<>&& ka, Try<T>&& t);
// shared core state object
// usually you should use `getCore()` instead of directly accessing `core_`.
Core* core_;
......
......@@ -28,6 +28,7 @@
#include <folly/ScopeGuard.h>
#include <folly/Try.h>
#include <folly/Utility.h>
#include <folly/futures/detail/Types.h>
#include <folly/lang/Assume.h>
#include <folly/lang/Exception.h>
#include <folly/synchronization/MicroSpinLock.h>
......@@ -44,9 +45,10 @@ enum class State : uint8_t {
Start = 1 << 0,
OnlyResult = 1 << 1,
OnlyCallback = 1 << 2,
Proxy = 1 << 3,
Done = 1 << 4,
Empty = 1 << 5,
OnlyCallbackAllowInline = 1 << 3,
Proxy = 1 << 4,
Done = 1 << 5,
Empty = 1 << 6,
};
constexpr State operator&(State a, State b) {
return State(uint8_t(a) & uint8_t(b));
......@@ -124,6 +126,7 @@ static_assert(sizeof(SpinLock) == 1, "missized");
/// | \ (setCallback()) (setResult()) |
/// | \ \ / |
/// | \ ---> OnlyCallback --- |
/// | \ or OnlyCallbackAllowInline |
/// | \ \ |
/// | (setProxy()) (setProxy()) |
/// | \ \ |
......@@ -151,6 +154,13 @@ static_assert(sizeof(SpinLock) == 1, "missized");
/// point forward only the producer thread can safely access that callback
/// (see `setResult()` and `doCallback()` where the producer thread can both
/// read and modify the callback).
/// - OnlyCallbackAllowInline: as for OnlyCallback but the core is allowed to
/// run the callback inline with the setResult call, and therefore in the
/// execution context and on the executor that executed the callback on the
/// previous core, rather than adding the callback to the current Core's
/// executor. This will only happen if the executor on which the previous
/// callback is executing, and on which it is calling setResult, is the same
/// as the executor the current core would add the callback to.
/// - Proxy: producer thread has set a proxy core which the callback should be
/// proxied to.
/// - Done: callback can be safely accessed only within `doCallback()`, which
......@@ -229,7 +239,8 @@ class Core final {
/// May call from any thread
bool hasCallback() const noexcept {
constexpr auto allowed = State::OnlyCallback | State::Done;
constexpr auto allowed =
State::OnlyCallback | State::OnlyCallbackAllowInline | State::Done;
auto const state = state_.load(std::memory_order_acquire);
return State() != (state & allowed);
}
......@@ -267,7 +278,8 @@ class Core final {
///
/// State dependent preconditions:
///
/// - Start or OnlyCallback: Never safe - do not call. (Access in those states
/// - Start, OnlyCallback or OnlyCallbackAllowInline: Never safe - do not
/// call. (Access in those states
/// would be undefined behavior since the producer thread can, in those
/// states, asynchronously set the referenced Try object.)
/// - OnlyResult: Always safe. (Though the consumer thread should not use the
......@@ -303,19 +315,24 @@ class Core final {
/// and might also synchronously execute that callback (e.g., if there is no
/// executor or if the executor is inline).
template <typename F>
void setCallback(F&& func, std::shared_ptr<folly::RequestContext>&& context) {
void setCallback(
F&& func,
std::shared_ptr<folly::RequestContext>&& context,
futures::detail::InlineContinuation allowInline) {
DCHECK(!hasCallback());
// construct callback_ first; if that fails, context_ will not leak
::new (&callback_) Callback(std::forward<F>(func));
::new (&context_) Context(std::move(context));
auto state = state_.load(std::memory_order_acquire);
State nextState = allowInline == futures::detail::InlineContinuation::permit
? State::OnlyCallbackAllowInline
: State::OnlyCallback;
if (state == State::Start) {
if (state_.compare_exchange_strong(
state,
State::OnlyCallback,
nextState,
std::memory_order_release,
std::memory_order_acquire)) {
return;
......@@ -325,12 +342,12 @@ class Core final {
if (state == State::OnlyResult) {
state_.store(State::Done, std::memory_order_relaxed);
doCallback();
doCallback(Executor::KeepAlive<>{}, state);
return;
}
if (state == State::Proxy) {
return proxyCallback();
return proxyCallback(state);
}
terminate_with<std::logic_error>("setCallback unexpected state");
......@@ -357,11 +374,14 @@ class Core final {
std::memory_order_acquire)) {
break;
}
assume(state == State::OnlyCallback);
assume(
state == State::OnlyCallback ||
state == State::OnlyCallbackAllowInline);
FOLLY_FALLTHROUGH;
case State::OnlyCallback:
proxyCallback();
case State::OnlyCallbackAllowInline:
proxyCallback(state);
break;
default:
......@@ -380,6 +400,19 @@ class Core final {
/// and might also synchronously execute that callback (e.g., if there is no
/// executor or if the executor is inline).
void setResult(Try<T>&& t) {
setResult(Executor::KeepAlive<>{}, std::move(t));
}
/// Call only from producer thread.
/// Call only once - else undefined behavior.
///
/// See FSM graph for allowed transitions.
///
/// If it transitions to Done, synchronously initiates a call to the callback,
/// and might also synchronously execute that callback (e.g., if there is no
/// executor, if the executor is inline or if completingKA represents the
/// same executor as does executor_).
void setResult(Executor::KeepAlive<>&& completingKA, Try<T>&& t) {
DCHECK(!hasResult());
::new (&result_) Result(std::move(t));
......@@ -394,12 +427,15 @@ class Core final {
std::memory_order_acquire)) {
return;
}
assume(state == State::OnlyCallback);
assume(
state == State::OnlyCallback ||
state == State::OnlyCallbackAllowInline);
FOLLY_FALLTHROUGH;
case State::OnlyCallback:
case State::OnlyCallbackAllowInline:
state_.store(State::Done, std::memory_order_relaxed);
doCallback();
doCallback(std::move(completingKA), state);
return;
default:
......@@ -426,7 +462,9 @@ class Core final {
/// after the callback has already been invoked, but not concurrently with
/// anything which might trigger invocation of the callback.
void setExecutor(Executor::KeepAlive<> x) {
DCHECK(state_ != State::OnlyCallback);
DCHECK(
state_ != State::OnlyCallback &&
state_ != State::OnlyCallbackAllowInline);
executor_ = std::move(x);
}
......@@ -560,12 +598,16 @@ class Core final {
};
// May be called at most once.
void doCallback() {
void doCallback(Executor::KeepAlive<>&& completingKA, State priorState) {
DCHECK(state_ == State::Done);
if (executor_) {
bool allowInline =
(executor_.get() == completingKA.get() &&
priorState == State::OnlyCallbackAllowInline);
// If we have no executor or if we are allowing inline executor on a
// matching executor, run inline.
if (executor_ && !allowInline) {
auto x = std::exchange(executor_, Executor::KeepAlive<>());
exception_wrapper ew;
// We need to reset `callback_` after it was executed (which can happen
// through the executor or, if `Executor::add` throws, below). The
......@@ -607,14 +649,21 @@ class Core final {
detachOne();
};
RequestContextScopeGuard rctx(std::move(context_));
callback_(Executor::KeepAlive<>{}, std::move(result_));
callback_(executor_.copy(), std::move(result_));
return;
}
}
void proxyCallback() {
void proxyCallback(State priorState) {
// If the state of the core being proxied had a callback that allows inline
// execution, maintain this information in the proxy
futures::detail::InlineContinuation allowInline =
(priorState == State::OnlyCallbackAllowInline
? futures::detail::InlineContinuation::permit
: futures::detail::InlineContinuation::forbid);
state_.store(State::Empty, std::memory_order_relaxed);
proxy_->setExecutor(std::move(executor_));
proxy_->setCallback(std::move(callback_), std::move(context_));
proxy_->setCallback(std::move(callback_), std::move(context_), allowInline);
proxy_->detachFuture();
context_.~Context();
callback_.~Callback();
......
......@@ -35,4 +35,9 @@ namespace folly {
/// futures::sleep(std::chrono::seconds(1));
using Duration = std::chrono::milliseconds;
namespace futures {
namespace detail {
enum class InlineContinuation { permit, forbid };
} // namespace detail
} // namespace futures
} // namespace folly
......@@ -241,6 +241,71 @@ TEST(Via, then2) {
EXPECT_TRUE(c);
}
TEST(Via, allowInline) {
ManualExecutor x1, x2;
bool a = false, b = false, c = false, d = false, e = false, f = false,
g = false, h = false, i = false, j = false;
via(&x1)
.thenValue([&](auto&&) { a = true; })
.thenTryInline([&](auto&&) { b = true; })
.thenTry([&](auto&&) { c = true; })
.via(&x2)
.thenTryInline([&](auto&&) { d = true; })
.thenValue([&](auto&&) {
e = true;
return via(&x2).thenValue([&](auto&&) { f = true; });
})
.thenValueInline([&](auto&&) { g = true; })
.thenValue([&](auto&&) {
h = true;
return via(&x1).thenValue([&](auto&&) { i = true; });
})
.thenValueInline([&](auto&&) { j = true; });
EXPECT_FALSE(a);
EXPECT_FALSE(b);
// Expect b to be satisfied inline with the task x1
x1.run();
EXPECT_TRUE(a);
EXPECT_TRUE(b);
EXPECT_FALSE(c);
x1.run();
EXPECT_TRUE(c);
EXPECT_FALSE(d);
// Demonstrate that the executor transition did not allow inline execution
x2.run();
EXPECT_TRUE(d);
EXPECT_FALSE(e);
x2.run();
EXPECT_TRUE(e);
EXPECT_FALSE(f);
EXPECT_FALSE(g);
// Completing nested continuation should satisfy inline continuation
x2.run();
EXPECT_TRUE(f);
EXPECT_TRUE(g);
EXPECT_FALSE(h);
x2.run();
EXPECT_TRUE(h);
EXPECT_FALSE(i);
EXPECT_FALSE(j);
// Nested continuation on different executor should not complete next entry
// inline
x1.run();
EXPECT_TRUE(i);
EXPECT_FALSE(j);
x2.run();
EXPECT_TRUE(j);
}
#ifndef __APPLE__ // TODO #7372389
/// Simple executor that does work in another thread
class ThreadExecutor : public Executor {
......
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