Revert D26163397: python event loop consumes one task at a time from C++ notification queue

Differential Revision:
D26163397 (https://github.com/facebook/folly/commit/7c9f69f9c72b598c5dd942d93a94f371c6fa901d)

Original commit changeset: 48cd6cb57c48

fbshipit-source-id: fe0dd79a4e9ec77194cabf0edc697d9746542d3c

folly/io/async/NotificationQueue.h

@@ -229,7 +229,7 @@ class NotificationQueue {
     }
 
     template <typename F>
-    void consume(F&& f);
+    void consumeUntilDrained(F&& foreach);
 
    private:
     NotificationQueue& queue_;
@@ -838,27 +838,30 @@ bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained(
 
 template <typename MessageT>
 template <typename F>
-void NotificationQueue<MessageT>::SimpleConsumer::consume(F&& foreach) {
+void NotificationQueue<MessageT>::SimpleConsumer::consumeUntilDrained(
+    F&& foreach) {
   SCOPE_EXIT { queue_.syncSignalAndQueue(); };
 
   queue_.checkPid();
 
-  std::unique_ptr<Node> data;
-  {
-    folly::SpinLockGuard g(queue_.spinlock_);
+  while (true) {
+    std::unique_ptr<Node> data;
+    {
+      folly::SpinLockGuard g(queue_.spinlock_);
 
-    if (UNLIKELY(queue_.queue_.empty())) {
-      return;
+      if (UNLIKELY(queue_.queue_.empty())) {
+        return;
+      }
+
+      data.reset(&queue_.queue_.front());
+      queue_.queue_.pop_front();
     }
 
-    data.reset(&queue_.queue_.front());
-    queue_.queue_.pop_front();
+    RequestContextScopeGuard rctx(std::move(data->ctx_));
+    foreach(std::move(data->msg_));
+    // Make sure message destructor is called with the correct RequestContext.
+    data.reset();
   }
-
-  RequestContextScopeGuard rctx(std::move(data->ctx_));
-  foreach(std::move(data->msg_));
-  // Make sure message destructor is called with the correct RequestContext.
-  data.reset();
 }
 
 /**

folly/python/AsyncioExecutor.h

@@ -48,7 +48,7 @@ class AsyncioExecutor : public DrivableExecutor, public SequencedExecutor {
   int fileno() const { return consumer_.getFd(); }
 
   void drive() noexcept override {
-    consumer_.consume([](Func&& func) {
+    consumer_.consumeUntilDrained([](Func&& func) {
       if (FOLLY_DETAIL_PY_ISFINALIZING()) {
         // if Python is finalizing calling scheduled functions MAY segfault.
         // any code that could have been called is now inconsequential.