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

Simplify EventBaseLoopController

Summary: Make EventBaseLoopController not dependent on EventBase's runOnDestruction. This will allow us to make FiberManager work with any Executor.

Reviewed By: yfeldblum

Differential Revision: D6828478

fbshipit-source-id: 4d74776044a02954d022ef6f6f1e6834839973b8
parent 425c8833
...@@ -96,6 +96,21 @@ class AtomicIntrusiveLinkedList { ...@@ -96,6 +96,21 @@ class AtomicIntrusiveLinkedList {
return oldHead == nullptr; return oldHead == nullptr;
} }
/**
* Replaces the head with nullptr,
* and calls func() on the removed elements in the order from tail to head.
* Returns false if the list was empty.
*/
template <typename F>
bool sweepOnce(F&& func) {
if (auto head = head_.exchange(nullptr)) {
auto rhead = reverse(head);
unlinkAll(rhead, std::forward<F>(func));
return true;
}
return false;
}
/** /**
* Repeatedly replaces the head with nullptr, * Repeatedly replaces the head with nullptr,
* and calls func() on the removed elements in the order from tail to head. * and calls func() on the removed elements in the order from tail to head.
...@@ -103,9 +118,7 @@ class AtomicIntrusiveLinkedList { ...@@ -103,9 +118,7 @@ class AtomicIntrusiveLinkedList {
*/ */
template <typename F> template <typename F>
void sweep(F&& func) { void sweep(F&& func) {
while (auto head = head_.exchange(nullptr)) { while (sweepOnce(func)) {
auto rhead = reverse(head);
unlinkAll(rhead, std::forward<F>(func));
} }
} }
......
...@@ -19,8 +19,7 @@ ...@@ -19,8 +19,7 @@
namespace folly { namespace folly {
namespace fibers { namespace fibers {
inline EventBaseLoopController::EventBaseLoopController() inline EventBaseLoopController::EventBaseLoopController() : callback_(*this) {}
: callback_(*this), aliveWeak_(destructionCallback_.getWeak()) {}
inline EventBaseLoopController::~EventBaseLoopController() { inline EventBaseLoopController::~EventBaseLoopController() {
callback_.cancelLoopCallback(); callback_.cancelLoopCallback();
...@@ -38,8 +37,6 @@ inline void EventBaseLoopController::attachEventBase( ...@@ -38,8 +37,6 @@ inline void EventBaseLoopController::attachEventBase(
} }
eventBase_ = &eventBase; eventBase_ = &eventBase;
eventBase_->runOnDestruction(&destructionCallback_);
eventBaseAttached_ = true; eventBaseAttached_ = true;
if (awaitingScheduling_) { if (awaitingScheduling_) {
...@@ -66,10 +63,6 @@ inline void EventBaseLoopController::schedule() { ...@@ -66,10 +63,6 @@ inline void EventBaseLoopController::schedule() {
} }
} }
inline void EventBaseLoopController::cancel() {
callback_.cancelLoopCallback();
}
inline void EventBaseLoopController::runLoop() { inline void EventBaseLoopController::runLoop() {
if (!eventBaseKeepAlive_) { if (!eventBaseKeepAlive_) {
// runLoop can be called twice if both schedule() and scheduleThreadSafe() // runLoop can be called twice if both schedule() and scheduleThreadSafe()
...@@ -80,7 +73,9 @@ inline void EventBaseLoopController::runLoop() { ...@@ -80,7 +73,9 @@ inline void EventBaseLoopController::runLoop() {
eventBaseKeepAlive_ = eventBase_->getKeepAliveToken(); eventBaseKeepAlive_ = eventBase_->getKeepAliveToken();
} }
if (loopRunner_) { if (loopRunner_) {
loopRunner_->run([&] { fm_->loopUntilNoReadyImpl(); }); if (fm_->hasReadyTasks()) {
loopRunner_->run([&] { fm_->loopUntilNoReadyImpl(); });
}
} else { } else {
fm_->loopUntilNoReadyImpl(); fm_->loopUntilNoReadyImpl();
} }
...@@ -98,13 +93,14 @@ inline void EventBaseLoopController::scheduleThreadSafe( ...@@ -98,13 +93,14 @@ inline void EventBaseLoopController::scheduleThreadSafe(
3) We fulfill the promise from the other thread. */ 3) We fulfill the promise from the other thread. */
assert(eventBaseAttached_); assert(eventBaseAttached_);
auto alive = aliveWeak_.lock(); if (func()) {
eventBase_->runInEventBaseThread([this]() {
if (fm_->shouldRunLoopRemote()) {
return runLoop();
}
if (func() && alive) { if (!fm_->hasTasks()) {
auto aliveWeak = aliveWeak_; eventBaseKeepAlive_.reset();
eventBase_->runInEventBaseThread([this, aliveWeak]() {
if (!aliveWeak.expired()) {
runLoop();
} }
}); });
} }
......
...@@ -57,50 +57,18 @@ class EventBaseLoopController : public LoopController { ...@@ -57,50 +57,18 @@ class EventBaseLoopController : public LoopController {
EventBaseLoopController& controller_; EventBaseLoopController& controller_;
}; };
class DestructionCallback : public folly::EventBase::LoopCallback {
public:
DestructionCallback() : alive_(new int(42)) {}
~DestructionCallback() override {
reset();
}
void runLoopCallback() noexcept override {
reset();
}
std::weak_ptr<void> getWeak() {
return {alive_};
}
private:
void reset() {
std::weak_ptr<void> aliveWeak(alive_);
alive_.reset();
while (!aliveWeak.expired()) {
// Spin until all operations requiring EventBaseLoopController to be
// alive are complete.
}
}
std::shared_ptr<void> alive_;
};
bool awaitingScheduling_{false}; bool awaitingScheduling_{false};
VirtualEventBase* eventBase_{nullptr}; VirtualEventBase* eventBase_{nullptr};
Executor::KeepAlive eventBaseKeepAlive_; Executor::KeepAlive eventBaseKeepAlive_;
ControllerCallback callback_; ControllerCallback callback_;
DestructionCallback destructionCallback_;
FiberManager* fm_{nullptr}; FiberManager* fm_{nullptr};
std::atomic<bool> eventBaseAttached_{false}; std::atomic<bool> eventBaseAttached_{false};
std::weak_ptr<void> aliveWeak_;
InlineFunctionRunner* loopRunner_{nullptr}; InlineFunctionRunner* loopRunner_{nullptr};
/* LoopController interface */ /* LoopController interface */
void setFiberManager(FiberManager* fm) override; void setFiberManager(FiberManager* fm) override;
void schedule() override; void schedule() override;
void cancel() override;
void runLoop() override; void runLoop() override;
void scheduleThreadSafe(std::function<bool()> func) override; void scheduleThreadSafe(std::function<bool()> func) override;
void timedSchedule(std::function<void()> func, TimePoint time) override; void timedSchedule(std::function<void()> func, TimePoint time) override;
......
...@@ -79,9 +79,7 @@ FiberManager::FiberManager( ...@@ -79,9 +79,7 @@ FiberManager::FiberManager(
std::move(options)) {} std::move(options)) {}
FiberManager::~FiberManager() { FiberManager::~FiberManager() {
if (isLoopScheduled_) { loopController_.reset();
loopController_->cancel();
}
while (!fibersPool_.empty()) { while (!fibersPool_.empty()) {
fibersPool_.pop_front_and_dispose([](Fiber* fiber) { delete fiber; }); fibersPool_.pop_front_and_dispose([](Fiber* fiber) { delete fiber; });
...@@ -100,7 +98,7 @@ const LoopController& FiberManager::loopController() const { ...@@ -100,7 +98,7 @@ const LoopController& FiberManager::loopController() const {
bool FiberManager::hasTasks() const { bool FiberManager::hasTasks() const {
return fibersActive_ > 0 || !remoteReadyQueue_.empty() || return fibersActive_ > 0 || !remoteReadyQueue_.empty() ||
!remoteTaskQueue_.empty(); !remoteTaskQueue_.empty() || remoteCount_ > 0;
} }
Fiber* FiberManager::getFiber() { Fiber* FiberManager::getFiber() {
......
...@@ -209,36 +209,42 @@ inline void FiberManager::loopUntilNoReadyImpl() { ...@@ -209,36 +209,42 @@ inline void FiberManager::loopUntilNoReadyImpl() {
CHECK_EQ(this, originalFiberManager); CHECK_EQ(this, originalFiberManager);
}; };
bool hadRemoteFiber = true; bool hadRemote = true;
while (hadRemoteFiber) { while (hadRemote) {
hadRemoteFiber = false;
while (!readyFibers_.empty()) { while (!readyFibers_.empty()) {
auto& fiber = readyFibers_.front(); auto& fiber = readyFibers_.front();
readyFibers_.pop_front(); readyFibers_.pop_front();
runReadyFiber(&fiber); runReadyFiber(&fiber);
} }
remoteReadyQueue_.sweep([this, &hadRemoteFiber](Fiber* fiber) { auto hadRemoteFiber = remoteReadyQueue_.sweepOnce(
runReadyFiber(fiber); [this](Fiber* fiber) { runReadyFiber(fiber); });
hadRemoteFiber = true;
});
remoteTaskQueue_.sweep([this, &hadRemoteFiber](RemoteTask* taskPtr) { if (hadRemoteFiber) {
std::unique_ptr<RemoteTask> task(taskPtr); ++remoteCount_;
auto fiber = getFiber(); }
if (task->localData) {
fiber->localData_ = *task->localData;
}
fiber->rcontext_ = std::move(task->rcontext);
fiber->setFunction(std::move(task->func)); auto hadRemoteTask =
if (observer_) { remoteTaskQueue_.sweepOnce([this](RemoteTask* taskPtr) {
observer_->runnable(reinterpret_cast<uintptr_t>(fiber)); std::unique_ptr<RemoteTask> task(taskPtr);
} auto fiber = getFiber();
runReadyFiber(fiber); if (task->localData) {
hadRemoteFiber = true; 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;
} }
if (observer_) { if (observer_) {
...@@ -249,6 +255,16 @@ inline void FiberManager::loopUntilNoReadyImpl() { ...@@ -249,6 +255,16 @@ inline void FiberManager::loopUntilNoReadyImpl() {
readyFibers_.splice(readyFibers_.end(), yieldedFibers_); readyFibers_.splice(readyFibers_.end(), yieldedFibers_);
} }
inline bool FiberManager::shouldRunLoopRemote() {
--remoteCount_;
return !remoteReadyQueue_.empty() || !remoteTaskQueue_.empty();
}
inline bool FiberManager::hasReadyTasks() const {
return !readyFibers_.empty() || !remoteReadyQueue_.empty() ||
!remoteTaskQueue_.empty();
}
// We need this to be in a struct, not inlined in addTask, because clang crashes // We need this to be in a struct, not inlined in addTask, because clang crashes
// otherwise. // otherwise.
template <typename F> template <typename F>
......
...@@ -164,11 +164,21 @@ class FiberManager : public ::folly::Executor { ...@@ -164,11 +164,21 @@ class FiberManager : public ::folly::Executor {
*/ */
void loopUntilNoReadyImpl(); void loopUntilNoReadyImpl();
/**
* This should only be called by a LoopController.
*/
bool shouldRunLoopRemote();
/** /**
* @return true if there are outstanding tasks. * @return true if there are outstanding tasks.
*/ */
bool hasTasks() const; bool hasTasks() const;
/**
* @return true if there are tasks ready to run.
*/
bool hasReadyTasks() const;
/** /**
* Sets exception callback which will be called if any of the tasks throws an * Sets exception callback which will be called if any of the tasks throws an
* exception. * exception.
...@@ -442,6 +452,8 @@ class FiberManager : public ::folly::Executor { ...@@ -442,6 +452,8 @@ class FiberManager : public ::folly::Executor {
folly::AtomicIntrusiveLinkedList<RemoteTask, &RemoteTask::nextRemoteTask> folly::AtomicIntrusiveLinkedList<RemoteTask, &RemoteTask::nextRemoteTask>
remoteTaskQueue_; remoteTaskQueue_;
ssize_t remoteCount_{0};
std::shared_ptr<TimeoutController> timeoutManager_; std::shared_ptr<TimeoutController> timeoutManager_;
struct FibersPoolResizer { struct FibersPoolResizer {
......
...@@ -37,7 +37,7 @@ class LoopController { ...@@ -37,7 +37,7 @@ class LoopController {
/** /**
* Called by FiberManager to schedule the loop function run * Called by FiberManager to schedule the loop function run
* at some point in the future. * at some point in the futufre.
*/ */
virtual void schedule() = 0; virtual void schedule() = 0;
...@@ -53,12 +53,6 @@ class LoopController { ...@@ -53,12 +53,6 @@ class LoopController {
*/ */
virtual void scheduleThreadSafe(std::function<bool()> func) = 0; virtual void scheduleThreadSafe(std::function<bool()> func) = 0;
/**
* Called by FiberManager to cancel a previously scheduled
* loop function run.
*/
virtual void cancel() = 0;
/** /**
* Called by FiberManager to schedule some function to be run at some time. * Called by FiberManager to schedule some function to be run at some time.
*/ */
......
...@@ -28,6 +28,10 @@ class SimpleLoopController : public LoopController { ...@@ -28,6 +28,10 @@ class SimpleLoopController : public LoopController {
public: public:
SimpleLoopController() : fm_(nullptr), stopRequested_(false) {} SimpleLoopController() : fm_(nullptr), stopRequested_(false) {}
~SimpleLoopController() {
scheduled_ = false;
}
/** /**
* Run FiberManager loop; if no ready task are present, * Run FiberManager loop; if no ready task are present,
* run provided function. Stops after both stop() has been called * run provided function. Stops after both stop() has been called
...@@ -72,7 +76,17 @@ class SimpleLoopController : public LoopController { ...@@ -72,7 +76,17 @@ class SimpleLoopController : public LoopController {
} }
void runLoop() override { void runLoop() override {
fm_->loopUntilNoReadyImpl(); do {
if (remoteLoopRun_ < remoteScheduleCalled_) {
for (; remoteLoopRun_ < remoteScheduleCalled_; ++remoteLoopRun_) {
if (fm_->shouldRunLoopRemote()) {
fm_->loopUntilNoReadyImpl();
}
}
} else {
fm_->loopUntilNoReadyImpl();
}
} while (remoteLoopRun_ < remoteScheduleCalled_);
} }
void schedule() override { void schedule() override {
...@@ -88,6 +102,7 @@ class SimpleLoopController : public LoopController { ...@@ -88,6 +102,7 @@ class SimpleLoopController : public LoopController {
std::atomic<bool> scheduled_{false}; std::atomic<bool> scheduled_{false};
bool stopRequested_; bool stopRequested_;
std::atomic<int> remoteScheduleCalled_{0}; std::atomic<int> remoteScheduleCalled_{0};
int remoteLoopRun_{0};
std::vector<std::pair<TimePoint, std::function<void()>>> scheduledFuncs_; std::vector<std::pair<TimePoint, std::function<void()>>> scheduledFuncs_;
/* LoopController interface */ /* LoopController interface */
...@@ -96,10 +111,6 @@ class SimpleLoopController : public LoopController { ...@@ -96,10 +111,6 @@ class SimpleLoopController : public LoopController {
fm_ = fm; fm_ = fm;
} }
void cancel() override {
scheduled_ = false;
}
void scheduleThreadSafe(std::function<bool()> func) override { void scheduleThreadSafe(std::function<bool()> func) override {
if (func()) { if (func()) {
++remoteScheduleCalled_; ++remoteScheduleCalled_;
......
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