folly/io/async/NotificationQueue.h

   1 /*
   2  * Copyright 2016 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #pragma once
  18
  19 #include <fcntl.h>
  20 #include <sys/types.h>
  21 #include <unistd.h>
  22
  23 #include <algorithm>
  24 #include <deque>
  25 #include <iterator>
  26 #include <memory>
  27 #include <stdexcept>
  28 #include <utility>
  29
  30 #include <folly/FileUtil.h>
  31 #include <folly/io/async/EventBase.h>
  32 #include <folly/io/async/EventHandler.h>
  33 #include <folly/io/async/DelayedDestruction.h>
  34 #include <folly/io/async/Request.h>
  35 #include <folly/Likely.h>
  36 #include <folly/ScopeGuard.h>
  37 #include <folly/SpinLock.h>
  38
  39 #include <glog/logging.h>
  40
  41 #if __linux__ && !__ANDROID__
  42 #define FOLLY_HAVE_EVENTFD
  43 #include <folly/io/async/EventFDWrapper.h>
  44 #endif
  45
  46 namespace folly {
  47
  48 /**
  49  * A producer-consumer queue for passing messages between EventBase threads.
  50  *
  51  * Messages can be added to the queue from any thread.  Multiple consumers may
  52  * listen to the queue from multiple EventBase threads.
  53  *
  54  * A NotificationQueue may not be destroyed while there are still consumers
  55  * registered to receive events from the queue.  It is the user's
  56  * responsibility to ensure that all consumers are unregistered before the
  57  * queue is destroyed.
  58  *
  59  * MessageT should be MoveConstructible (i.e., must support either a move
  60  * constructor or a copy constructor, or both).  Ideally it's move constructor
  61  * (or copy constructor if no move constructor is provided) should never throw
  62  * exceptions.  If the constructor may throw, the consumers could end up
  63  * spinning trying to move a message off the queue and failing, and then
  64  * retrying.
  65  */
  66 template<typename MessageT>
  67 class NotificationQueue {
  68  public:
  69   /**
  70    * A callback interface for consuming messages from the queue as they arrive.
  71    */
  72   class Consumer : public DelayedDestruction, private EventHandler {
  73    public:
  74     enum : uint16_t { kDefaultMaxReadAtOnce = 10 };
  75
  76     Consumer()
  77       : queue_(nullptr),
  78         destroyedFlagPtr_(nullptr),
  79         maxReadAtOnce_(kDefaultMaxReadAtOnce) {}
  80
  81     // create a consumer in-place, without the need to build new class
  82     template <typename TCallback>
  83     static std::unique_ptr<Consumer, DelayedDestruction::Destructor> make(
  84         TCallback&& callback);
  85
  86     /**
  87      * messageAvailable() will be invoked whenever a new
  88      * message is available from the pipe.
  89      */
  90     virtual void messageAvailable(MessageT&& message) = 0;
  91
  92     /**
  93      * Begin consuming messages from the specified queue.
  94      *
  95      * messageAvailable() will be called whenever a message is available.  This
  96      * consumer will continue to consume messages until stopConsuming() is
  97      * called.
  98      *
  99      * A Consumer may only consume messages from a single NotificationQueue at
 100      * a time.  startConsuming() should not be called if this consumer is
 101      * already consuming.
 102      */
 103     void startConsuming(EventBase* eventBase, NotificationQueue* queue) {
 104       init(eventBase, queue);
 105       registerHandler(READ | PERSIST);
 106     }
 107
 108     /**
 109      * Same as above but registers this event handler as internal so that it
 110      * doesn't count towards the pending reader count for the IOLoop.
 111      */
 112     void startConsumingInternal(
 113         EventBase* eventBase, NotificationQueue* queue) {
 114       init(eventBase, queue);
 115       registerInternalHandler(READ | PERSIST);
 116     }
 117
 118     /**
 119      * Stop consuming messages.
 120      *
 121      * startConsuming() may be called again to resume consumption of messages
 122      * at a later point in time.
 123      */
 124     void stopConsuming();
 125
 126     /**
 127      * Consume messages off the queue until it is empty. No messages may be
 128      * added to the queue while it is draining, so that the process is bounded.
 129      * To that end, putMessage/tryPutMessage will throw an std::runtime_error,
 130      * and tryPutMessageNoThrow will return false.
 131      *
 132      * @returns true if the queue was drained, false otherwise. In practice,
 133      * this will only fail if someone else is already draining the queue.
 134      */
 135     bool consumeUntilDrained(size_t* numConsumed = nullptr) noexcept;
 136
 137     /**
 138      * Get the NotificationQueue that this consumer is currently consuming
 139      * messages from.  Returns nullptr if the consumer is not currently
 140      * consuming events from any queue.
 141      */
 142     NotificationQueue* getCurrentQueue() const {
 143       return queue_;
 144     }
 145
 146     /**
 147      * Set a limit on how many messages this consumer will read each iteration
 148      * around the event loop.
 149      *
 150      * This helps rate-limit how much work the Consumer will do each event loop
 151      * iteration, to prevent it from starving other event handlers.
 152      *
 153      * A limit of 0 means no limit will be enforced.  If unset, the limit
 154      * defaults to kDefaultMaxReadAtOnce (defined to 10 above).
 155      */
 156     void setMaxReadAtOnce(uint32_t maxAtOnce) {
 157       maxReadAtOnce_ = maxAtOnce;
 158     }
 159     uint32_t getMaxReadAtOnce() const {
 160       return maxReadAtOnce_;
 161     }
 162
 163     EventBase* getEventBase() {
 164       return base_;
 165     }
 166
 167     void handlerReady(uint16_t events) noexcept override;
 168
 169    protected:
 170
 171     void destroy() override;
 172
 173     virtual ~Consumer() {}
 174
 175    private:
 176     /**
 177      * Consume messages off the the queue until
 178      *   - the queue is empty (1), or
 179      *   - until the consumer is destroyed, or
 180      *   - until the consumer is uninstalled, or
 181      *   - an exception is thrown in the course of dequeueing, or
 182      *   - unless isDrain is true, until the maxReadAtOnce_ limit is hit
 183      *
 184      * (1) Well, maybe. See logic/comments around "wasEmpty" in implementation.
 185      */
 186     void consumeMessages(bool isDrain, size_t* numConsumed = nullptr) noexcept;
 187
 188     void setActive(bool active, bool shouldLock = false) {
 189       if (!queue_) {
 190         active_ = active;
 191         return;
 192       }
 193       if (shouldLock) {
 194         queue_->spinlock_.lock();
 195       }
 196       if (!active_ && active) {
 197         ++queue_->numActiveConsumers_;
 198       } else if (active_ && !active) {
 199         --queue_->numActiveConsumers_;
 200       }
 201       active_ = active;
 202       if (shouldLock) {
 203         queue_->spinlock_.unlock();
 204       }
 205     }
 206     void init(EventBase* eventBase, NotificationQueue* queue);
 207
 208     NotificationQueue* queue_;
 209     bool* destroyedFlagPtr_;
 210     uint32_t maxReadAtOnce_;
 211     EventBase* base_;
 212     bool active_{false};
 213   };
 214
 215   enum class FdType {
 216     PIPE,
 217 #ifdef FOLLY_HAVE_EVENTFD
 218     EVENTFD,
 219 #endif
 220   };
 221
 222   /**
 223    * Create a new NotificationQueue.
 224    *
 225    * If the maxSize parameter is specified, this sets the maximum queue size
 226    * that will be enforced by tryPutMessage().  (This size is advisory, and may
 227    * be exceeded if producers explicitly use putMessage() instead of
 228    * tryPutMessage().)
 229    *
 230    * The fdType parameter determines the type of file descriptor used
 231    * internally to signal message availability.  The default (eventfd) is
 232    * preferable for performance and because it won't fail when the queue gets
 233    * too long.  It is not available on on older and non-linux kernels, however.
 234    * In this case the code will fall back to using a pipe, the parameter is
 235    * mostly for testing purposes.
 236    */
 237   explicit NotificationQueue(uint32_t maxSize = 0,
 238 #ifdef FOLLY_HAVE_EVENTFD
 239                              FdType fdType = FdType::EVENTFD)
 240 #else
 241                              FdType fdType = FdType::PIPE)
 242 #endif
 243     : eventfd_(-1),
 244       pipeFds_{-1, -1},
 245       advisoryMaxQueueSize_(maxSize),
 246       pid_(getpid()),
 247       queue_() {
 248
 249     RequestContext::saveContext();
 250
 251 #ifdef FOLLY_HAVE_EVENTFD
 252     if (fdType == FdType::EVENTFD) {
 253       eventfd_ = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK | EFD_SEMAPHORE);
 254       if (eventfd_ == -1) {
 255         if (errno == ENOSYS || errno == EINVAL) {
 256           // eventfd not availalble
 257           LOG(ERROR) << "failed to create eventfd for NotificationQueue: "
 258                      << errno << ", falling back to pipe mode (is your kernel "
 259                      << "> 2.6.30?)";
 260           fdType = FdType::PIPE;
 261         } else {
 262           // some other error
 263           folly::throwSystemError("Failed to create eventfd for "
 264                                   "NotificationQueue", errno);
 265         }
 266       }
 267     }
 268 #endif
 269     if (fdType == FdType::PIPE) {
 270       if (pipe(pipeFds_)) {
 271         folly::throwSystemError("Failed to create pipe for NotificationQueue",
 272                                 errno);
 273       }
 274       try {
 275         // put both ends of the pipe into non-blocking mode
 276         if (fcntl(pipeFds_[0], F_SETFL, O_RDONLY | O_NONBLOCK) != 0) {
 277           folly::throwSystemError("failed to put NotificationQueue pipe read "
 278                                   "endpoint into non-blocking mode", errno);
 279         }
 280         if (fcntl(pipeFds_[1], F_SETFL, O_WRONLY | O_NONBLOCK) != 0) {
 281           folly::throwSystemError("failed to put NotificationQueue pipe write "
 282                                   "endpoint into non-blocking mode", errno);
 283         }
 284       } catch (...) {
 285         ::close(pipeFds_[0]);
 286         ::close(pipeFds_[1]);
 287         throw;
 288       }
 289     }
 290   }
 291
 292   ~NotificationQueue() {
 293     if (eventfd_ >= 0) {
 294       ::close(eventfd_);
 295       eventfd_ = -1;
 296     }
 297     if (pipeFds_[0] >= 0) {
 298       ::close(pipeFds_[0]);
 299       pipeFds_[0] = -1;
 300     }
 301     if (pipeFds_[1] >= 0) {
 302       ::close(pipeFds_[1]);
 303       pipeFds_[1] = -1;
 304     }
 305   }
 306
 307   /**
 308    * Set the advisory maximum queue size.
 309    *
 310    * This maximum queue size affects calls to tryPutMessage().  Message
 311    * producers can still use the putMessage() call to unconditionally put a
 312    * message on the queue, ignoring the configured maximum queue size.  This
 313    * can cause the queue size to exceed the configured maximum.
 314    */
 315   void setMaxQueueSize(uint32_t max) {
 316     advisoryMaxQueueSize_ = max;
 317   }
 318
 319   /**
 320    * Attempt to put a message on the queue if the queue is not already full.
 321    *
 322    * If the queue is full, a std::overflow_error will be thrown.  The
 323    * setMaxQueueSize() function controls the maximum queue size.
 324    *
 325    * If the queue is currently draining, an std::runtime_error will be thrown.
 326    *
 327    * This method may contend briefly on a spinlock if many threads are
 328    * concurrently accessing the queue, but for all intents and purposes it will
 329    * immediately place the message on the queue and return.
 330    *
 331    * tryPutMessage() may throw std::bad_alloc if memory allocation fails, and
 332    * may throw any other exception thrown by the MessageT move/copy
 333    * constructor.
 334    */
 335   void tryPutMessage(MessageT&& message) {
 336     putMessageImpl(std::move(message), advisoryMaxQueueSize_);
 337   }
 338   void tryPutMessage(const MessageT& message) {
 339     putMessageImpl(message, advisoryMaxQueueSize_);
 340   }
 341
 342   /**
 343    * No-throw versions of the above.  Instead returns true on success, false on
 344    * failure.
 345    *
 346    * Only std::overflow_error (the common exception case) and std::runtime_error
 347    * (which indicates that the queue is being drained) are prevented from being
 348    * thrown. User code must still catch std::bad_alloc errors.
 349    */
 350   bool tryPutMessageNoThrow(MessageT&& message) {
 351     return putMessageImpl(std::move(message), advisoryMaxQueueSize_, false);
 352   }
 353   bool tryPutMessageNoThrow(const MessageT& message) {
 354     return putMessageImpl(message, advisoryMaxQueueSize_, false);
 355   }
 356
 357   /**
 358    * Unconditionally put a message on the queue.
 359    *
 360    * This method is like tryPutMessage(), but ignores the maximum queue size
 361    * and always puts the message on the queue, even if the maximum queue size
 362    * would be exceeded.
 363    *
 364    * putMessage() may throw
 365    *   - std::bad_alloc if memory allocation fails, and may
 366    *   - std::runtime_error if the queue is currently draining
 367    *   - any other exception thrown by the MessageT move/copy constructor.
 368    */
 369   void putMessage(MessageT&& message) {
 370     putMessageImpl(std::move(message), 0);
 371   }
 372   void putMessage(const MessageT& message) {
 373     putMessageImpl(message, 0);
 374   }
 375
 376   /**
 377    * Put several messages on the queue.
 378    */
 379   template<typename InputIteratorT>
 380   void putMessages(InputIteratorT first, InputIteratorT last) {
 381     typedef typename std::iterator_traits<InputIteratorT>::iterator_category
 382       IterCategory;
 383     putMessagesImpl(first, last, IterCategory());
 384   }
 385
 386   /**
 387    * Try to immediately pull a message off of the queue, without blocking.
 388    *
 389    * If a message is immediately available, the result parameter will be
 390    * updated to contain the message contents and true will be returned.
 391    *
 392    * If no message is available, false will be returned and result will be left
 393    * unmodified.
 394    */
 395   bool tryConsume(MessageT& result) {
 396     checkPid();
 397
 398     try {
 399
 400       folly::SpinLockGuard g(spinlock_);
 401
 402       if (UNLIKELY(queue_.empty())) {
 403         return false;
 404       }
 405
 406       auto data = std::move(queue_.front());
 407       result = data.first;
 408       RequestContext::setContext(data.second);
 409
 410       queue_.pop_front();
 411     } catch (...) {
 412       // Handle an exception if the assignment operator happens to throw.
 413       // We consumed an event but weren't able to pop the message off the
 414       // queue.  Signal the event again since the message is still in the
 415       // queue.
 416       signalEvent(1);
 417       throw;
 418     }
 419
 420     return true;
 421   }
 422
 423   size_t size() {
 424     folly::SpinLockGuard g(spinlock_);
 425     return queue_.size();
 426   }
 427
 428   /**
 429    * Check that the NotificationQueue is being used from the correct process.
 430    *
 431    * If you create a NotificationQueue in one process, then fork, and try to
 432    * send messages to the queue from the child process, you're going to have a
 433    * bad time.  Unfortunately users have (accidentally) run into this.
 434    *
 435    * Because we use an eventfd/pipe, the child process can actually signal the
 436    * parent process that an event is ready.  However, it can't put anything on
 437    * the parent's queue, so the parent wakes up and finds an empty queue.  This
 438    * check ensures that we catch the problem in the misbehaving child process
 439    * code, and crash before signalling the parent process.
 440    */
 441   void checkPid() const {
 442     CHECK_EQ(pid_, getpid());
 443   }
 444
 445  private:
 446   // Forbidden copy constructor and assignment operator
 447   NotificationQueue(NotificationQueue const &) = delete;
 448   NotificationQueue& operator=(NotificationQueue const &) = delete;
 449
 450   inline bool checkQueueSize(size_t maxSize, bool throws=true) const {
 451     DCHECK(0 == spinlock_.trylock());
 452     if (maxSize > 0 && queue_.size() >= maxSize) {
 453       if (throws) {
 454         throw std::overflow_error("unable to add message to NotificationQueue: "
 455                                   "queue is full");
 456       }
 457       return false;
 458     }
 459     return true;
 460   }
 461
 462   inline bool checkDraining(bool throws=true) {
 463     if (UNLIKELY(draining_ && throws)) {
 464       throw std::runtime_error("queue is draining, cannot add message");
 465     }
 466     return draining_;
 467   }
 468
 469   inline void signalEvent(size_t numAdded = 1) const {
 470     static const uint8_t kPipeMessage[] = {
 471       1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1
 472     };
 473
 474     ssize_t bytes_written = 0;
 475     ssize_t bytes_expected = 0;
 476     if (eventfd_ >= 0) {
 477       // eventfd(2) dictates that we must write a 64-bit integer
 478       uint64_t numAdded64(numAdded);
 479       bytes_expected = static_cast<ssize_t>(sizeof(numAdded64));
 480       bytes_written = ::write(eventfd_, &numAdded64, sizeof(numAdded64));
 481     } else {
 482       // pipe semantics, add one message for each numAdded
 483       bytes_expected = numAdded;
 484       do {
 485         size_t messageSize = std::min(numAdded, sizeof(kPipeMessage));
 486         ssize_t rc = ::write(pipeFds_[1], kPipeMessage, messageSize);
 487         if (rc < 0) {
 488           // TODO: if the pipe is full, write will fail with EAGAIN.
 489           // See task #1044651 for how this could be handled
 490           break;
 491         }
 492         numAdded -= rc;
 493         bytes_written += rc;
 494       } while (numAdded > 0);
 495     }
 496     if (bytes_written != bytes_expected) {
 497       folly::throwSystemError("failed to signal NotificationQueue after "
 498                               "write", errno);
 499     }
 500   }
 501
 502   bool tryConsumeEvent() {
 503     uint64_t value = 0;
 504     ssize_t rc = -1;
 505     if (eventfd_ >= 0) {
 506       rc = readNoInt(eventfd_, &value, sizeof(value));
 507     } else {
 508       uint8_t value8;
 509       rc = readNoInt(pipeFds_[0], &value8, sizeof(value8));
 510       value = value8;
 511     }
 512     if (rc < 0) {
 513       // EAGAIN should pretty much be the only error we can ever get.
 514       // This means someone else already processed the only available message.
 515       CHECK_EQ(errno, EAGAIN);
 516       return false;
 517     }
 518     assert(value == 1);
 519     return true;
 520   }
 521
 522   bool putMessageImpl(MessageT&& message, size_t maxSize, bool throws=true) {
 523     checkPid();
 524     bool signal = false;
 525     {
 526       folly::SpinLockGuard g(spinlock_);
 527       if (checkDraining(throws) || !checkQueueSize(maxSize, throws)) {
 528         return false;
 529       }
 530       // We only need to signal an event if not all consumers are
 531       // awake.
 532       if (numActiveConsumers_ < numConsumers_) {
 533         signal = true;
 534       }
 535       queue_.emplace_back(std::move(message), RequestContext::saveContext());
 536     }
 537     if (signal) {
 538       signalEvent();
 539     }
 540     return true;
 541   }
 542
 543   bool putMessageImpl(
 544     const MessageT& message, size_t maxSize, bool throws=true) {
 545     checkPid();
 546     bool signal = false;
 547     {
 548       folly::SpinLockGuard g(spinlock_);
 549       if (checkDraining(throws) || !checkQueueSize(maxSize, throws)) {
 550         return false;
 551       }
 552       if (numActiveConsumers_ < numConsumers_) {
 553         signal = true;
 554       }
 555       queue_.emplace_back(message, RequestContext::saveContext());
 556     }
 557     if (signal) {
 558       signalEvent();
 559     }
 560     return true;
 561   }
 562
 563   template<typename InputIteratorT>
 564   void putMessagesImpl(InputIteratorT first, InputIteratorT last,
 565                        std::input_iterator_tag) {
 566     checkPid();
 567     bool signal = false;
 568     size_t numAdded = 0;
 569     {
 570       folly::SpinLockGuard g(spinlock_);
 571       checkDraining();
 572       while (first != last) {
 573         queue_.emplace_back(*first, RequestContext::saveContext());
 574         ++first;
 575         ++numAdded;
 576       }
 577       if (numActiveConsumers_ < numConsumers_) {
 578         signal = true;
 579       }
 580     }
 581     if (signal) {
 582       signalEvent();
 583     }
 584   }
 585
 586   mutable folly::SpinLock spinlock_;
 587   int eventfd_;
 588   int pipeFds_[2]; // to fallback to on older/non-linux systems
 589   uint32_t advisoryMaxQueueSize_;
 590   pid_t pid_;
 591   std::deque<std::pair<MessageT, std::shared_ptr<RequestContext>>> queue_;
 592   int numConsumers_{0};
 593   std::atomic<int> numActiveConsumers_{0};
 594   bool draining_{false};
 595 };
 596
 597 template<typename MessageT>
 598 void NotificationQueue<MessageT>::Consumer::destroy() {
 599   // If we are in the middle of a call to handlerReady(), destroyedFlagPtr_
 600   // will be non-nullptr.  Mark the value that it points to, so that
 601   // handlerReady() will know the callback is destroyed, and that it cannot
 602   // access any member variables anymore.
 603   if (destroyedFlagPtr_) {
 604     *destroyedFlagPtr_ = true;
 605   }
 606   stopConsuming();
 607   DelayedDestruction::destroy();
 608 }
 609
 610 template<typename MessageT>
 611 void NotificationQueue<MessageT>::Consumer::handlerReady(uint16_t /*events*/)
 612     noexcept {
 613   consumeMessages(false);
 614 }
 615
 616 template<typename MessageT>
 617 void NotificationQueue<MessageT>::Consumer::consumeMessages(
 618     bool isDrain, size_t* numConsumed) noexcept {
 619   DestructorGuard dg(this);
 620   uint32_t numProcessed = 0;
 621   bool firstRun = true;
 622   setActive(true);
 623   SCOPE_EXIT { setActive(false, /* shouldLock = */ true); };
 624   SCOPE_EXIT {
 625     if (numConsumed != nullptr) {
 626       *numConsumed = numProcessed;
 627     }
 628   };
 629   while (true) {
 630     // Try to decrement the eventfd.
 631     //
 632     // The eventfd is only used to wake up the consumer - there may or
 633     // may not actually be an event available (another consumer may
 634     // have read it).  We don't really care, we only care about
 635     // emptying the queue.
 636     if (!isDrain && firstRun) {
 637       queue_->tryConsumeEvent();
 638       firstRun = false;
 639     }
 640
 641     // Now pop the message off of the queue.
 642     //
 643     // We have to manually acquire and release the spinlock here, rather than
 644     // using SpinLockHolder since the MessageT has to be constructed while
 645     // holding the spinlock and available after we release it.  SpinLockHolder
 646     // unfortunately doesn't provide a release() method.  (We can't construct
 647     // MessageT first since we have no guarantee that MessageT has a default
 648     // constructor.
 649     queue_->spinlock_.lock();
 650     bool locked = true;
 651
 652     try {
 653       if (UNLIKELY(queue_->queue_.empty())) {
 654         // If there is no message, we've reached the end of the queue, return.
 655         setActive(false);
 656         queue_->spinlock_.unlock();
 657         return;
 658       }
 659
 660       // Pull a message off the queue.
 661       auto& data = queue_->queue_.front();
 662
 663       MessageT msg(std::move(data.first));
 664       auto old_ctx =
 665         RequestContext::setContext(data.second);
 666       queue_->queue_.pop_front();
 667
 668       // Check to see if the queue is empty now.
 669       // We use this as an optimization to see if we should bother trying to
 670       // loop again and read another message after invoking this callback.
 671       bool wasEmpty = queue_->queue_.empty();
 672       if (wasEmpty) {
 673         setActive(false);
 674       }
 675
 676       // Now unlock the spinlock before we invoke the callback.
 677       queue_->spinlock_.unlock();
 678       locked = false;
 679
 680       // Call the callback
 681       bool callbackDestroyed = false;
 682       CHECK(destroyedFlagPtr_ == nullptr);
 683       destroyedFlagPtr_ = &callbackDestroyed;
 684       messageAvailable(std::move(msg));
 685       destroyedFlagPtr_ = nullptr;
 686
 687       RequestContext::setContext(old_ctx);
 688
 689       // If the callback was destroyed before it returned, we are done
 690       if (callbackDestroyed) {
 691         return;
 692       }
 693
 694       // If the callback is no longer installed, we are done.
 695       if (queue_ == nullptr) {
 696         return;
 697       }
 698
 699       // If we have hit maxReadAtOnce_, we are done.
 700       ++numProcessed;
 701       if (!isDrain && maxReadAtOnce_ > 0 &&
 702           numProcessed >= maxReadAtOnce_) {
 703         queue_->signalEvent(1);
 704         return;
 705       }
 706
 707       // If the queue was empty before we invoked the callback, it's probable
 708       // that it is still empty now.  Just go ahead and return, rather than
 709       // looping again and trying to re-read from the eventfd.  (If a new
 710       // message had in fact arrived while we were invoking the callback, we
 711       // will simply be woken up the next time around the event loop and will
 712       // process the message then.)
 713       if (wasEmpty) {
 714         return;
 715       }
 716     } catch (const std::exception& ex) {
 717       // This catch block is really just to handle the case where the MessageT
 718       // constructor throws.  The messageAvailable() callback itself is
 719       // declared as noexcept and should never throw.
 720       //
 721       // If the MessageT constructor does throw we try to handle it as best as
 722       // we can, but we can't work miracles.  We will just ignore the error for
 723       // now and return.  The next time around the event loop we will end up
 724       // trying to read the message again.  If MessageT continues to throw we
 725       // will never make forward progress and will keep trying each time around
 726       // the event loop.
 727       if (locked) {
 728         // Unlock the spinlock.
 729         queue_->spinlock_.unlock();
 730
 731         // Push a notification back on the eventfd since we didn't actually
 732         // read the message off of the queue.
 733         if (!isDrain) {
 734           queue_->signalEvent(1);
 735         }
 736       }
 737
 738       return;
 739     }
 740   }
 741 }
 742
 743 template<typename MessageT>
 744 void NotificationQueue<MessageT>::Consumer::init(
 745     EventBase* eventBase,
 746     NotificationQueue* queue) {
 747   assert(eventBase->isInEventBaseThread());
 748   assert(queue_ == nullptr);
 749   assert(!isHandlerRegistered());
 750   queue->checkPid();
 751
 752   base_ = eventBase;
 753
 754   queue_ = queue;
 755
 756   {
 757     folly::SpinLockGuard g(queue_->spinlock_);
 758     queue_->numConsumers_++;
 759   }
 760   queue_->signalEvent();
 761
 762   if (queue_->eventfd_ >= 0) {
 763     initHandler(eventBase, queue_->eventfd_);
 764   } else {
 765     initHandler(eventBase, queue_->pipeFds_[0]);
 766   }
 767 }
 768
 769 template<typename MessageT>
 770 void NotificationQueue<MessageT>::Consumer::stopConsuming() {
 771   if (queue_ == nullptr) {
 772     assert(!isHandlerRegistered());
 773     return;
 774   }
 775
 776   {
 777     folly::SpinLockGuard g(queue_->spinlock_);
 778     queue_->numConsumers_--;
 779     setActive(false);
 780   }
 781
 782   assert(isHandlerRegistered());
 783   unregisterHandler();
 784   detachEventBase();
 785   queue_ = nullptr;
 786 }
 787
 788 template<typename MessageT>
 789 bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained(
 790     size_t* numConsumed) noexcept {
 791   DestructorGuard dg(this);
 792   {
 793     folly::SpinLockGuard g(queue_->spinlock_);
 794     if (queue_->draining_) {
 795       return false;
 796     }
 797     queue_->draining_ = true;
 798   }
 799   consumeMessages(true, numConsumed);
 800   {
 801     folly::SpinLockGuard g(queue_->spinlock_);
 802     queue_->draining_ = false;
 803   }
 804   return true;
 805 }
 806
 807 /**
 808  * Creates a NotificationQueue::Consumer wrapping a function object
 809  * Modeled after AsyncTimeout::make
 810  *
 811  */
 812
 813 namespace detail {
 814
 815 template <typename MessageT, typename TCallback>
 816 struct notification_queue_consumer_wrapper
 817     : public NotificationQueue<MessageT>::Consumer {
 818
 819   template <typename UCallback>
 820   explicit notification_queue_consumer_wrapper(UCallback&& callback)
 821       : callback_(std::forward<UCallback>(callback)) {}
 822
 823   // we are being stricter here and requiring noexcept for callback
 824   void messageAvailable(MessageT&& message) override {
 825     static_assert(
 826       noexcept(std::declval<TCallback>()(std::forward<MessageT>(message))),
 827       "callback must be declared noexcept, e.g.: `[]() noexcept {}`"
 828     );
 829
 830     callback_(std::forward<MessageT>(message));
 831   }
 832
 833  private:
 834   TCallback callback_;
 835 };
 836
 837 } // namespace detail
 838
 839 template <typename MessageT>
 840 template <typename TCallback>
 841 std::unique_ptr<typename NotificationQueue<MessageT>::Consumer,
 842                 DelayedDestruction::Destructor>
 843 NotificationQueue<MessageT>::Consumer::make(TCallback&& callback) {
 844   return std::unique_ptr<NotificationQueue<MessageT>::Consumer,
 845                          DelayedDestruction::Destructor>(
 846       new detail::notification_queue_consumer_wrapper<
 847           MessageT,
 848           typename std::decay<TCallback>::type>(
 849           std::forward<TCallback>(callback)));
 850 }
 851
 852 } // folly