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

Support inline defer 2/n - Add addFrom to DeferredExecutor

Summary: addFrom on DeferredExecutor allows the DeferredExecutor to know what executor is running the work enqueuing to it, which it can check against its internal state to run the work inline if the executors match.

Reviewed By: andriigrynenko

Differential Revision: D15836532

fbshipit-source-id: 40c35d976b90072c88bb544360b059c1b151d4d0
parent b619a10f
......@@ -567,18 +567,43 @@ FutureBase<T>::withinImplementation(Duration dur, E e, Timekeeper* tk) && {
class DeferredExecutor final : public Executor {
public:
void add(Func func) override {
addFrom(
Executor::KeepAlive<>{},
[func = std::move(func)](Executor::KeepAlive<>&& /*ka*/) mutable {
func();
});
}
// addFrom will:
// * run func inline if there is a stored executor and completingKA matches
// the stored executor
// * enqueue func into the stored executor if one exists
// * store func until an executor is set otherwise
void addFrom(
Executor::KeepAlive<>&& completingKA,
Executor::KeepAlive<>::KeepAliveFunc func) {
auto state = state_.load(std::memory_order_acquire);
if (state == State::DETACHED) {
return;
}
auto kaFunc = [func = std::move(func)](
Executor::KeepAlive<>&& /* ka */) mutable { func(); };
// If we are completing on the current executor, call inline, otherwise
// add
auto addWithInline =
[&](Executor::KeepAlive<>::KeepAliveFunc&& addFunc) mutable {
if (completingKA.get() == executor_.get()) {
addFunc(std::move(completingKA));
} else {
executor_.copy().add(std::move(addFunc));
}
};
if (state == State::HAS_EXECUTOR) {
executor_.copy().add(std::move(kaFunc));
addWithInline(std::move(func));
return;
}
DCHECK(state == State::EMPTY);
func_ = std::move(kaFunc);
func_ = std::move(func);
if (detail::compare_exchange_strong_release_acquire(
state_, state, State::HAS_FUNCTION)) {
return;
......@@ -588,7 +613,7 @@ class DeferredExecutor final : public Executor {
std::exchange(func_, nullptr);
return;
}
executor_.copy().add(std::exchange(func_, nullptr));
addWithInline(std::exchange(func_, nullptr));
}
Executor* getExecutor() const {
......
......@@ -1240,3 +1240,23 @@ TEST(SemiFuture, ensure) {
EXPECT_TRUE(ensureCalled);
}
}
TEST(SemiFuture, deferredExecutorInlineTest) {
bool a = false, b = false, c = false;
auto manualExec1 = ManualExecutor{};
auto manualExec1KA = getKeepAliveToken(manualExec1);
auto manualExec2 = ManualExecutor{};
auto manualExec2KA = getKeepAliveToken(manualExec2);
auto de = futures::detail::DeferredExecutor::create();
de->setExecutor(manualExec1KA);
de->add([&]() { a = true; });
EXPECT_FALSE(a);
manualExec1.run();
EXPECT_TRUE(a);
de->addFrom(manualExec2KA.copy(), [&](auto&&) { b = true; });
EXPECT_FALSE(b);
manualExec1.run();
EXPECT_TRUE(b);
de->addFrom(manualExec1KA.copy(), [&](auto&&) { c = true; });
EXPECT_TRUE(c);
}
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