folly::NotificationQueue< MessageT >::Consumer Class Reference

#include <NotificationQueue.h>

Inheritance diagram for folly::NotificationQueue< MessageT >::Consumer:

Public Types
enum	: uint16_t { kDefaultMaxReadAtOnce = 10 }

Public Member Functions
	Consumer ()
virtual void	messageAvailable (MessageT &&message) noexcept=0
void	startConsuming (EventBase *eventBase, NotificationQueue *queue)
void	startConsumingInternal (EventBase *eventBase, NotificationQueue *queue)
void	stopConsuming ()
bool	consumeUntilDrained (size_t *numConsumed=nullptr) noexcept
NotificationQueue *	getCurrentQueue () const
void	setMaxReadAtOnce (uint32_t maxAtOnce)
uint32_t	getMaxReadAtOnce () const
EventBase *	getEventBase ()
void	handlerReady (uint16_t events) noexceptoverride
template<typename TCallback >
std::unique_ptr< typename NotificationQueue< MessageT >::Consumer, DelayedDestruction::Destructor >	make (TCallback &&callback)
Public Member Functions inherited from folly::DelayedDestruction
bool	getDestroyPending () const
Public Member Functions inherited from folly::DelayedDestructionBase
virtual	~DelayedDestructionBase ()=default

Static Public Member Functions
template<typename TCallback >
static std::unique_ptr< Consumer, DelayedDestruction::Destructor >	make (TCallback &&callback)

Protected Member Functions
void	destroy () override
	~Consumer () override
Protected Member Functions inherited from folly::DelayedDestruction
	~DelayedDestruction () override=default
	DelayedDestruction ()
Protected Member Functions inherited from folly::DelayedDestructionBase
	DelayedDestructionBase ()
uint32_t	getDestructorGuardCount () const

Private Member Functions
void	consumeMessages (bool isDrain, size_t *numConsumed=nullptr) noexcept
void	setActive (bool active, bool shouldLock=false)
void	init (EventBase *eventBase, NotificationQueue *queue)
Private Member Functions inherited from folly::EventHandler
	EventHandler (EventBase *eventBase, int fd)
	EventHandler (EventBase *eventBase=nullptr, NetworkSocket fd=NetworkSocket())
virtual	~EventHandler ()
bool	registerHandler (uint16_t events)
void	unregisterHandler ()
bool	isHandlerRegistered () const
void	attachEventBase (EventBase *eventBase)
void	detachEventBase ()
void	changeHandlerFD (int fd)
void	changeHandlerFD (NetworkSocket fd)
void	initHandler (EventBase *eventBase, int fd)
void	initHandler (EventBase *eventBase, NetworkSocket fd)
uint16_t	getRegisteredEvents () const
bool	registerInternalHandler (uint16_t events)
bool	isPending () const

Private Attributes
NotificationQueue *	queue_
bool *	destroyedFlagPtr_
uint32_t	maxReadAtOnce_
EventBase *	base_
bool	active_ {false}

Additional Inherited Members
Private Types inherited from folly::EventHandler
enum	EventFlags {   NONE = 0, READ = EV_READ, WRITE = EV_WRITE, READ_WRITE = (READ | WRITE),   PERSIST = EV_PERSIST }

Detailed Description

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

A callback interface for consuming messages from the queue as they arrive.

Definition at line 83 of file NotificationQueue.h.

Member Enumeration Documentation

template<typename MessageT>

anonymous enum : uint16_t

Enumerator
kDefaultMaxReadAtOnce

Definition at line 85 of file NotificationQueue.h.

: uint16_t { kDefaultMaxReadAtOnce = 10 };

Constructor & Destructor Documentation

template<typename MessageT>

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

inline

Definition at line 87 of file NotificationQueue.h.

References folly::pushmi::__adl::noexcept().

        : queue_(nullptr),
          destroyedFlagPtr_(nullptr),
          maxReadAtOnce_(kDefaultMaxReadAtOnce) {}

template<typename MessageT>

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

inlineoverrideprotected

Definition at line 184 of file NotificationQueue.h.

References folly::pushmi::__adl::noexcept().

{}

Member Function Documentation

template<typename MessageT >

void folly::NotificationQueue< MessageT >::Consumer::consumeMessages ( bool  isDrain, size_t *  numConsumed = nullptr  )

privatenoexcept

Consume messages off the the queue until

• the queue is empty (1), or
• until the consumer is destroyed, or
• until the consumer is uninstalled, or
• an exception is thrown in the course of dequeueing, or
• unless isDrain is true, until the maxReadAtOnce_ limit is hit

(1) Well, maybe. See logic/comments around "wasEmpty" in implementation.

Definition at line 691 of file NotificationQueue.h.

References folly::data(), folly::gen::move, folly::NotificationQueue< MessageT >::queue_, SCOPE_EXIT, uint32_t, and UNLIKELY.

                                  {
  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;
    }
  };
  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 {
      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;
      }

      // Pull a message off the queue.
      std::unique_ptr<Node> data;
      data.reset(&queue_->queue_.front());
      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();
      RequestContextScopeGuard rctx(std::move(data->ctx_));

      locked = false;

      // Call the callback
      bool callbackDestroyed = false;
      CHECK(destroyedFlagPtr_ == nullptr);
      destroyedFlagPtr_ = &callbackDestroyed;
      messageAvailable(std::move(data->msg_));
      destroyedFlagPtr_ = nullptr;

      // If the callback was destroyed before it returned, we are done
      if (callbackDestroyed) {
        return;
      }

      // If the callback is no longer installed, we are done.
      if (queue_ == nullptr) {
        return;
      }

      // If we have hit maxReadAtOnce_, we are done.
      ++numProcessed;
      if (!isDrain && maxReadAtOnce_ > 0 && numProcessed >= maxReadAtOnce_) {
        return;
      }

      // If the queue was empty before we invoked the callback, it's probable
      // that it is still empty now.  Just go ahead and return, rather than
      // 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.)
      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
      // declared as noexcept and should never throw.
      //
      // If the MessageT constructor does throw we try to handle it as best as
      // we can, but we can't work miracles.  We will just ignore the error for
      // now and return.  The next time around the event loop we will end up
      // 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;
    }
  }
}

template<typename MessageT >

bool folly::NotificationQueue< MessageT >::Consumer::consumeUntilDrained ( size_t *  numConsumed = nullptr )

noexcept

Consume messages off the queue until it is empty. No messages may be added to the queue while it is draining, so that the process is bounded. To that end, putMessage/tryPutMessage will throw an std::runtime_error, and tryPutMessageNoThrow will return false.

Returns

true if the queue was drained, false otherwise. In practice, this will only fail if someone else is already draining the queue.

Definition at line 848 of file NotificationQueue.h.

References g(), and folly::NotificationQueue< MessageT >::queue_.

Referenced by TEST().

                                  {
  DestructorGuard dg(this);
  {
    folly::SpinLockGuard g(queue_->spinlock_);
    if (queue_->draining_) {
      return false;
    }
    queue_->draining_ = true;
  }
  consumeMessages(true, numConsumed);
  {
    folly::SpinLockGuard g(queue_->spinlock_);
    queue_->draining_ = false;
  }
  return true;
}

template<typename MessageT >

void folly::NotificationQueue< MessageT >::Consumer::destroy ( )

overrideprotectedvirtual

destroy() requests destruction of the object.

This method will destroy the object after it has no more functions running higher up on the stack. (i.e., No more DestructorGuard objects exist for this object.) This method must be used instead of the destructor.

Reimplemented from folly::DelayedDestruction.

Definition at line 672 of file NotificationQueue.h.

References folly::DelayedDestruction::destroy().

                                                  {
  // If we are in the middle of a call to handlerReady(), destroyedFlagPtr_
  // will be non-nullptr.  Mark the value that it points to, so that
  // handlerReady() will know the callback is destroyed, and that it cannot
  // access any member variables anymore.
  if (destroyedFlagPtr_) {
    *destroyedFlagPtr_ = true;
  }
  stopConsuming();
  DelayedDestruction::destroy();
}

template<typename MessageT>

NotificationQueue* folly::NotificationQueue< MessageT >::Consumer::getCurrentQueue ( ) const

inline

Get the NotificationQueue that this consumer is currently consuming messages from. Returns nullptr if the consumer is not currently consuming events from any queue.

Definition at line 154 of file NotificationQueue.h.

References folly::NotificationQueue< MessageT >::queue_.

                                               {
      return queue_;
    }

template<typename MessageT>

EventBase* folly::NotificationQueue< MessageT >::Consumer::getEventBase ( )

inline

Definition at line 175 of file NotificationQueue.h.

References base_, destroy(), and uint16_t.

Referenced by folly::EventBase::FunctionRunner::messageAvailable().

                              {
      return base_;
    }

template<typename MessageT>

uint32_t folly::NotificationQueue< MessageT >::Consumer::getMaxReadAtOnce ( ) const

inline

Definition at line 171 of file NotificationQueue.h.

                                      {
      return maxReadAtOnce_;
    }

template<typename MessageT >

void folly::NotificationQueue< MessageT >::Consumer::handlerReady ( uint16_t  events )

overridevirtualnoexcept

handlerReady() is invoked when the handler is ready.

Parameters

events

A bitset indicating the events that are ready.

Implements folly::EventHandler.

Definition at line 685 of file NotificationQueue.h.

                         {
  consumeMessages(false);
}

template<typename MessageT >

void folly::NotificationQueue< MessageT >::Consumer::init ( EventBase *  eventBase, NotificationQueue *  queue  )

private

Definition at line 803 of file NotificationQueue.h.

References base_, folly::EventBase::dcheckIsInEventBaseThread(), g(), and folly::NotificationQueue< MessageT >::queue_.

                              {
  eventBase->dcheckIsInEventBaseThread();
  assert(queue_ == nullptr);
  assert(!isHandlerRegistered());
  queue->checkPid();

  base_ = eventBase;

  queue_ = queue;

  {
    folly::SpinLockGuard g(queue_->spinlock_);
    queue_->numConsumers_++;
  }
  queue_->ensureSignal();

  if (queue_->eventfd_ >= 0) {
    initHandler(eventBase, queue_->eventfd_);
  } else {
    initHandler(eventBase, queue_->pipeFds_[0]);
  }
}

template<typename MessageT>

template<typename TCallback >

static std::unique_ptr<Consumer, DelayedDestruction::Destructor> folly::NotificationQueue< MessageT >::Consumer::make ( TCallback &&  callback )

static

template<typename MessageT>

template<typename TCallback >

std::unique_ptr< typename NotificationQueue<MessageT>::Consumer, DelayedDestruction::Destructor> folly::NotificationQueue< MessageT >::Consumer::make

(

TCallback &&

callback

)

Definition at line 931 of file NotificationQueue.h.

References type.

                                                              {
  return std::unique_ptr<
      NotificationQueue<MessageT>::Consumer,
      DelayedDestruction::Destructor>(
      new detail::notification_queue_consumer_wrapper<
          MessageT,
          typename std::decay<TCallback>::type>(
          std::forward<TCallback>(callback)));
}

template<typename MessageT>

virtual void folly::NotificationQueue< MessageT >::Consumer::messageAvailable ( MessageT &&  message )

pure virtualnoexcept

messageAvailable() will be invoked whenever a new message is available from the pipe.

Implemented in folly::detail::notification_queue_consumer_wrapper< MessageT, TCallback >.

template<typename MessageT>

void folly::NotificationQueue< MessageT >::Consumer::setActive ( bool  active, bool  shouldLock = false  )

inlineprivate

Definition at line 199 of file NotificationQueue.h.

References folly::init(), and folly::NotificationQueue< MessageT >::queue_.

                                                         {
      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();
      }
    }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::Consumer::setMaxReadAtOnce ( uint32_t  maxAtOnce )

inline

Set a limit on how many messages this consumer will read each iteration around the event loop.

This helps rate-limit how much work the Consumer will do each event loop iteration, to prevent it from starving other event handlers.

A limit of 0 means no limit will be enforced. If unset, the limit defaults to kDefaultMaxReadAtOnce (defined to 10 above).

Definition at line 168 of file NotificationQueue.h.

Referenced by QueueTest::maxReadAtOnce(), folly::AsyncServerSocket::RemoteAcceptor::start(), and TEST().

                                              {
      maxReadAtOnce_ = maxAtOnce;
    }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::Consumer::startConsuming ( EventBase *  eventBase, NotificationQueue *  queue  )

inline

Begin consuming messages from the specified queue.

messageAvailable() will be called whenever a message is available. This consumer will continue to consume messages until stopConsuming() is called.

A Consumer may only consume messages from a single NotificationQueue at a time. startConsuming() should not be called if this consumer is already consuming.

Definition at line 114 of file NotificationQueue.h.

References folly::init().

Referenced by QueueTest::maxReadAtOnce(), QueueTest::multiConsumer(), QueueTest::putMessages(), QueueTest::sendOne(), and TEST().

                                                                        {
      init(eventBase, queue);
      registerHandler(READ | PERSIST);
    }

template<typename MessageT>

void folly::NotificationQueue< MessageT >::Consumer::startConsumingInternal ( EventBase *  eventBase, NotificationQueue *  queue  )

inline

Same as above but registers this event handler as internal so that it doesn't count towards the pending reader count for the IOLoop.

Definition at line 123 of file NotificationQueue.h.

References folly::init(), and folly::pushmi::__adl::noexcept().

                                  {
      init(eventBase, queue);
      registerInternalHandler(READ | PERSIST);
    }

template<typename MessageT >

void folly::NotificationQueue< MessageT >::Consumer::stopConsuming

(

)

Stop consuming messages.

startConsuming() may be called again to resume consumption of messages at a later point in time.

Definition at line 829 of file NotificationQueue.h.

References g(), and folly::NotificationQueue< MessageT >::queue_.

Referenced by QueueTest::multiConsumer(), QueueTest::putMessages(), and QueueTest::sendOne().

                                                        {
  if (queue_ == nullptr) {
    assert(!isHandlerRegistered());
    return;
  }

  {
    folly::SpinLockGuard g(queue_->spinlock_);
    queue_->numConsumers_--;
    setActive(false);
  }

  assert(isHandlerRegistered());
  unregisterHandler();
  detachEventBase();
  queue_ = nullptr;
}

Member Data Documentation

template<typename MessageT>

bool folly::NotificationQueue< MessageT >::Consumer::active_ {false}

private

Definition at line 223 of file NotificationQueue.h.

template<typename MessageT>

EventBase* folly::NotificationQueue< MessageT >::Consumer::base_

private

Definition at line 222 of file NotificationQueue.h.

template<typename MessageT>

bool* folly::NotificationQueue< MessageT >::Consumer::destroyedFlagPtr_

private

Definition at line 220 of file NotificationQueue.h.

template<typename MessageT>

uint32_t folly::NotificationQueue< MessageT >::Consumer::maxReadAtOnce_

private

Definition at line 221 of file NotificationQueue.h.

template<typename MessageT>

NotificationQueue* folly::NotificationQueue< MessageT >::Consumer::queue_

private

Definition at line 219 of file NotificationQueue.h.

---

The documentation for this class was generated from the following file:

• proxygen/folly/folly/io/async/NotificationQueue.h