Commit d25334fb authored by Dan Melnic's avatar Dan Melnic Committed by Facebook Github Bot

Use unbounded queue in NotificationQueue

Summary: Use unbounded queue in NotificationQueue (prev changes: D7164130, D7251343, D7278275, D7359322, D7471153, D7606076)

Reviewed By: djwatson

Differential Revision: D7408628

fbshipit-source-id: 04ae235d9b3de603e28115e11fe98fe7e9aa1f8c
parent e752dcd0
...@@ -387,9 +387,19 @@ struct StaticMeta : StaticMetaBase { ...@@ -387,9 +387,19 @@ struct StaticMeta : StaticMetaBase {
#else #else
threadEntry = new ThreadEntry(); threadEntry = new ThreadEntry();
#endif #endif
threadEntry->list = threadEntryList; // if the ThreadEntry already exists
threadEntry->listNext = threadEntryList->head; // but pthread_getspecific returns NULL
threadEntryList->head = threadEntry; // do not add the same entry twice to the list
// since this would create a loop in the list
if (!threadEntry->list) {
threadEntry->list = threadEntryList;
threadEntry->listNext = threadEntryList->head;
threadEntryList->head = threadEntry;
}
// if we're adding a thread entry
// we need to increment the list count
// even if the entry is reused
threadEntryList->count++; threadEntryList->count++;
threadEntry->meta = &meta; threadEntry->meta = &meta;
......
...@@ -19,7 +19,6 @@ ...@@ -19,7 +19,6 @@
#include <sys/types.h> #include <sys/types.h>
#include <algorithm> #include <algorithm>
#include <deque>
#include <iterator> #include <iterator>
#include <memory> #include <memory>
#include <stdexcept> #include <stdexcept>
...@@ -28,8 +27,10 @@ ...@@ -28,8 +27,10 @@
#include <folly/Exception.h> #include <folly/Exception.h>
#include <folly/FileUtil.h> #include <folly/FileUtil.h>
#include <folly/Likely.h> #include <folly/Likely.h>
#include <folly/Optional.h>
#include <folly/ScopeGuard.h> #include <folly/ScopeGuard.h>
#include <folly/SpinLock.h> #include <folly/SpinLock.h>
#include <folly/concurrency/UnboundedQueue.h>
#include <folly/io/async/DelayedDestruction.h> #include <folly/io/async/DelayedDestruction.h>
#include <folly/io/async/EventBase.h> #include <folly/io/async/EventBase.h>
#include <folly/io/async/EventHandler.h> #include <folly/io/async/EventHandler.h>
...@@ -186,25 +187,6 @@ class NotificationQueue { ...@@ -186,25 +187,6 @@ class NotificationQueue {
* (1) Well, maybe. See logic/comments around "wasEmpty" in implementation. * (1) Well, maybe. See logic/comments around "wasEmpty" in implementation.
*/ */
void consumeMessages(bool isDrain, size_t* numConsumed = nullptr) noexcept; void consumeMessages(bool isDrain, size_t* numConsumed = nullptr) noexcept;
void setActive(bool active, bool shouldLock = false) {
if (!queue_) {
active_ = active;
return;
}
if (shouldLock) {
queue_->spinlock_.lock();
}
if (!active_ && active) {
++queue_->numActiveConsumers_;
} else if (active_ && !active) {
--queue_->numActiveConsumers_;
}
active_ = active;
if (shouldLock) {
queue_->spinlock_.unlock();
}
}
void init(EventBase* eventBase, NotificationQueue* queue); void init(EventBase* eventBase, NotificationQueue* queue);
NotificationQueue* queue_; NotificationQueue* queue_;
...@@ -217,12 +199,9 @@ class NotificationQueue { ...@@ -217,12 +199,9 @@ class NotificationQueue {
class SimpleConsumer { class SimpleConsumer {
public: public:
explicit SimpleConsumer(NotificationQueue& queue) : queue_(queue) { explicit SimpleConsumer(NotificationQueue& queue) : queue_(queue) {
++queue_.numConsumers_;
} }
~SimpleConsumer() { ~SimpleConsumer() = default;
--queue_.numConsumers_;
}
int getFd() const { int getFd() const {
return queue_.eventfd_ >= 0 ? queue_.eventfd_ : queue_.pipeFds_[0]; return queue_.eventfd_ >= 0 ? queue_.eventfd_ : queue_.pipeFds_[0];
...@@ -410,23 +389,18 @@ class NotificationQueue { ...@@ -410,23 +389,18 @@ class NotificationQueue {
checkPid(); checkPid();
folly::SpinLockGuard g(spinlock_); folly::Optional<Pair> data = queue_.try_dequeue();
if (!data) {
if (UNLIKELY(queue_.empty())) {
return false; return false;
} }
auto& data = queue_.front(); result = std::move(data.value().first);
result = std::move(data.first); RequestContext::setContext(std::move(data.value().second));
RequestContext::setContext(std::move(data.second));
queue_.pop_front();
return true; return true;
} }
size_t size() const { size_t size() const {
folly::SpinLockGuard g(spinlock_);
return queue_.size(); return queue_.size();
} }
...@@ -451,7 +425,6 @@ class NotificationQueue { ...@@ -451,7 +425,6 @@ class NotificationQueue {
NotificationQueue& operator=(NotificationQueue const &) = delete; NotificationQueue& operator=(NotificationQueue const &) = delete;
inline bool checkQueueSize(size_t maxSize, bool throws=true) const { inline bool checkQueueSize(size_t maxSize, bool throws=true) const {
DCHECK(0 == spinlock_.try_lock());
if (maxSize > 0 && queue_.size() >= maxSize) { if (maxSize > 0 && queue_.size() >= maxSize) {
if (throws) { if (throws) {
throw std::overflow_error("unable to add message to NotificationQueue: " throw std::overflow_error("unable to add message to NotificationQueue: "
...@@ -463,10 +436,11 @@ class NotificationQueue { ...@@ -463,10 +436,11 @@ class NotificationQueue {
} }
inline bool checkDraining(bool throws=true) { inline bool checkDraining(bool throws=true) {
if (UNLIKELY(draining_ && throws)) { auto draining = draining_.load(std::memory_order_relaxed);
if (UNLIKELY(draining && throws)) {
throw std::runtime_error("queue is draining, cannot add message"); throw std::runtime_error("queue is draining, cannot add message");
} }
return draining_; return draining;
} }
#ifdef __ANDROID__ #ifdef __ANDROID__
...@@ -518,6 +492,10 @@ class NotificationQueue { ...@@ -518,6 +492,10 @@ class NotificationQueue {
} }
void drainSignalsLocked() { void drainSignalsLocked() {
if (!signal_) {
return;
}
ssize_t bytes_read = 0; ssize_t bytes_read = 0;
if (eventfd_ > 0) { if (eventfd_ > 0) {
uint64_t message; uint64_t message;
...@@ -566,23 +544,12 @@ class NotificationQueue { ...@@ -566,23 +544,12 @@ class NotificationQueue {
template <typename MessageTT> template <typename MessageTT>
bool putMessageImpl(MessageTT&& message, size_t maxSize, bool throws = true) { bool putMessageImpl(MessageTT&& message, size_t maxSize, bool throws = true) {
checkPid(); checkPid();
bool signal = false; if (checkDraining(throws) || !checkQueueSize(maxSize, throws)) {
{ return false;
folly::SpinLockGuard g(spinlock_);
if (checkDraining(throws) || !checkQueueSize(maxSize, throws)) {
return false;
}
// We only need to signal an event if not all consumers are
// awake.
if (numActiveConsumers_ < numConsumers_) {
signal = true;
}
queue_.emplace_back(
std::forward<MessageTT>(message), RequestContext::saveContext());
if (signal) {
ensureSignalLocked();
}
} }
queue_.enqueue(std::make_pair(
std::forward<MessageTT>(message), RequestContext::saveContext()));
ensureSignal();
return true; return true;
} }
...@@ -590,23 +557,12 @@ class NotificationQueue { ...@@ -590,23 +557,12 @@ class NotificationQueue {
void putMessagesImpl(InputIteratorT first, InputIteratorT last, void putMessagesImpl(InputIteratorT first, InputIteratorT last,
std::input_iterator_tag) { std::input_iterator_tag) {
checkPid(); checkPid();
bool signal = false; checkDraining();
size_t numAdded = 0; while (first != last) {
{ queue_.enqueue(std::make_pair(*first, RequestContext::saveContext()));
folly::SpinLockGuard g(spinlock_); ++first;
checkDraining();
while (first != last) {
queue_.emplace_back(*first, RequestContext::saveContext());
++first;
++numAdded;
}
if (numActiveConsumers_ < numConsumers_) {
signal = true;
}
if (signal) {
ensureSignalLocked();
}
} }
ensureSignal();
} }
mutable folly::SpinLock spinlock_; mutable folly::SpinLock spinlock_;
...@@ -615,10 +571,9 @@ class NotificationQueue { ...@@ -615,10 +571,9 @@ class NotificationQueue {
int pipeFds_[2]; // to fallback to on older/non-linux systems int pipeFds_[2]; // to fallback to on older/non-linux systems
uint32_t advisoryMaxQueueSize_; uint32_t advisoryMaxQueueSize_;
pid_t pid_; pid_t pid_;
std::deque<std::pair<MessageT, std::shared_ptr<RequestContext>>> queue_; using Pair = std::pair<MessageT, std::shared_ptr<RequestContext>>;
int numConsumers_{0}; UMPMCQueue<Pair, false> queue_;
std::atomic<int> numActiveConsumers_{0}; std::atomic<bool> draining_{false};
bool draining_{false};
}; };
template <typename MessageT> template <typename MessageT>
...@@ -645,15 +600,11 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages( ...@@ -645,15 +600,11 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
bool isDrain, size_t* numConsumed) noexcept { bool isDrain, size_t* numConsumed) noexcept {
DestructorGuard dg(this); DestructorGuard dg(this);
uint32_t numProcessed = 0; uint32_t numProcessed = 0;
setActive(true);
SCOPE_EXIT { SCOPE_EXIT {
if (queue_) { if (queue_) {
queue_->syncSignalAndQueue(); queue_->syncSignalAndQueue();
} }
}; };
SCOPE_EXIT {
setActive(false, /* shouldLock = */ true);
};
SCOPE_EXIT { SCOPE_EXIT {
if (numConsumed != nullptr) { if (numConsumed != nullptr) {
*numConsumed = numProcessed; *numConsumed = numProcessed;
...@@ -661,42 +612,17 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages( ...@@ -661,42 +612,17 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
}; };
while (true) { while (true) {
// Now pop the message off of the queue. // Now pop the message off of the queue.
//
// We have to manually acquire and release the spinlock here, rather than
// using SpinLockHolder since the MessageT has to be constructed while
// holding the spinlock and available after we release it. SpinLockHolder
// unfortunately doesn't provide a release() method. (We can't construct
// MessageT first since we have no guarantee that MessageT has a default
// constructor.
queue_->spinlock_.lock();
bool locked = true;
try { try {
if (UNLIKELY(queue_->queue_.empty())) { // Pull a message off the queue.
// If there is no message, we've reached the end of the queue, return. folly::Optional<Pair> data = queue_->queue_.try_dequeue();
setActive(false); if (!data) {
queue_->spinlock_.unlock();
return; return;
} }
// Pull a message off the queue. MessageT msg(std::move(data.value().first));
auto& data = queue_->queue_.front(); RequestContextScopeGuard rctx(std::move(data.value().second));
MessageT msg(std::move(data.first));
RequestContextScopeGuard rctx(std::move(data.second));
queue_->queue_.pop_front();
// Check to see if the queue is empty now.
// We use this as an optimization to see if we should bother trying to
// loop again and read another message after invoking this callback.
bool wasEmpty = queue_->queue_.empty(); bool wasEmpty = queue_->queue_.empty();
if (wasEmpty) {
setActive(false);
}
// Now unlock the spinlock before we invoke the callback.
queue_->spinlock_.unlock();
locked = false;
// Call the callback // Call the callback
bool callbackDestroyed = false; bool callbackDestroyed = false;
...@@ -742,10 +668,6 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages( ...@@ -742,10 +668,6 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
// trying to read the message again. If MessageT continues to throw we // trying to read the message again. If MessageT continues to throw we
// will never make forward progress and will keep trying each time around // will never make forward progress and will keep trying each time around
// the event loop. // the event loop.
if (locked) {
// Unlock the spinlock.
queue_->spinlock_.unlock();
}
return; return;
} }
...@@ -765,10 +687,6 @@ void NotificationQueue<MessageT>::Consumer::init( ...@@ -765,10 +687,6 @@ void NotificationQueue<MessageT>::Consumer::init(
queue_ = queue; queue_ = queue;
{
folly::SpinLockGuard g(queue_->spinlock_);
queue_->numConsumers_++;
}
queue_->ensureSignal(); queue_->ensureSignal();
if (queue_->eventfd_ >= 0) { if (queue_->eventfd_ >= 0) {
...@@ -785,12 +703,6 @@ void NotificationQueue<MessageT>::Consumer::stopConsuming() { ...@@ -785,12 +703,6 @@ void NotificationQueue<MessageT>::Consumer::stopConsuming() {
return; return;
} }
{
folly::SpinLockGuard g(queue_->spinlock_);
queue_->numConsumers_--;
setActive(false);
}
assert(isHandlerRegistered()); assert(isHandlerRegistered());
unregisterHandler(); unregisterHandler();
detachEventBase(); detachEventBase();
...@@ -801,18 +713,11 @@ template <typename MessageT> ...@@ -801,18 +713,11 @@ template <typename MessageT>
bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained( bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained(
size_t* numConsumed) noexcept { size_t* numConsumed) noexcept {
DestructorGuard dg(this); DestructorGuard dg(this);
{ if (queue_->draining_.exchange(true, std::memory_order_relaxed)) {
folly::SpinLockGuard g(queue_->spinlock_); return false;
if (queue_->draining_) {
return false;
}
queue_->draining_ = true;
} }
consumeMessages(true, numConsumed); consumeMessages(true, numConsumed);
{ queue_->draining_.store(false, std::memory_order_relaxed);
folly::SpinLockGuard g(queue_->spinlock_);
queue_->draining_ = false;
}
return true; return true;
} }
......
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