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

Use unbounded queues

Summary: There's no way we can size such queues appropriately.

Reviewed By: yfeldblum

Differential Revision: D15271382

fbshipit-source-id: 7d8c5b0a3faffac3e86cb87fa18d0597facc3189
parent 2b928806
...@@ -19,9 +19,9 @@ ...@@ -19,9 +19,9 @@
#include <folly/ExceptionString.h> #include <folly/ExceptionString.h>
#include <folly/Format.h> #include <folly/Format.h>
#include <folly/MPMCQueue.h>
#include <folly/Range.h> #include <folly/Range.h>
#include <folly/Singleton.h> #include <folly/Singleton.h>
#include <folly/concurrency/UnboundedQueue.h>
#include <folly/portability/GFlags.h> #include <folly/portability/GFlags.h>
#include <folly/system/ThreadName.h> #include <folly/system/ThreadName.h>
...@@ -37,16 +37,14 @@ DEFINE_int32( ...@@ -37,16 +37,14 @@ DEFINE_int32(
4, 4,
"How many internal threads ObserverManager should use"); "How many internal threads ObserverManager should use");
static constexpr StringPiece kObserverManagerThreadNamePrefix{"ObserverMngr"};
namespace { namespace {
constexpr size_t kCurrentQueueSize{10 * 1024}; constexpr StringPiece kObserverManagerThreadNamePrefix{"ObserverMngr"};
constexpr size_t kNextQueueSize{10 * 1024}; constexpr size_t kNextBatchSize{1024};
} // namespace } // namespace
class ObserverManager::CurrentQueue { class ObserverManager::CurrentQueue {
public: public:
CurrentQueue() : queue_(kCurrentQueueSize) { CurrentQueue() {
if (FLAGS_observer_manager_pool_size < 1) { if (FLAGS_observer_manager_pool_size < 1) {
LOG(ERROR) << "--observer_manager_pool_size should be >= 1"; LOG(ERROR) << "--observer_manager_pool_size should be >= 1";
FLAGS_observer_manager_pool_size = 1; FLAGS_observer_manager_pool_size = 1;
...@@ -59,7 +57,7 @@ class ObserverManager::CurrentQueue { ...@@ -59,7 +57,7 @@ class ObserverManager::CurrentQueue {
while (true) { while (true) {
Function<void()> task; Function<void()> task;
queue_.blockingRead(task); queue_.dequeue(task);
if (!task) { if (!task) {
return; return;
...@@ -78,40 +76,33 @@ class ObserverManager::CurrentQueue { ...@@ -78,40 +76,33 @@ class ObserverManager::CurrentQueue {
~CurrentQueue() { ~CurrentQueue() {
for (size_t i = 0; i < threads_.size(); ++i) { for (size_t i = 0; i < threads_.size(); ++i) {
queue_.blockingWrite(nullptr); queue_.enqueue(nullptr);
} }
for (auto& thread : threads_) { for (auto& thread : threads_) {
thread.join(); thread.join();
} }
CHECK(queue_.isEmpty()); CHECK(queue_.empty());
} }
void add(Function<void()> task) { void add(Function<void()> task) {
if (ObserverManager::inManagerThread()) { queue_.enqueue(std::move(task));
if (!queue_.write(std::move(task))) {
throw std::runtime_error("Too many Observers scheduled for update.");
}
} else {
queue_.blockingWrite(std::move(task));
}
} }
private: private:
MPMCQueue<Function<void()>> queue_; UMPMCQueue<Function<void()>, true> queue_;
std::vector<std::thread> threads_; std::vector<std::thread> threads_;
}; };
class ObserverManager::NextQueue { class ObserverManager::NextQueue {
public: public:
explicit NextQueue(ObserverManager& manager) explicit NextQueue(ObserverManager& manager) : manager_(manager) {
: manager_(manager), queue_(kNextQueueSize) {
thread_ = std::thread([&]() { thread_ = std::thread([&]() {
Core::WeakPtr queueCoreWeak; Core::WeakPtr queueCoreWeak;
while (true) { while (true) {
queue_.blockingRead(queueCoreWeak); queue_.dequeue(queueCoreWeak);
if (stop_) { if (stop_) {
return; return;
} }
...@@ -128,7 +119,8 @@ class ObserverManager::NextQueue { ...@@ -128,7 +119,8 @@ class ObserverManager::NextQueue {
// We can't pick more tasks from the queue after we bumped the // We can't pick more tasks from the queue after we bumped the
// version, so we have to do this while holding the lock. // version, so we have to do this while holding the lock.
while (cores.size() < kNextQueueSize && queue_.read(queueCoreWeak)) { while (cores.size() < kNextBatchSize &&
queue_.try_dequeue(queueCoreWeak)) {
if (stop_) { if (stop_) {
return; return;
} }
...@@ -152,7 +144,7 @@ class ObserverManager::NextQueue { ...@@ -152,7 +144,7 @@ class ObserverManager::NextQueue {
auto wEmptyWaiters = emptyWaiters_.wlock(); auto wEmptyWaiters = emptyWaiters_.wlock();
// We don't want any new waiters to be added while we are checking the // We don't want any new waiters to be added while we are checking the
// queue. // queue.
if (queue_.isEmpty()) { if (queue_.empty()) {
for (auto& promise : *wEmptyWaiters) { for (auto& promise : *wEmptyWaiters) {
promise.set_value(); promise.set_value();
} }
...@@ -164,13 +156,13 @@ class ObserverManager::NextQueue { ...@@ -164,13 +156,13 @@ class ObserverManager::NextQueue {
} }
void add(Core::WeakPtr core) { void add(Core::WeakPtr core) {
queue_.blockingWrite(std::move(core)); queue_.enqueue(std::move(core));
} }
~NextQueue() { ~NextQueue() {
stop_ = true; stop_ = true;
// Write to the queue to notify the thread. // Write to the queue to notify the thread.
queue_.blockingWrite(Core::WeakPtr()); queue_.enqueue(Core::WeakPtr());
thread_.join(); thread_.join();
} }
...@@ -180,14 +172,14 @@ class ObserverManager::NextQueue { ...@@ -180,14 +172,14 @@ class ObserverManager::NextQueue {
emptyWaiters_.wlock()->push_back(std::move(promise)); emptyWaiters_.wlock()->push_back(std::move(promise));
// Write to the queue to notify the thread. // Write to the queue to notify the thread.
queue_.blockingWrite(Core::WeakPtr()); queue_.enqueue(Core::WeakPtr());
future.get(); future.get();
} }
private: private:
ObserverManager& manager_; ObserverManager& manager_;
MPMCQueue<Core::WeakPtr> queue_; UMPSCQueue<Core::WeakPtr, true> queue_;
std::thread thread_; std::thread thread_;
std::atomic<bool> stop_{false}; std::atomic<bool> stop_{false};
folly::Synchronized<std::vector<std::promise<void>>> emptyWaiters_; folly::Synchronized<std::vector<std::promise<void>>> emptyWaiters_;
......
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