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

addTaskEager/addTaskEagerFuture

Summary: This allows avoiding extra executor/queueing operations if caller knows that task can be safely run inline.

Reviewed By: yfeldblum

Differential Revision: D13404703

fbshipit-source-id: 4aa71e59b48572671e8031ee7ea4dcd4e3a6138b
parent 59b94083
...@@ -37,6 +37,22 @@ auto FiberManager::addTaskFuture(F&& func) ...@@ -37,6 +37,22 @@ auto FiberManager::addTaskFuture(F&& func)
return f; return f;
} }
template <typename F>
auto FiberManager::addTaskEagerFuture(F&& func)
-> folly::Future<folly::lift_unit_t<invoke_result_t<F>>> {
using T = invoke_result_t<F>;
using FutureT = typename folly::lift_unit<T>::type;
folly::Promise<FutureT> p;
auto f = p.getFuture();
addTaskFinallyEager(
[func = std::forward<F>(func)]() mutable { return func(); },
[p = std::move(p)](folly::Try<T>&& t) mutable {
p.setTry(std::move(t));
});
return f;
}
template <typename F> template <typename F>
auto FiberManager::addTaskRemoteFuture(F&& func) auto FiberManager::addTaskRemoteFuture(F&& func)
-> folly::Future<folly::lift_unit_t<invoke_result_t<F>>> { -> folly::Future<folly::lift_unit_t<invoke_result_t<F>>> {
......
...@@ -179,7 +179,7 @@ inline void FiberManager::runReadyFiber(Fiber* fiber) { ...@@ -179,7 +179,7 @@ inline void FiberManager::runReadyFiber(Fiber* fiber) {
currentFiber_ = nullptr; currentFiber_ = nullptr;
fiber->rcontext_ = RequestContext::saveContext(); fiber->rcontext_ = RequestContext::saveContext();
fiber->state_ = Fiber::READY_TO_RUN; fiber->state_ = Fiber::READY_TO_RUN;
yieldedFibers_.push_back(*fiber); yieldedFibers_->push_back(*fiber);
} }
} }
...@@ -187,7 +187,8 @@ inline void FiberManager::loopUntilNoReady() { ...@@ -187,7 +187,8 @@ inline void FiberManager::loopUntilNoReady() {
return loopController_->runLoop(); return loopController_->runLoop();
} }
inline void FiberManager::loopUntilNoReadyImpl() { template <typename LoopFunc>
void FiberManager::runFibersHelper(LoopFunc&& loopFunc) {
#ifndef _WIN32 #ifndef _WIN32
if (UNLIKELY(!alternateSignalStackRegistered_)) { if (UNLIKELY(!alternateSignalStackRegistered_)) {
registerAlternateSignalStack(); registerAlternateSignalStack();
...@@ -195,17 +196,25 @@ inline void FiberManager::loopUntilNoReadyImpl() { ...@@ -195,17 +196,25 @@ inline void FiberManager::loopUntilNoReadyImpl() {
#endif #endif
// Support nested FiberManagers // Support nested FiberManagers
auto originalFiberManager = this; auto originalFiberManager = std::exchange(currentFiberManager_, this);
std::swap(currentFiberManager_, originalFiberManager);
// Save current context, and reset it after executing all fibers. // Save current context, and reset it after executing all fibers.
// This can avoid a lot of context swapping, // This can avoid a lot of context swapping,
// if the Fibers share the same context // if the Fibers share the same context
auto curCtx = RequestContext::saveContext(); auto curCtx = RequestContext::saveContext();
FiberTailQueue yieldedFibers;
auto prevYieldedFibers = std::exchange(yieldedFibers_, &yieldedFibers);
SCOPE_EXIT { SCOPE_EXIT {
yieldedFibers_ = prevYieldedFibers;
if (observer_) {
for (auto& yielded : yieldedFibers) {
observer_->runnable(reinterpret_cast<uintptr_t>(&yielded));
}
}
readyFibers_.splice(readyFibers_.end(), yieldedFibers);
RequestContext::setContext(std::move(curCtx)); RequestContext::setContext(std::move(curCtx));
isLoopScheduled_ = false;
if (!readyFibers_.empty()) { if (!readyFibers_.empty()) {
ensureLoopScheduled(); ensureLoopScheduled();
} }
...@@ -213,50 +222,63 @@ inline void FiberManager::loopUntilNoReadyImpl() { ...@@ -213,50 +222,63 @@ inline void FiberManager::loopUntilNoReadyImpl() {
CHECK_EQ(this, originalFiberManager); CHECK_EQ(this, originalFiberManager);
}; };
bool hadRemote = true; loopFunc();
while (hadRemote) { }
while (!readyFibers_.empty()) {
auto& fiber = readyFibers_.front();
readyFibers_.pop_front();
runReadyFiber(&fiber);
}
auto hadRemoteFiber = remoteReadyQueue_.sweepOnce( inline void FiberManager::loopUntilNoReadyImpl() {
[this](Fiber* fiber) { runReadyFiber(fiber); }); runFibersHelper([&] {
SCOPE_EXIT {
isLoopScheduled_ = false;
};
bool hadRemote = true;
while (hadRemote) {
while (!readyFibers_.empty()) {
auto& fiber = readyFibers_.front();
readyFibers_.pop_front();
runReadyFiber(&fiber);
}
if (hadRemoteFiber) { auto hadRemoteFiber = remoteReadyQueue_.sweepOnce(
++remoteCount_; [this](Fiber* fiber) { runReadyFiber(fiber); });
}
auto hadRemoteTask = if (hadRemoteFiber) {
remoteTaskQueue_.sweepOnce([this](RemoteTask* taskPtr) { ++remoteCount_;
std::unique_ptr<RemoteTask> task(taskPtr); }
auto fiber = getFiber();
if (task->localData) {
fiber->localData_ = *task->localData;
}
fiber->rcontext_ = std::move(task->rcontext);
fiber->setFunction(std::move(task->func));
if (observer_) {
observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
}
runReadyFiber(fiber);
});
if (hadRemoteTask) {
++remoteCount_;
}
hadRemote = hadRemoteTask || hadRemoteFiber; auto hadRemoteTask =
} remoteTaskQueue_.sweepOnce([this](RemoteTask* taskPtr) {
std::unique_ptr<RemoteTask> task(taskPtr);
auto fiber = getFiber();
if (task->localData) {
fiber->localData_ = *task->localData;
}
fiber->rcontext_ = std::move(task->rcontext);
fiber->setFunction(std::move(task->func));
if (observer_) {
observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
}
runReadyFiber(fiber);
});
if (hadRemoteTask) {
++remoteCount_;
}
if (observer_) { hadRemote = hadRemoteTask || hadRemoteFiber;
for (auto& yielded : yieldedFibers_) {
observer_->runnable(reinterpret_cast<uintptr_t>(&yielded));
} }
} });
readyFibers_.splice(readyFibers_.end(), yieldedFibers_); }
inline void FiberManager::runEagerFiber(Fiber* fiber) {
runInMainContext([&] {
auto prevCurrentFiber = std::exchange(currentFiber_, fiber);
SCOPE_EXIT {
currentFiber_ = prevCurrentFiber;
};
runFibersHelper([&] { runReadyFiber(fiber); });
});
} }
inline bool FiberManager::shouldRunLoopRemote() { inline bool FiberManager::shouldRunLoopRemote() {
...@@ -303,7 +325,7 @@ struct FiberManager::AddTaskHelper { ...@@ -303,7 +325,7 @@ struct FiberManager::AddTaskHelper {
}; };
template <typename F> template <typename F>
void FiberManager::addTask(F&& func) { Fiber* FiberManager::createTask(F&& func) {
typedef AddTaskHelper<F> Helper; typedef AddTaskHelper<F> Helper;
auto fiber = getFiber(); auto fiber = getFiber();
...@@ -320,14 +342,24 @@ void FiberManager::addTask(F&& func) { ...@@ -320,14 +342,24 @@ void FiberManager::addTask(F&& func) {
fiber->setFunction(std::ref(*funcLoc)); fiber->setFunction(std::ref(*funcLoc));
} }
readyFibers_.push_back(*fiber);
if (observer_) { if (observer_) {
observer_->runnable(reinterpret_cast<uintptr_t>(fiber)); observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
} }
return fiber;
}
template <typename F>
void FiberManager::addTask(F&& func) {
readyFibers_.push_back(*createTask(std::forward<F>(func)));
ensureLoopScheduled(); ensureLoopScheduled();
} }
template <typename F>
void FiberManager::addTaskEager(F&& func) {
runEagerFiber(createTask(std::forward<F>(func)));
}
template <typename F> template <typename F>
void FiberManager::addTaskRemote(F&& func) { void FiberManager::addTaskRemote(F&& func) {
auto task = [&]() { auto task = [&]() {
...@@ -414,7 +446,7 @@ struct FiberManager::AddTaskFinallyHelper { ...@@ -414,7 +446,7 @@ struct FiberManager::AddTaskFinallyHelper {
}; };
template <typename F, typename G> template <typename F, typename G>
void FiberManager::addTaskFinally(F&& func, G&& finally) { Fiber* FiberManager::createTaskFinally(F&& func, G&& finally) {
typedef invoke_result_t<F> Result; typedef invoke_result_t<F> Result;
static_assert( static_assert(
...@@ -453,14 +485,26 @@ void FiberManager::addTaskFinally(F&& func, G&& finally) { ...@@ -453,14 +485,26 @@ void FiberManager::addTaskFinally(F&& func, G&& finally) {
fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc)); fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc));
} }
readyFibers_.push_back(*fiber);
if (observer_) { if (observer_) {
observer_->runnable(reinterpret_cast<uintptr_t>(fiber)); observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
} }
return fiber;
}
template <typename F, typename G>
void FiberManager::addTaskFinally(F&& func, G&& finally) {
readyFibers_.push_back(
*createTaskFinally(std::forward<F>(func), std::forward<G>(finally)));
ensureLoopScheduled(); ensureLoopScheduled();
} }
template <typename F, typename G>
void FiberManager::addTaskFinallyEager(F&& func, G&& finally) {
runEagerFiber(
createTaskFinally(std::forward<F>(func), std::forward<G>(finally)));
}
template <typename F> template <typename F>
invoke_result_t<F> FiberManager::runInMainContext(F&& func) { invoke_result_t<F> FiberManager::runInMainContext(F&& func) {
if (UNLIKELY(activeFiber_ == nullptr)) { if (UNLIKELY(activeFiber_ == nullptr)) {
......
...@@ -215,6 +215,31 @@ class FiberManager : public ::folly::Executor { ...@@ -215,6 +215,31 @@ class FiberManager : public ::folly::Executor {
template <typename F> template <typename F>
auto addTaskFuture(F&& func) auto addTaskFuture(F&& func)
-> folly::Future<folly::lift_unit_t<invoke_result_t<F>>>; -> folly::Future<folly::lift_unit_t<invoke_result_t<F>>>;
/**
* Add a new task to be executed. Must be called from FiberManager's thread.
* The new task is run eagerly. addTaskEager will return only once the new
* task reaches its first suspension point or is completed.
*
* @param func Task functor; must have a signature of `void func()`.
* The object will be destroyed once task execution is complete.
*/
template <typename F>
void addTaskEager(F&& func);
/**
* Add a new task to be executed and return a future that will be set on
* return from func. Must be called from FiberManager's thread.
* The new task is run eagerly. addTaskEager will return only once the new
* task reaches its first suspension point or is completed.
*
* @param func Task functor; must have a signature of `void func()`.
* The object will be destroyed once task execution is complete.
*/
template <typename F>
auto addTaskEagerFuture(F&& func)
-> folly::Future<folly::lift_unit_t<invoke_result_t<F>>>;
/** /**
* Add a new task to be executed. Safe to call from other threads. * Add a new task to be executed. Safe to call from other threads.
* *
...@@ -252,6 +277,20 @@ class FiberManager : public ::folly::Executor { ...@@ -252,6 +277,20 @@ class FiberManager : public ::folly::Executor {
template <typename F, typename G> template <typename F, typename G>
void addTaskFinally(F&& func, G&& finally); void addTaskFinally(F&& func, G&& finally);
/**
* Add a new task. When the task is complete, execute finally(Try<Result>&&)
* on the main context.
* The new task is run eagerly. addTaskEager will return only once the new
* task reaches its first suspension point or is completed.
*
* @param func Task functor; must have a signature of `T func()` for some T.
* @param finally Finally functor; must have a signature of
* `void finally(Try<T>&&)` and will be passed
* the result of func() (including the exception if occurred).
*/
template <typename F, typename G>
void addTaskFinallyEager(F&& func, G&& finally);
/** /**
* If called from a fiber, immediately switches to the FiberManager's context * If called from a fiber, immediately switches to the FiberManager's context
* and runs func(), going back to the Fiber's context after completion. * and runs func(), going back to the Fiber's context after completion.
...@@ -335,7 +374,7 @@ class FiberManager : public ::folly::Executor { ...@@ -335,7 +374,7 @@ class FiberManager : public ::folly::Executor {
* not include fibers or tasks scheduled remotely). * not include fibers or tasks scheduled remotely).
*/ */
size_t runQueueSize() const { size_t runQueueSize() const {
return readyFibers_.size() + yieldedFibers_.size(); return readyFibers_.size() + (yieldedFibers_ ? yieldedFibers_->size() : 0);
} }
static FiberManager& getFiberManager(); static FiberManager& getFiberManager();
...@@ -364,9 +403,20 @@ class FiberManager : public ::folly::Executor { ...@@ -364,9 +403,20 @@ class FiberManager : public ::folly::Executor {
AtomicIntrusiveLinkedListHook<RemoteTask> nextRemoteTask; AtomicIntrusiveLinkedListHook<RemoteTask> nextRemoteTask;
}; };
template <typename F>
Fiber* createTask(F&& func);
template <typename F, typename G>
Fiber* createTaskFinally(F&& func, G&& finally);
void runEagerFiber(Fiber* fiber);
void activateFiber(Fiber* fiber); void activateFiber(Fiber* fiber);
void deactivateFiber(Fiber* fiber); void deactivateFiber(Fiber* fiber);
template <typename LoopFunc>
void runFibersHelper(LoopFunc&& loopFunc);
typedef folly::IntrusiveList<Fiber, &Fiber::listHook_> FiberTailQueue; typedef folly::IntrusiveList<Fiber, &Fiber::listHook_> FiberTailQueue;
typedef folly::IntrusiveList<Fiber, &Fiber::globalListHook_> typedef folly::IntrusiveList<Fiber, &Fiber::globalListHook_>
GlobalFiberTailQueue; GlobalFiberTailQueue;
...@@ -379,8 +429,8 @@ class FiberManager : public ::folly::Executor { ...@@ -379,8 +429,8 @@ class FiberManager : public ::folly::Executor {
Fiber* currentFiber_{nullptr}; Fiber* currentFiber_{nullptr};
FiberTailQueue readyFibers_; /**< queue of fibers ready to be executed */ FiberTailQueue readyFibers_; /**< queue of fibers ready to be executed */
FiberTailQueue yieldedFibers_; /**< queue of fibers which have yielded FiberTailQueue* yieldedFibers_{nullptr}; /**< queue of fibers which have
execution */ yielded execution */
FiberTailQueue fibersPool_; /**< pool of uninitialized Fiber objects */ FiberTailQueue fibersPool_; /**< pool of uninitialized Fiber objects */
GlobalFiberTailQueue allFibers_; /**< list of all Fiber objects owned */ GlobalFiberTailQueue allFibers_; /**< list of all Fiber objects owned */
...@@ -530,6 +580,11 @@ inline void addTask(F&& func) { ...@@ -530,6 +580,11 @@ inline void addTask(F&& func) {
return FiberManager::getFiberManager().addTask(std::forward<F>(func)); return FiberManager::getFiberManager().addTask(std::forward<F>(func));
} }
template <typename F>
inline void addTaskEager(F&& func) {
return FiberManager::getFiberManager().addTaskEager(std::forward<F>(func));
}
/** /**
* Add a new task. When the task is complete, execute finally(Try<Result>&&) * Add a new task. When the task is complete, execute finally(Try<Result>&&)
* on the main context. * on the main context.
...@@ -547,6 +602,12 @@ inline void addTaskFinally(F&& func, G&& finally) { ...@@ -547,6 +602,12 @@ inline void addTaskFinally(F&& func, G&& finally) {
std::forward<F>(func), std::forward<G>(finally)); std::forward<F>(func), std::forward<G>(finally));
} }
template <typename F, typename G>
inline void addTaskFinallyEager(F&& func, G&& finally) {
return FiberManager::getFiberManager().addTaskFinallyEager(
std::forward<F>(func), std::forward<G>(finally));
}
/** /**
* Blocks task execution until given promise is fulfilled. * Blocks task execution until given promise is fulfilled.
* *
......
...@@ -2259,3 +2259,87 @@ TEST(FiberManager, highWaterMarkViaRecordCurrentPosition) { ...@@ -2259,3 +2259,87 @@ TEST(FiberManager, highWaterMarkViaRecordCurrentPosition) {
}; };
std::thread(f).join(); std::thread(f).join();
} }
TEST(FiberManager, addTaskEager) {
folly::EventBase evb;
auto& fm = getFiberManager(evb);
bool eagerTaskStarted{false};
bool eagerTaskDone{false};
bool firstTaskDone{false};
fm.addTask([&] { firstTaskDone = true; });
fm.addTaskEager([&] {
EXPECT_FALSE(firstTaskDone);
eagerTaskStarted = true;
fm.yield();
EXPECT_TRUE(firstTaskDone);
eagerTaskDone = true;
});
EXPECT_TRUE(eagerTaskStarted);
evb.loop();
EXPECT_TRUE(eagerTaskDone);
EXPECT_TRUE(firstTaskDone);
}
TEST(FiberManager, addTaskEagerFuture) {
folly::EventBase evb;
auto& fm = getFiberManager(evb);
bool eagerTaskStarted{false};
bool eagerTaskDone{false};
EXPECT_TRUE(fm.addTaskEagerFuture([&] {}).isReady());
auto f = fm.addTaskEagerFuture([&] {
eagerTaskStarted = true;
fm.yield();
eagerTaskDone = true;
});
EXPECT_TRUE(eagerTaskStarted);
evb.loop();
EXPECT_TRUE(f.isReady());
EXPECT_TRUE(eagerTaskDone);
}
TEST(FiberManager, addTaskEagerNested) {
folly::EventBase evb;
auto& fm = getFiberManager(evb);
bool eagerTaskStarted{false};
bool eagerTaskDone{false};
bool firstTaskDone{false};
bool secondTaskDone{false};
fm.addTask([&] {
fm.addTaskEager([&] {
EXPECT_FALSE(firstTaskDone);
EXPECT_FALSE(secondTaskDone);
fm.runInMainContext([&] { eagerTaskStarted = true; });
fm.yield();
EXPECT_TRUE(firstTaskDone);
EXPECT_TRUE(secondTaskDone);
eagerTaskDone = true;
});
EXPECT_TRUE(eagerTaskStarted);
firstTaskDone = true;
});
fm.addTask([&] {
EXPECT_TRUE(firstTaskDone);
secondTaskDone = true;
});
evb.loop();
EXPECT_TRUE(eagerTaskDone);
EXPECT_TRUE(secondTaskDone);
}
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