Commit 2d2aed32 authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook Github Bot 6

Implement LoopKeepAlive for EventBase

Summary: LoopKeepAlive can be useful to tell EventBase, that loop() shouldn't return even if there are no events registered, because some tasks will later be added via runInEventBaseThread. While at least one LoopKeepAlive is alive - EventBase::loop() behaves like EventBase::loopForever().

Reviewed By: yfeldblum

Differential Revision: D3261706

fb-gh-sync-id: d91424d3d12cae11abd72cffdbd57f136f628dae
fbshipit-source-id: d91424d3d12cae11abd72cffdbd57f136f628dae
parent 93db3df4
...@@ -319,6 +319,7 @@ bool EventBase::loopBody(int flags) { ...@@ -319,6 +319,7 @@ bool EventBase::loopBody(int flags) {
} }
while (!stop_.load(std::memory_order_acquire)) { while (!stop_.load(std::memory_order_acquire)) {
applyLoopKeepAlive();
++nextLoopCnt_; ++nextLoopCnt_;
// Run the before loop callbacks // Run the before loop callbacks
...@@ -425,18 +426,34 @@ bool EventBase::loopBody(int flags) { ...@@ -425,18 +426,34 @@ bool EventBase::loopBody(int flags) {
return true; return true;
} }
void EventBase::loopForever() { void EventBase::applyLoopKeepAlive() {
// Update the notification queue event to treat it as a normal (non-internal) if (loopKeepAliveActive_ && loopKeepAlive_.unique()) {
// event. The notification queue event always remains installed, and the main // Restore the notification queue internal flag
// loop won't exit with it installed. fnRunner_->stopConsuming();
fnRunner_->stopConsuming(); fnRunner_->startConsumingInternal(this, queue_.get());
fnRunner_->startConsuming(this, queue_.get()); loopKeepAliveActive_ = false;
} else if (!loopKeepAliveActive_ && !loopKeepAlive_.unique()) {
bool ret = loop(); // Update the notification queue event to treat it as a normal
// (non-internal) event. The notification queue event always remains
// installed, and the main loop won't exit with it installed.
fnRunner_->stopConsuming();
fnRunner_->startConsuming(this, queue_.get());
loopKeepAliveActive_ = true;
}
}
// Restore the notification queue internal flag void EventBase::loopForever() {
fnRunner_->stopConsuming(); bool ret;
fnRunner_->startConsumingInternal(this, queue_.get()); {
SCOPE_EXIT {
applyLoopKeepAlive();
loopForeverActive_ = false;
};
loopForeverActive_ = true;
// Make sure notification queue events are treated as normal events.
auto loopKeepAlive = loopKeepAlive_;
ret = loop();
}
if (!ret) { if (!ret) {
folly::throwSystemError("error in EventBase::loopForever()"); folly::throwSystemError("error in EventBase::loopForever()");
......
...@@ -586,6 +586,22 @@ class EventBase : private boost::noncopyable, ...@@ -586,6 +586,22 @@ class EventBase : private boost::noncopyable,
loopOnce(); loopOnce();
} }
using LoopKeepAlive = std::shared_ptr<void>;
/// Returns you a handle which make loop() behave like loopForever() until
/// destroyed. loop() will return to its original behavior only when all
/// loop keep-alives are released. Loop holder is safe to release only from
/// EventBase thread.
///
/// May return no op LoopKeepAlive if loopForever() is already running.
LoopKeepAlive loopKeepAlive() {
if (loopForeverActive_) {
return nullptr;
} else {
return loopKeepAlive_;
}
}
private: private:
// TimeoutManager // TimeoutManager
void attachTimeoutManager(AsyncTimeout* obj, void attachTimeoutManager(AsyncTimeout* obj,
...@@ -602,6 +618,8 @@ class EventBase : private boost::noncopyable, ...@@ -602,6 +618,8 @@ class EventBase : private boost::noncopyable,
return isInEventBaseThread(); return isInEventBaseThread();
} }
void applyLoopKeepAlive();
/* /*
* Helper function that tells us whether we have already handled * Helper function that tells us whether we have already handled
* some event/timeout/callback in this loop iteration. * some event/timeout/callback in this loop iteration.
...@@ -673,6 +691,9 @@ class EventBase : private boost::noncopyable, ...@@ -673,6 +691,9 @@ class EventBase : private boost::noncopyable,
// to send function requests to the EventBase thread. // to send function requests to the EventBase thread.
std::unique_ptr<NotificationQueue<Func>> queue_; std::unique_ptr<NotificationQueue<Func>> queue_;
std::unique_ptr<FunctionRunner> fnRunner_; std::unique_ptr<FunctionRunner> fnRunner_;
LoopKeepAlive loopKeepAlive_{std::make_shared<int>(42)};
bool loopKeepAliveActive_{false};
std::atomic<bool> loopForeverActive_{false};
// limit for latency in microseconds (0 disables) // limit for latency in microseconds (0 disables)
int64_t maxLatency_; int64_t maxLatency_;
......
...@@ -1728,3 +1728,41 @@ TEST(EventBaseTest, RunCallbacksOnDestruction) { ...@@ -1728,3 +1728,41 @@ TEST(EventBaseTest, RunCallbacksOnDestruction) {
ASSERT_TRUE(ran); ASSERT_TRUE(ran);
} }
TEST(EventBaseTest, LoopKeepAlive) {
EventBase evb;
bool done = false;
std::thread t([&, loopKeepAlive = evb.loopKeepAlive() ] {
/* sleep override */ std::this_thread::sleep_for(
std::chrono::milliseconds(100));
evb.runInEventBaseThread([&] { done = true; });
});
evb.loop();
ASSERT_TRUE(done);
t.join();
}
TEST(EventBaseTest, LoopKeepAliveInLoop) {
EventBase evb;
bool done = false;
std::thread t;
evb.runInEventBaseThread([&] {
t = std::thread([&, loopKeepAlive = evb.loopKeepAlive() ] {
/* sleep override */ std::this_thread::sleep_for(
std::chrono::milliseconds(100));
evb.runInEventBaseThread([&] { done = true; });
});
});
evb.loop();
ASSERT_TRUE(done);
t.join();
}
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