proxygen: folly::NotificationQueue< MessageT

#include <EventBase.h>

Classes
class	Consumer

struct	Node

class	SimpleConsumer

Public Types
enum	FdType { FdType::PIPE }

Public Member Functions
	NotificationQueue (uint32_t maxSize=0, FdType fdType=FdType::PIPE)

	~NotificationQueue ()

void	setMaxQueueSize (uint32_t max)

template<typename MessageTT >
void	tryPutMessage (MessageTT &&message)

template<typename MessageTT >
bool	tryPutMessageNoThrow (MessageTT &&message)

template<typename MessageTT >
void	putMessage (MessageTT &&message)

template<typename InputIteratorT >
void	putMessages (InputIteratorT first, InputIteratorT last)

bool	tryConsume (MessageT &result)

size_t	size () const

void	checkPid () const

Private Member Functions
	NotificationQueue (NotificationQueue const &)=delete

NotificationQueue &	operator= (NotificationQueue const &)=delete

bool	checkQueueSize (size_t maxSize, bool throws=true) const

bool	checkDraining (bool throws=true)

void	ensureSignalLocked () const

void	drainSignalsLocked ()

void	ensureSignal () const

void	syncSignalAndQueue ()

template<typename MessageTT >
bool	putMessageImpl (MessageTT &&message, size_t maxSize, bool throws=true)

template<typename InputIteratorT >
void	putMessagesImpl (InputIteratorT first, InputIteratorT last, std::input_iterator_tag)

Private Attributes
folly::SpinLock	spinlock_

bool	signal_ {false}

int	eventfd_

int	pipeFds_ [2]

uint32_t	advisoryMaxQueueSize_

pid_t	pid_

boost::intrusive::slist< Node, boost::intrusive::cache_last< true > >	queue_

int	numConsumers_ {0}

std::atomic< int >	numActiveConsumers_ {0}

bool	draining_ {false}

Detailed Description

template<typename MessageT>
class folly::NotificationQueue< MessageT >

A producer-consumer queue for passing messages between EventBase threads.

Messages can be added to the queue from any thread. Multiple consumers may listen to the queue from multiple EventBase threads.

A NotificationQueue may not be destroyed while there are still consumers registered to receive events from the queue. It is the user's responsibility to ensure that all consumers are unregistered before the queue is destroyed.

MessageT should be MoveConstructible (i.e., must support either a move constructor or a copy constructor, or both). Ideally it's move constructor (or copy constructor if no move constructor is provided) should never throw exceptions. If the constructor may throw, the consumers could end up spinning trying to move a message off the queue and failing, and then retrying.

Definition at line 56 of file EventBase.h.

Member Enumeration Documentation

template<typename MessageT>

enum folly::NotificationQueue::FdType

strong

Enumerator
PIPE

Definition at line 247 of file NotificationQueue.h.

                     {
     PIPE,
 #ifdef FOLLY_HAVE_EVENTFD
     EVENTFD,
 #endif
   };

Constructor & Destructor Documentation

template<typename MessageT>

folly::NotificationQueue< MessageT >::NotificationQueue	(	uint32_t	maxSize = `0`,
		FdType	fdType = `FdType::PIPE`
	)

inlineexplicit

Create a new NotificationQueue.

If the maxSize parameter is specified, this sets the maximum queue size that will be enforced by tryPutMessage(). (This size is advisory, and may be exceeded if producers explicitly use putMessage() instead of tryPutMessage().)

The fdType parameter determines the type of file descriptor used internally to signal message availability. The default (eventfd) is preferable for performance and because it won't fail when the queue gets too long. It is not available on on older and non-linux kernels, however. In this case the code will fall back to using a pipe, the parameter is mostly for testing purposes.

Definition at line 269 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::checkPid().

       : eventfd_(-1),
         pipeFds_{-1, -1},
         advisoryMaxQueueSize_(maxSize),
         pid_(pid_t(getpid())) {
 
 #ifdef FOLLY_HAVE_EVENTFD
     if (fdType == FdType::EVENTFD) {
       eventfd_ = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
       if (eventfd_ == -1) {
         if (errno == ENOSYS || errno == EINVAL) {
           // eventfd not availalble
           LOG(ERROR) << "failed to create eventfd for NotificationQueue: "
                      << errno << ", falling back to pipe mode (is your kernel "
                      << "> 2.6.30?)";
           fdType = FdType::PIPE;
         } else {
           // some other error
           folly::throwSystemError(
               "Failed to create eventfd for "
               "NotificationQueue",
               errno);
         }
       }
     }
 #endif
     if (fdType == FdType::PIPE) {
       if (pipe(pipeFds_)) {
         folly::throwSystemError(
             "Failed to create pipe for NotificationQueue", errno);
       }
       try {
         // put both ends of the pipe into non-blocking mode
         if (fcntl(pipeFds_[0], F_SETFL, O_RDONLY | O_NONBLOCK) != 0) {
           folly::throwSystemError(
               "failed to put NotificationQueue pipe read "
               "endpoint into non-blocking mode",
               errno);
         }
         if (fcntl(pipeFds_[1], F_SETFL, O_WRONLY | O_NONBLOCK) != 0) {
           folly::throwSystemError(
               "failed to put NotificationQueue pipe write "
               "endpoint into non-blocking mode",
               errno);
         }
       } catch (...) {
         ::close(pipeFds_[0]);
         ::close(pipeFds_[1]);
         throw;
       }
     }
   }

template<typename MessageT>

folly::NotificationQueue< MessageT >::~NotificationQueue ( )

inline

Definition at line 328 of file NotificationQueue.h.

                        {
     std::unique_ptr<Node> data;
     while (!queue_.empty()) {
       data.reset(&queue_.front());
       queue_.pop_front();
     }
     if (eventfd_ >= 0) {
       ::close(eventfd_);
       eventfd_ = -1;
     }
     if (pipeFds_[0] >= 0) {
       ::close(pipeFds_[0]);
       pipeFds_[0] = -1;
     }
     if (pipeFds_[1] >= 0) {
       ::close(pipeFds_[1]);
       pipeFds_[1] = -1;
     }
   }

template<typename MessageT>

folly::NotificationQueue< MessageT >::NotificationQueue ( NotificationQueue< MessageT > const & )

privatedelete

Member Function Documentation

template<typename MessageT>

bool folly::NotificationQueue< MessageT >::checkDraining ( bool throws = true )

inlineprivate

Definition at line 496 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessageImpl(), and folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessagesImpl().

                                                 {
     if (UNLIKELY(draining_ && throws)) {
       throw std::runtime_error("queue is draining, cannot add message");
     }
     return draining_;
   }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::checkPid ( ) const

inline

Check that the NotificationQueue is being used from the correct process.

If you create a NotificationQueue in one process, then fork, and try to send messages to the queue from the child process, you're going to have a bad time. Unfortunately users have (accidentally) run into this.

Because we use an eventfd/pipe, the child process can actually signal the parent process that an event is ready. However, it can't put anything on the parent's queue, so the parent wakes up and finds an empty queue. This check ensures that we catch the problem in the misbehaving child process code, and crash before signalling the parent process.

Definition at line 474 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessageImpl(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessagesImpl(), and folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::tryConsume().

                         {
     CHECK_EQ(pid_, pid_t(getpid()));
   }

template<typename MessageT>

bool folly::NotificationQueue< MessageT >::checkQueueSize	(	size_t	maxSize,
		bool	throws = `true`
	)		const

inlineprivate

Definition at line 483 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessageImpl().

                                                                        {
     DCHECK(0 == spinlock_.try_lock());
     if (maxSize > 0 && queue_.size() >= maxSize) {
       if (throws) {
         throw std::overflow_error(
             "unable to add message to NotificationQueue: "
             "queue is full");
       }
       return false;
     }
     return true;
   }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::drainSignalsLocked ( )

inlineprivate

Definition at line 553 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::syncSignalAndQueue().

                             {
     ssize_t bytes_read = 0;
     if (eventfd_ > 0) {
       uint64_t message;
       bytes_read = readNoInt(eventfd_, &message, sizeof(message));
       CHECK(bytes_read != -1 || errno == EAGAIN);
     } else {
       // There should only be one byte in the pipe. To avoid potential leaks we
       // still drain.
       uint8_t message[32];
       ssize_t result;
       while ((result = readNoInt(pipeFds_[0], &message, sizeof(message))) !=
              -1) {
         bytes_read += result;
       }
       CHECK(result == -1 && errno == EAGAIN);
       LOG_IF(ERROR, bytes_read > 1)
           << "[NotificationQueue] Unexpected state while draining pipe: bytes_read="
           << bytes_read << " bytes, expected <= 1";
     }
     LOG_IF(ERROR, (signal_ && bytes_read == 0) || (!signal_ && bytes_read > 0))
         << "[NotificationQueue] Unexpected state while draining signals: signal_="
         << signal_ << " bytes_read=" << bytes_read;
 
     signal_ = false;
 
 #ifdef __ANDROID__
     if (bytes_read > 0) {
       eventBytes_ -= bytes_read;
     }
 #endif
   }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::ensureSignal ( ) const

inlineprivate

Definition at line 586 of file NotificationQueue.h.

                             {
     folly::SpinLockGuard g(spinlock_);
     ensureSignalLocked();
   }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::ensureSignalLocked ( ) const

inlineprivate

Definition at line 509 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::ensureSignal(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessageImpl(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessagesImpl(), and folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::syncSignalAndQueue().

                                   {
     // semantics: empty fd == empty queue <=> !signal_
     if (signal_) {
       return;
     }
 
     ssize_t bytes_written = 0;
     size_t bytes_expected = 0;
 
     do {
       if (eventfd_ >= 0) {
         // eventfd(2) dictates that we must write a 64-bit integer
         uint64_t signal = 1;
         bytes_expected = sizeof(signal);
         bytes_written = ::write(eventfd_, &signal, bytes_expected);
       } else {
         uint8_t signal = 1;
         bytes_expected = sizeof(signal);
         bytes_written = ::write(pipeFds_[1], &signal, bytes_expected);
       }
     } while (bytes_written == -1 && errno == EINTR);
 
 #ifdef __ANDROID__
     if (bytes_written > 0) {
       eventBytes_ += bytes_written;
       maxEventBytes_ = std::max((int)maxEventBytes_, (int)eventBytes_);
     }
 #endif
 
     if (bytes_written == ssize_t(bytes_expected)) {
       signal_ = true;
     } else {
 #ifdef __ANDROID__
       LOG(ERROR) << "NotificationQueue Write Error=" << errno
                  << " bytesInPipe=" << eventBytes_
                  << " maxInPipe=" << maxEventBytes_ << " queue=" << size();
 #endif
       folly::throwSystemError(
           "failed to signal NotificationQueue after "
           "write",
           errno);
     }
   }

template<typename MessageT>

NotificationQueue& folly::NotificationQueue< MessageT >::operator= ( NotificationQueue< MessageT > const & )

privatedelete

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::checkPid().

template<typename MessageT>

template<typename MessageTT >

void folly::NotificationQueue< MessageT >::putMessage ( MessageTT && message )

inline

Unconditionally put a message on the queue.

This method is like tryPutMessage(), but ignores the maximum queue size and always puts the message on the queue, even if the maximum queue size would be exceeded.

putMessage() may throw

std::bad_alloc if memory allocation fails, and may
std::runtime_error if the queue is currently draining
any other exception thrown by the MessageT move/copy constructor.

Definition at line 408 of file NotificationQueue.h.

Referenced by TEST().

                                        {
     putMessageImpl(std::forward<MessageTT>(message), 0);
   }

template<typename MessageT>

template<typename MessageTT >

bool folly::NotificationQueue< MessageT >::putMessageImpl	(	MessageTT &&	message,
		size_t	maxSize,
		bool	throws = `true`
	)

inlineprivate

Definition at line 602 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessage(), folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::tryPutMessage(), and folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::tryPutMessageNoThrow().

                                                                                {
     checkPid();
     bool signal = false;
     {
       auto data = std::make_unique<Node>(
           std::forward<MessageTT>(message), RequestContext::saveContext());
       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_.push_back(*data.release());
       if (signal) {
         ensureSignalLocked();
       }
     }
     return true;
   }

template<typename MessageT>

template<typename InputIteratorT >

void folly::NotificationQueue< MessageT >::putMessages	(	InputIteratorT	first,
		InputIteratorT	last
	)

inline

Put several messages on the queue.

Definition at line 416 of file NotificationQueue.h.

Referenced by TEST().

                                                               {
     typedef typename std::iterator_traits<InputIteratorT>::iterator_category
         IterCategory;
     putMessagesImpl(first, last, IterCategory());
   }

template<typename MessageT>

template<typename InputIteratorT >

void folly::NotificationQueue< MessageT >::putMessagesImpl	(	InputIteratorT	first,
		InputIteratorT	last,
		std::input_iterator_tag
	)

inlineprivate

Definition at line 626 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::putMessages().

                              {
     checkPid();
     bool signal = false;
     boost::intrusive::slist<Node, boost::intrusive::cache_last<true>> q;
     try {
       while (first != last) {
         auto data = std::make_unique<Node>(
             std::move(*first), RequestContext::saveContext());
         q.push_back(*data.release());
         ++first;
       }
       folly::SpinLockGuard g(spinlock_);
       checkDraining();
       queue_.splice(queue_.end(), q);
       if (numActiveConsumers_ < numConsumers_) {
         signal = true;
       }
       if (signal) {
         ensureSignalLocked();
       }
     } catch (...) {
       std::unique_ptr<Node> data;
       while (!q.empty()) {
         data.reset(&q.front());
         q.pop_front();
       }
       throw;
     }
   }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::setMaxQueueSize ( uint32_t max )

inline

Set the advisory maximum queue size.

This maximum queue size affects calls to tryPutMessage(). Message producers can still use the putMessage() call to unconditionally put a message on the queue, ignoring the configured maximum queue size. This can cause the queue size to exceed the configured maximum.

Definition at line 356 of file NotificationQueue.h.

                                      {
     advisoryMaxQueueSize_ = max;
   }

template<typename MessageT>

size_t folly::NotificationQueue< MessageT >::size ( ) const

inline

Definition at line 456 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::ensureSignalLocked().

                       {
     folly::SpinLockGuard g(spinlock_);
     return queue_.size();
   }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::syncSignalAndQueue ( )

inlineprivate

Definition at line 591 of file NotificationQueue.h.

Referenced by folly::NotificationQueue< folly::AsyncServerSocket::QueueMessage >::tryConsume().

                             {
     folly::SpinLockGuard g(spinlock_);
 
     if (queue_.empty()) {
       drainSignalsLocked();
     } else {
       ensureSignalLocked();
     }
   }

template<typename MessageT>

bool folly::NotificationQueue< MessageT >::tryConsume ( MessageT & result )

inline

Try to immediately pull a message off of the queue, without blocking.

If a message is immediately available, the result parameter will be updated to contain the message contents and true will be returned.

If no message is available, false will be returned and result will be left unmodified.

Definition at line 431 of file NotificationQueue.h.

                                     {
     SCOPE_EXIT {
       syncSignalAndQueue();
     };
 
     checkPid();
     std::unique_ptr<Node> data;
 
     {
       folly::SpinLockGuard g(spinlock_);
 
       if (UNLIKELY(queue_.empty())) {
         return false;
       }
 
       data.reset(&queue_.front());
       queue_.pop_front();
     }
 
     result = std::move(data->msg_);
     RequestContext::setContext(std::move(data->ctx_));
 
     return true;
   }

template<typename MessageT>

template<typename MessageTT >

void folly::NotificationQueue< MessageT >::tryPutMessage ( MessageTT && message )

inline

Attempt to put a message on the queue if the queue is not already full.

If the queue is full, a std::overflow_error will be thrown. The setMaxQueueSize() function controls the maximum queue size.

If the queue is currently draining, an std::runtime_error will be thrown.

This method may contend briefly on a spinlock if many threads are concurrently accessing the queue, but for all intents and purposes it will immediately place the message on the queue and return.

tryPutMessage() may throw std::bad_alloc if memory allocation fails, and may throw any other exception thrown by the MessageT move/copy constructor.

Definition at line 377 of file NotificationQueue.h.

Referenced by TEST().

                                           {
     putMessageImpl(std::forward<MessageTT>(message), advisoryMaxQueueSize_);
   }

template<typename MessageT>

template<typename MessageTT >

bool folly::NotificationQueue< MessageT >::tryPutMessageNoThrow ( MessageTT && message )

inline

No-throw versions of the above. Instead returns true on success, false on failure.

Only std::overflow_error (the common exception case) and std::runtime_error (which indicates that the queue is being drained) are prevented from being thrown. User code must still catch std::bad_alloc errors.

Definition at line 390 of file NotificationQueue.h.

Referenced by TEST().

                                                  {
     return putMessageImpl(
         std::forward<MessageTT>(message), advisoryMaxQueueSize_, false);
   }