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

Use unbounded queue in NotificationQueue (same as in D7164130 but with the...

Use unbounded queue in NotificationQueue (same as in D7164130 but with the hazptr fix from D7256905 and the singleton one from D7283390)

Summary: Use unbounded queue in NotificationQueue (same as in D7164130 but with the hazptr fix from D7256905 and the singleton one from D7283390)

Reviewed By: yfeldblum

Differential Revision: D7278275

fbshipit-source-id: 78ea94e1c9d492febd33106b06ce275d4a90a56a
parent a2aff3ba
...@@ -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;
{
folly::SpinLockGuard g(spinlock_);
if (checkDraining(throws) || !checkQueueSize(maxSize, throws)) { if (checkDraining(throws) || !checkQueueSize(maxSize, throws)) {
return false; return false;
} }
// We only need to signal an event if not all consumers are queue_.enqueue(std::make_pair(
// awake. std::forward<MessageTT>(message), RequestContext::saveContext()));
if (numActiveConsumers_ < numConsumers_) { ensureSignal();
signal = true;
}
queue_.emplace_back(
std::forward<MessageTT>(message), RequestContext::saveContext());
if (signal) {
ensureSignalLocked();
}
}
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;
size_t numAdded = 0;
{
folly::SpinLockGuard g(spinlock_);
checkDraining(); checkDraining();
while (first != last) { while (first != last) {
queue_.emplace_back(*first, RequestContext::saveContext()); queue_.enqueue(std::make_pair(*first, RequestContext::saveContext()));
++first; ++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,13 +600,11 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages( ...@@ -645,13 +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;
...@@ -659,42 +612,17 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages( ...@@ -659,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;
...@@ -725,10 +653,11 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages( ...@@ -725,10 +653,11 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
// looping again and trying to re-read from the eventfd. (If a new // looping again and trying to re-read from the eventfd. (If a new
// message had in fact arrived while we were invoking the callback, we // message had in fact arrived while we were invoking the callback, we
// will simply be woken up the next time around the event loop and will // will simply be woken up the next time around the event loop and will
// process the message then.) // // process the message then.)
if (wasEmpty) { if (wasEmpty) {
return; return;
} }
} catch (const std::exception&) { } catch (const std::exception&) {
// This catch block is really just to handle the case where the MessageT // This catch block is really just to handle the case where the MessageT
// constructor throws. The messageAvailable() callback itself is // constructor throws. The messageAvailable() callback itself is
...@@ -740,10 +669,6 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages( ...@@ -740,10 +669,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;
} }
...@@ -763,10 +688,6 @@ void NotificationQueue<MessageT>::Consumer::init( ...@@ -763,10 +688,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) {
...@@ -783,12 +704,6 @@ void NotificationQueue<MessageT>::Consumer::stopConsuming() { ...@@ -783,12 +704,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();
...@@ -799,18 +714,11 @@ template <typename MessageT> ...@@ -799,18 +714,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_);
if (queue_->draining_) {
return false; 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