Replace queue with UMPSCQueue and decouple queueing from scheduling

Summary:
Reduced contention with removing connection between queue_ and schedule_ access.
Replaced std::queue with unbounded multiple producer single consumer queue.

Reviewed By: yfeldblum

Differential Revision: D10345872

fbshipit-source-id: 3307d908d6e4826eab4d981447c8771021bfe580

folly/executors/SerialExecutor.cpp

@@ -56,36 +56,24 @@ void SerialExecutor::keepAliveRelease() {
 }
 
 void SerialExecutor::add(Func func) {
-  {
-    std::lock_guard<std::mutex> lock(mutex_);
-    queue_.push(std::move(func));
-  }
+  queue_.enqueue(std::move(func));
   parent_->add([keepAlive = getKeepAliveToken(this)] { keepAlive->run(); });
 }
 
 void SerialExecutor::addWithPriority(Func func, int8_t priority) {
-  {
-    std::lock_guard<std::mutex> lock(mutex_);
-    queue_.push(std::move(func));
-  }
+  queue_.enqueue(std::move(func));
   parent_->addWithPriority(
       [keepAlive = getKeepAliveToken(this)] { keepAlive->run(); }, priority);
 }
 
 void SerialExecutor::run() {
-  std::unique_lock<std::mutex> lock(mutex_);
-
-  ++scheduled_;
-
-  if (scheduled_ > 1) {
+  if (scheduled_.fetch_add(1, std::memory_order_relaxed) > 0) {
     return;
   }
 
   do {
-    DCHECK(!queue_.empty());
-    Func func = std::move(queue_.front());
-    queue_.pop();
-    lock.unlock();
+    Func func;
+    queue_.dequeue(func);
 
     try {
       func();
@@ -97,13 +85,7 @@ void SerialExecutor::run() {
                     "object";
     }
 
-    // Destroy the function (and the data it captures) before we acquire the
-    // lock again.
-    func = {};
-
-    lock.lock();
-    --scheduled_;
-  } while (scheduled_);
+  } while (scheduled_.fetch_sub(1, std::memory_order_relaxed) > 1);
 }
 
 } // namespace folly

folly/executors/SerialExecutor.h

@@ -19,8 +19,8 @@
 #include <atomic>
 #include <memory>
 #include <mutex>
-#include <queue>
 
+#include <folly/concurrency/UnboundedQueue.h>
 #include <folly/executors/GlobalExecutor.h>
 #include <folly/executors/SequencedExecutor.h>
 
@@ -111,9 +111,12 @@ class SerialExecutor : public SequencedExecutor {
   void run();
 
   KeepAlive<Executor> parent_;
-  std::mutex mutex_;
-  std::size_t scheduled_{0};
-  std::queue<Func> queue_;
+  std::atomic<std::size_t> scheduled_{0};
+  /**
+   * Unbounded multi producer single consumer queue where consumers don't block
+   * on dequeue.
+   */
+  folly::UnboundedQueue<Func, false, true, false> queue_;
 
   std::atomic<ssize_t> keepAliveCounter_{1};
 };