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

Revert NotificationQueue changes until the thread local changes land

Summary:
Revert NotificationQueue changes until the thread local changes land

(Note: this ignores all push blocking failures!)

Reviewed By: yfeldblum

Differential Revision: D7359322

fbshipit-source-id: c59c81c34b341e4a968b2ce14947b558b8db3a85
parent cf080bee
main v2022.02.14.00 v2022.02.07.00 v2022.01.31.00 v2022.01.24.00 v2022.01.17.00 v2022.01.10.00 v2022.01.03.00 v2021.12.27.00 v2021.12.20.00 v2021.12.13.00 v2021.12.06.00 v2021.11.29.00 v2021.11.15.00 v2021.11.08.00 v2021.11.01.00 v2021.10.25.00 v2021.10.18.00 v2021.10.11.00 v2021.10.04.00 v2021.09.27.00 v2021.09.20.00 v2021.09.13.00 v2021.09.06.00 v2021.08.30.00 v2021.08.23.00 v2021.08.02.00 v2021.07.22.00 v2021.07.20.01 v2021.07.20.00 v2021.06.28.00 v2021.06.14.00 v2021.06.07.00 v2021.05.31.00 v2021.05.24.00 v2021.05.17.00 v2021.05.10.00 v2021.05.03.00 v2021.04.26.00 v2021.04.19.00 v2021.04.12.00 v2021.04.05.00 v2021.03.29.00 v2021.03.22.00 v2021.03.15.00 v2021.03.08.00 v2021.03.01.00 v2021.02.22.00 v2021.02.15.00 v2021.02.08.00 v2021.02.01.00 v2021.01.25.00 v2021.01.18.01 v2021.01.18.00 v2021.01.11.00 v2021.01.04.00 v2020.12.28.00 v2020.12.21.00 v2020.12.14.00 v2020.12.07.00 v2020.11.30.00 v2020.11.23.00 v2020.11.16.00 v2020.11.09.00 v2020.11.02.00 v2020.10.26.00 v2020.10.19.00 v2020.10.12.00 v2020.10.05.00 v2020.09.28.00 v2020.09.21.00 v2020.09.14.00 v2020.09.07.00 v2020.08.31.00 v2020.08.24.00 v2020.08.17.00 v2020.08.10.00 v2020.08.03.00 v2020.07.27.00 v2020.07.20.00 v2020.07.13.00 v2020.07.06.00 v2020.06.29.00 v2020.06.15.00 v2020.06.08.00 v2020.06.01.00 v2020.05.25.00 v2020.05.18.00 v2020.05.11.00 v2020.05.04.00 v2020.04.27.00 v2020.04.20.00 v2020.04.13.00 v2020.04.06.00 v2020.03.30.00 v2020.03.23.00 v2020.03.16.00 v2020.03.09.00 v2020.03.02.00 v2020.02.24.00 v2020.02.17.00 v2020.02.10.00 v2020.02.03.00 v2020.01.27.00 v2020.01.20.00 v2020.01.13.00 v2020.01.06.00 v2019.12.30.00 v2019.12.23.00 v2019.12.16.00 v2019.12.09.00 v2019.12.06.00 v2019.12.02.00 v2019.11.11.00 v2019.11.04.00 v2019.10.28.00 v2019.10.21.00 v2019.10.14.00 v2019.10.07.00 v2019.09.30.00 v2019.09.23.00 v2019.09.16.00 v2019.09.09.00 v2019.09.02.00 v2019.08.26.00 v2019.08.19.00 v2019.08.12.00 v2019.08.05.00 v2019.07.29.00 v2019.07.22.00 v2019.06.17.00 v2019.06.10.00 v2019.06.03.00 v2019.05.27.00 v2019.05.20.00 v2019.05.13.00 v2019.05.06.00 v2019.04.29.00 v2019.04.22.00 v2019.04.15.00 v2019.04.08.00 v2019.04.01.00 v2019.03.25.00 v2019.03.18.00 v2019.03.04.00 v2019.02.25.00 v2019.02.18.00 v2019.02.11.00 v2019.02.04.00 v2019.01.28.00 v2019.01.21.00 v2019.01.14.00 v2019.01.07.00 v2018.12.31.00 v2018.12.24.00 v2018.12.17.00 v2018.12.10.00 v2018.12.03.00 v2018.11.26.00 v2018.11.19.00 v2018.11.12.00 v2018.11.05.00 v2018.10.29.00 v2018.10.22.00 v2018.10.15.00 v2018.10.08.00 v2018.10.01.00 v2018.09.24.00 v2018.09.17.00 v2018.09.10.01 v2018.09.10.00 v2018.09.03.01 v2018.09.03.00 v2018.08.27.00 v2018.08.20.00 v2018.08.13.00 v2018.08.09.00 v2018.08.06.00 v2018.07.30.00 v2018.07.23.00 v2018.07.16.00 v2018.07.09.00 v2018.07.02.00 v2018.06.25.00 v2018.06.18.00 v2018.06.11.00 v2018.06.04.00 v2018.05.28.00 v2018.05.21.00 v2018.05.14.00 v2018.05.07.00 v2018.04.30.00 v2018.04.23.00 v2018.04.16.00 v2018.04.09.00 v2018.04.02.00 v2018.03.26.00
No related merge requests found
......@@ -19,6 +19,7 @@
#include <sys/types.h>
#include <algorithm>
#include <deque>
#include <iterator>
#include <memory>
#include <stdexcept>
......@@ -27,10 +28,8 @@
#include <folly/Exception.h>
#include <folly/FileUtil.h>
#include <folly/Likely.h>
#include <folly/Optional.h>
#include <folly/ScopeGuard.h>
#include <folly/SpinLock.h>
#include <folly/concurrency/UnboundedQueue.h>
#include <folly/io/async/DelayedDestruction.h>
#include <folly/io/async/EventBase.h>
#include <folly/io/async/EventHandler.h>
......@@ -187,6 +186,25 @@ class NotificationQueue {
* (1) Well, maybe. See logic/comments around "wasEmpty" in implementation.
*/
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);
NotificationQueue* queue_;
......@@ -199,9 +217,12 @@ class NotificationQueue {
class SimpleConsumer {
public:
explicit SimpleConsumer(NotificationQueue& queue) : queue_(queue) {
++queue_.numConsumers_;
}
~SimpleConsumer() = default;
~SimpleConsumer() {
--queue_.numConsumers_;
}
int getFd() const {
return queue_.eventfd_ >= 0 ? queue_.eventfd_ : queue_.pipeFds_[0];
......@@ -389,18 +410,23 @@ class NotificationQueue {
checkPid();
folly::Optional<Pair> data = queue_.try_dequeue();
if (!data) {
folly::SpinLockGuard g(spinlock_);
if (UNLIKELY(queue_.empty())) {
return false;
}
result = std::move(data.value().first);
RequestContext::setContext(std::move(data.value().second));
auto& data = queue_.front();
result = std::move(data.first);
RequestContext::setContext(std::move(data.second));
queue_.pop_front();
return true;
}
size_t size() const {
folly::SpinLockGuard g(spinlock_);
return queue_.size();
}
......@@ -425,6 +451,7 @@ class NotificationQueue {
NotificationQueue& operator=(NotificationQueue const &) = delete;
inline bool checkQueueSize(size_t maxSize, bool throws=true) const {
DCHECK(0 == spinlock_.try_lock());
if (maxSize > 0 && queue_.size() >= maxSize) {
if (throws) {
throw std::overflow_error("unable to add message to NotificationQueue: "
......@@ -436,11 +463,10 @@ class NotificationQueue {
}
inline bool checkDraining(bool throws=true) {
auto draining = draining_.load(std::memory_order_relaxed);
if (UNLIKELY(draining && throws)) {
if (UNLIKELY(draining_ && throws)) {
throw std::runtime_error("queue is draining, cannot add message");
}
return draining;
return draining_;
}
#ifdef __ANDROID__
......@@ -492,10 +518,6 @@ class NotificationQueue {
}
void drainSignalsLocked() {
if (!signal_) {
return;
}
ssize_t bytes_read = 0;
if (eventfd_ > 0) {
uint64_t message;
......@@ -544,12 +566,23 @@ class NotificationQueue {
template <typename MessageTT>
bool putMessageImpl(MessageTT&& message, size_t maxSize, bool throws = true) {
checkPid();
if (checkDraining(throws) || !checkQueueSize(maxSize, throws)) {
return false;
bool signal = 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;
}
......@@ -557,12 +590,23 @@ class NotificationQueue {
void putMessagesImpl(InputIteratorT first, InputIteratorT last,
std::input_iterator_tag) {
checkPid();
checkDraining();
while (first != last) {
queue_.enqueue(std::make_pair(*first, RequestContext::saveContext()));
++first;
bool signal = false;
size_t numAdded = 0;
{
folly::SpinLockGuard g(spinlock_);
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_;
......@@ -571,9 +615,10 @@ class NotificationQueue {
int pipeFds_[2]; // to fallback to on older/non-linux systems
uint32_t advisoryMaxQueueSize_;
pid_t pid_;
using Pair = std::pair<MessageT, std::shared_ptr<RequestContext>>;
UMPMCQueue<Pair, false> queue_;
std::atomic<bool> draining_{false};
std::deque<std::pair<MessageT, std::shared_ptr<RequestContext>>> queue_;
int numConsumers_{0};
std::atomic<int> numActiveConsumers_{0};
bool draining_{false};
};
template <typename MessageT>
......@@ -600,11 +645,15 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
bool isDrain, size_t* numConsumed) noexcept {
DestructorGuard dg(this);
uint32_t numProcessed = 0;
setActive(true);
SCOPE_EXIT {
if (queue_) {
queue_->syncSignalAndQueue();
}
};
SCOPE_EXIT {
setActive(false, /* shouldLock = */ true);
};
SCOPE_EXIT {
if (numConsumed != nullptr) {
*numConsumed = numProcessed;
......@@ -612,17 +661,42 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
};
while (true) {
// 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 {
// Pull a message off the queue.
folly::Optional<Pair> data = queue_->queue_.try_dequeue();
if (!data) {
if (UNLIKELY(queue_->queue_.empty())) {
// If there is no message, we've reached the end of the queue, return.
setActive(false);
queue_->spinlock_.unlock();
return;
}
MessageT msg(std::move(data.value().first));
RequestContextScopeGuard rctx(std::move(data.value().second));
// Pull a message off the queue.
auto& data = queue_->queue_.front();
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();
if (wasEmpty) {
setActive(false);
}
// Now unlock the spinlock before we invoke the callback.
queue_->spinlock_.unlock();
locked = false;
// Call the callback
bool callbackDestroyed = false;
......@@ -653,11 +727,10 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
// 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
// will simply be woken up the next time around the event loop and will
// // process the message then.)
// process the message then.)
if (wasEmpty) {
return;
}
} catch (const std::exception&) {
// This catch block is really just to handle the case where the MessageT
// constructor throws. The messageAvailable() callback itself is
......@@ -669,6 +742,10 @@ void NotificationQueue<MessageT>::Consumer::consumeMessages(
// trying to read the message again. If MessageT continues to throw we
// will never make forward progress and will keep trying each time around
// the event loop.
if (locked) {
// Unlock the spinlock.
queue_->spinlock_.unlock();
}
return;
}
......@@ -688,6 +765,10 @@ void NotificationQueue<MessageT>::Consumer::init(
queue_ = queue;
{
folly::SpinLockGuard g(queue_->spinlock_);
queue_->numConsumers_++;
}
queue_->ensureSignal();
if (queue_->eventfd_ >= 0) {
......@@ -704,6 +785,12 @@ void NotificationQueue<MessageT>::Consumer::stopConsuming() {
return;
}
{
folly::SpinLockGuard g(queue_->spinlock_);
queue_->numConsumers_--;
setActive(false);
}
assert(isHandlerRegistered());
unregisterHandler();
detachEventBase();
......@@ -714,11 +801,18 @@ template <typename MessageT>
bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained(
size_t* numConsumed) noexcept {
DestructorGuard dg(this);
if (queue_->draining_.exchange(true, std::memory_order_relaxed)) {
return false;
{
folly::SpinLockGuard g(queue_->spinlock_);
if (queue_->draining_) {
return false;
}
queue_->draining_ = true;
}
consumeMessages(true, numConsumed);
queue_->draining_.store(false, std::memory_order_relaxed);
{
folly::SpinLockGuard g(queue_->spinlock_);
queue_->draining_ = false;
}
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