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

Thread-safe version of loopKeepAlive()

Reviewed By: yfeldblum

Differential Revision: D4152380

fbshipit-source-id: 8b3c6dc4b14b9138bb5012e05f50496e51c0fa4b
parent bfa9b99a
...@@ -176,7 +176,7 @@ EventBase::~EventBase() { ...@@ -176,7 +176,7 @@ EventBase::~EventBase() {
// Keep looping until all keep-alive handles are released. Each keep-alive // Keep looping until all keep-alive handles are released. Each keep-alive
// handle signals that some external code will still schedule some work on // handle signals that some external code will still schedule some work on
// this EventBase (so it's not safe to destroy it). // this EventBase (so it's not safe to destroy it).
while (loopKeepAliveCount_ > 0) { while (loopKeepAliveCount() > 0) {
applyLoopKeepAlive(); applyLoopKeepAlive();
loopOnce(); loopOnce();
} }
...@@ -412,13 +412,22 @@ bool EventBase::loopBody(int flags) { ...@@ -412,13 +412,22 @@ bool EventBase::loopBody(int flags) {
return true; return true;
} }
ssize_t EventBase::loopKeepAliveCount() {
if (loopKeepAliveCountAtomic_.load(std::memory_order_relaxed)) {
loopKeepAliveCount_ +=
loopKeepAliveCountAtomic_.exchange(0, std::memory_order_relaxed);
}
DCHECK_GE(loopKeepAliveCount_, 0);
return loopKeepAliveCount_;
}
void EventBase::applyLoopKeepAlive() { void EventBase::applyLoopKeepAlive() {
if (loopKeepAliveActive_ && loopKeepAliveCount_ == 0) { if (loopKeepAliveActive_ && loopKeepAliveCount() == 0) {
// Restore the notification queue internal flag // Restore the notification queue internal flag
fnRunner_->stopConsuming(); fnRunner_->stopConsuming();
fnRunner_->startConsumingInternal(this, queue_.get()); fnRunner_->startConsumingInternal(this, queue_.get());
loopKeepAliveActive_ = false; loopKeepAliveActive_ = false;
} else if (!loopKeepAliveActive_ && loopKeepAliveCount_ > 0) { } else if (!loopKeepAliveActive_ && loopKeepAliveCount() > 0) {
// Update the notification queue event to treat it as a normal // Update the notification queue event to treat it as a normal
// (non-internal) event. The notification queue event always remains // (non-internal) event. The notification queue event always remains
// installed, and the main loop won't exit with it installed. // installed, and the main loop won't exit with it installed.
...@@ -435,7 +444,12 @@ void EventBase::loopForever() { ...@@ -435,7 +444,12 @@ void EventBase::loopForever() {
applyLoopKeepAlive(); applyLoopKeepAlive();
}; };
// Make sure notification queue events are treated as normal events. // Make sure notification queue events are treated as normal events.
auto keepAlive = loopKeepAlive(); // We can't use loopKeepAlive() here since LoopKeepAlive token can only be
// released inside a loop.
++loopKeepAliveCount_;
SCOPE_EXIT {
--loopKeepAliveCount_;
};
ret = loop(); ret = loop();
} }
......
...@@ -37,6 +37,7 @@ ...@@ -37,6 +37,7 @@
#include <folly/Executor.h> #include <folly/Executor.h>
#include <folly/Function.h> #include <folly/Function.h>
#include <folly/Portability.h> #include <folly/Portability.h>
#include <folly/ScopeGuard.h>
#include <folly/experimental/ExecutionObserver.h> #include <folly/experimental/ExecutionObserver.h>
#include <folly/futures/DrivableExecutor.h> #include <folly/futures/DrivableExecutor.h>
#include <folly/io/async/AsyncTimeout.h> #include <folly/io/async/AsyncTimeout.h>
...@@ -555,7 +556,12 @@ class EventBase : private boost::noncopyable, ...@@ -555,7 +556,12 @@ class EventBase : private boost::noncopyable,
/// Implements the DrivableExecutor interface /// Implements the DrivableExecutor interface
void drive() override { void drive() override {
auto keepAlive = loopKeepAlive(); // We can't use loopKeepAlive() here since LoopKeepAlive token can only be
// released inside a loop.
++loopKeepAliveCount_;
SCOPE_EXIT {
--loopKeepAliveCount_;
};
loopOnce(); loopOnce();
} }
...@@ -579,6 +585,15 @@ class EventBase : private boost::noncopyable, ...@@ -579,6 +585,15 @@ class EventBase : private boost::noncopyable,
return LoopKeepAlive(this); return LoopKeepAlive(this);
} }
// Thread-safe version of loopKeepAlive()
LoopKeepAlive loopKeepAliveAtomic() {
if (inRunningEventBaseThread()) {
return loopKeepAlive();
}
loopKeepAliveCountAtomic_.fetch_add(1, std::memory_order_relaxed);
return LoopKeepAlive(this);
}
// TimeoutManager // TimeoutManager
void attachTimeoutManager( void attachTimeoutManager(
AsyncTimeout* obj, AsyncTimeout* obj,
...@@ -598,6 +613,8 @@ class EventBase : private boost::noncopyable, ...@@ -598,6 +613,8 @@ class EventBase : private boost::noncopyable,
private: private:
void applyLoopKeepAlive(); void applyLoopKeepAlive();
ssize_t loopKeepAliveCount();
/* /*
* 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.
...@@ -645,7 +662,8 @@ class EventBase : private boost::noncopyable, ...@@ -645,7 +662,8 @@ 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_;
size_t loopKeepAliveCount_{0}; ssize_t loopKeepAliveCount_{0};
std::atomic<ssize_t> loopKeepAliveCountAtomic_{0};
bool loopKeepAliveActive_{false}; bool loopKeepAliveActive_{false};
// limit for latency in microseconds (0 disables) // limit for latency in microseconds (0 disables)
......
...@@ -1826,6 +1826,51 @@ TEST(EventBaseTest, LoopKeepAliveShutdown) { ...@@ -1826,6 +1826,51 @@ TEST(EventBaseTest, LoopKeepAliveShutdown) {
t.join(); t.join();
} }
TEST(EventBaseTest, LoopKeepAliveAtomic) {
auto evb = folly::make_unique<EventBase>();
constexpr size_t kNumThreads = 100;
constexpr size_t kNumTasks = 100;
std::vector<std::thread> ts;
std::vector<std::unique_ptr<Baton<>>> batons;
size_t done{0};
for (size_t i = 0; i < kNumThreads; ++i) {
batons.emplace_back(std::make_unique<Baton<>>());
}
for (size_t i = 0; i < kNumThreads; ++i) {
ts.emplace_back([ evbPtr = evb.get(), batonPtr = batons[i].get(), &done ] {
std::vector<EventBase::LoopKeepAlive> keepAlives;
for (size_t i = 0; i < kNumTasks; ++i) {
keepAlives.emplace_back(evbPtr->loopKeepAliveAtomic());
}
batonPtr->post();
/* sleep override */ std::this_thread::sleep_for(std::chrono::seconds(1));
for (auto& keepAlive : keepAlives) {
evbPtr->runInEventBaseThread(
[&done, keepAlive = std::move(keepAlive) ]() { ++done; });
}
});
}
for (auto& baton : batons) {
baton->wait();
}
evb.reset();
EXPECT_EQ(kNumThreads * kNumTasks, done);
for (auto& t : ts) {
t.join();
}
}
TEST(EventBaseTest, DrivableExecutorTest) { TEST(EventBaseTest, DrivableExecutorTest) {
folly::Promise<bool> p; folly::Promise<bool> p;
auto f = p.getFuture(); auto f = p.getFuture();
......
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