Introduced `folly/executors/EDFThreadPoolExecutor`.

Summary: This patch introduces an earliest-deadline-first (EDF) executor: `folly/executors/EDFThreadPoolExecutor`.

Reviewed By: interwq

Differential Revision: D13983430

fbshipit-source-id: 0df9bc4a1cdbe9059489d1e1abad88b9e6d17ad9

folly/executors/EDFThreadPoolExecutor.h

+/*
+ * Copyright 2019-present Facebook, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <atomic>
+#include <cstddef>
+#include <memory>
+#include <vector>
+
+#include <folly/executors/SoftRealTimeExecutor.h>
+#include <folly/executors/ThreadPoolExecutor.h>
+#include <folly/synchronization/LifoSem.h>
+
+namespace folly {
+
+/**
+ * `EDFThreadPoolExecutor` is a `SoftRealTimeExecutor` that implements
+ * the earliest-deadline-first scheduling policy.
+ */
+class EDFThreadPoolExecutor : public SoftRealTimeExecutor,
+                              public ThreadPoolExecutor {
+ public:
+  class Task;
+  class TaskQueue;
+
+  static constexpr uint64_t kEarliestDeadline = 0;
+  static constexpr uint64_t kLatestDeadline =
+      std::numeric_limits<uint64_t>::max();
+
+  explicit EDFThreadPoolExecutor(
+      std::size_t numThreads,
+      std::shared_ptr<ThreadFactory> threadFactory =
+          std::make_shared<NamedThreadFactory>("EDFThreadPool"));
+
+  ~EDFThreadPoolExecutor() override;
+
+  using ThreadPoolExecutor::add;
+
+  void add(Func f) override;
+  void add(Func f, uint64_t deadline) override;
+  void add(Func f, std::size_t total, uint64_t deadline);
+  void add(std::vector<Func> fs, uint64_t deadline);
+
+  folly::Executor::KeepAlive<> deadlineExecutor(uint64_t deadline);
+
+ private:
+  void threadRun(ThreadPtr thread) override;
+  void stopThreads(std::size_t numThreads) override;
+  std::size_t getPendingTaskCountImpl() const override;
+
+  bool shouldStop();
+  std::shared_ptr<Task> take();
+
+  std::unique_ptr<TaskQueue> taskQueue_;
+  LifoSem sem_;
+  std::atomic<int> threadsToStop_{0};
+
+  // All operations performed on `numIdleThreads_` explicitly specify memory
+  // ordering of `std::memory_order_seq_cst`. This is due to `numIdleThreads_`
+  // performing Dekker's algorithm with `numItems` prior to consumer threads
+  // (workers) wait on `sem_`.
+  std::atomic<std::size_t> numIdleThreads_{0};
+};
+
+} // namespace folly

folly/executors/SoftRealTimeExecutor.h

+/*
+ * Copyright 2019-present Facebook, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <folly/Executor.h>
+
+namespace folly {
+
+// `SoftRealTimeExecutor` is an executor that performs some priority-based
+// scheduling with a deadline assigned to each task. __Soft__ real-time
+// means that not every deadline is guaranteed to be met.
+class SoftRealTimeExecutor : public virtual Executor {
+  void add(Func) override = 0;
+
+  // Add a task with an assigned abstract deadline.
+  //
+  // NOTE: The type of `deadline` was chosen to be an integral rather than
+  // a typed time point or duration (e.g., `std::chrono::time_point`) to allow
+  // for flexbility. While the deadline for a task may be a time point,
+  // it could also be a duration or the size of the task, which emulates
+  // rate-monotonic scheduling that prioritizes small tasks. It also enables
+  // for exmaple, tiered scheduling (strictly prioritizing a category of tasks)
+  // by assigning the high-bit of the deadline.
+  virtual void add(Func, uint64_t deadline) = 0;
+};
+
+} // namespace folly

folly/executors/ThreadPoolExecutor.cpp

@@ -113,6 +113,11 @@ void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) {
   });
 }
 
+void ThreadPoolExecutor::add(Func, std::chrono::milliseconds, Func) {
+  throw std::runtime_error(
+      "add() with expiration is not implemented for this Executor");
+}
+
 size_t ThreadPoolExecutor::numThreads() const {
   return maxThreads_.load(std::memory_order_relaxed);
 }

folly/executors/ThreadPoolExecutor.h

@@ -14,6 +14,11 @@
  * limitations under the License.
  */
 #pragma once
+
+#include <algorithm>
+#include <mutex>
+#include <queue>
+
 #include <folly/DefaultKeepAliveExecutor.h>
 #include <folly/Memory.h>
 #include <folly/SharedMutex.h>
@@ -24,10 +29,6 @@
 #include <folly/portability/GFlags.h>
 #include <folly/synchronization/Baton.h>
 
-#include <algorithm>
-#include <mutex>
-#include <queue>
-
 #include <glog/logging.h>
 
 namespace folly {
@@ -64,7 +65,7 @@ class ThreadPoolExecutor : public DefaultKeepAliveExecutor {
 
   void add(Func func) override = 0;
   virtual void
-  add(Func func, std::chrono::milliseconds expiration, Func expireCallback) = 0;
+  add(Func func, std::chrono::milliseconds expiration, Func expireCallback);
 
   void setThreadFactory(std::shared_ptr<ThreadFactory> threadFactory) {
     CHECK(numThreads() == 0);

folly/executors/test/ThreadPoolExecutorTest.cpp

@@ -23,6 +23,7 @@
 #include <folly/Exception.h>
 #include <folly/VirtualExecutor.h>
 #include <folly/executors/CPUThreadPoolExecutor.h>
+#include <folly/executors/EDFThreadPoolExecutor.h>
 #include <folly/executors/FutureExecutor.h>
 #include <folly/executors/IOThreadPoolExecutor.h>
 #include <folly/executors/ThreadPoolExecutor.h>
@@ -50,10 +51,14 @@ TEST(ThreadPoolExecutorTest, CPUBasic) {
   basic<CPUThreadPoolExecutor>();
 }
 
-TEST(IOThreadPoolExecutorTest, IOBasic) {
+TEST(ThreadPoolExecutorTest, IOBasic) {
   basic<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFBasic) {
+  basic<EDFThreadPoolExecutor>();
+}
+
 template <class TPE>
 static void resize() {
   TPE tpe(100);
@@ -72,6 +77,10 @@ TEST(ThreadPoolExecutorTest, IOResize) {
   resize<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFResize) {
+  resize<EDFThreadPoolExecutor>();
+}
+
 template <class TPE>
 static void stop() {
   TPE tpe(1);
@@ -113,6 +122,10 @@ TEST(ThreadPoolExecutorTest, IOStop) {
   stop<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFStop) {
+  stop<EDFThreadPoolExecutor>();
+}
+
 template <class TPE>
 static void join() {
   TPE tpe(10);
@@ -136,6 +149,10 @@ TEST(ThreadPoolExecutorTest, IOJoin) {
   join<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFJoin) {
+  join<EDFThreadPoolExecutor>();
+}
+
 template <class TPE>
 static void destroy() {
   TPE tpe(1);
@@ -177,6 +194,10 @@ TEST(ThreadPoolExecutorTest, IODestroy) {
   destroy<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFDestroy) {
+  destroy<EDFThreadPoolExecutor>();
+}
+
 template <class TPE>
 static void resizeUnderLoad() {
   TPE tpe(10);
@@ -202,6 +223,10 @@ TEST(ThreadPoolExecutorTest, IOResizeUnderLoad) {
   resizeUnderLoad<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFResizeUnderLoad) {
+  resizeUnderLoad<EDFThreadPoolExecutor>();
+}
+
 template <class TPE>
 static void poolStats() {
   folly::Baton<> startBaton, endBaton;
@@ -262,6 +287,10 @@ TEST(ThreadPoolExecutorTest, IOTaskStats) {
   taskStats<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFTaskStats) {
+  taskStats<EDFThreadPoolExecutor>();
+}
+
 template <class TPE>
 static void expiration() {
   TPE tpe(1);
@@ -347,6 +376,10 @@ TEST(ThreadPoolExecutorTest, IOFuturePool) {
   futureExecutor<IOThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, EDFFuturePool) {
+  futureExecutor<EDFThreadPoolExecutor>();
+}
+
 TEST(ThreadPoolExecutorTest, PriorityPreemptionTest) {
   bool tookLopri = false;
   auto completed = 0;
@@ -394,11 +427,12 @@ class TestObserver : public ThreadPoolExecutor::Observer {
   std::atomic<int> threads_{0};
 };
 
-TEST(ThreadPoolExecutorTest, IOObserver) {
+template <typename TPE>
+static void testObserver() {
   auto observer = std::make_shared<TestObserver>();
 
   {
-    IOThreadPoolExecutor exe(10);
+    TPE exe(10);
     exe.addObserver(observer);
     exe.setNumThreads(3);
     exe.setNumThreads(0);
@@ -410,20 +444,16 @@ TEST(ThreadPoolExecutorTest, IOObserver) {
   observer->checkCalls();
 }
 
-TEST(ThreadPoolExecutorTest, CPUObserver) {
-  auto observer = std::make_shared<TestObserver>();
+TEST(ThreadPoolExecutorTest, IOObserver) {
+  testObserver<IOThreadPoolExecutor>();
+}
 
-  {
-    CPUThreadPoolExecutor exe(10);
-    exe.addObserver(observer);
-    exe.setNumThreads(3);
-    exe.setNumThreads(0);
-    exe.setNumThreads(7);
-    exe.removeObserver(observer);
-    exe.setNumThreads(10);
-  }
+TEST(ThreadPoolExecutorTest, CPUObserver) {
+  testObserver<CPUThreadPoolExecutor>();
+}
 
-  observer->checkCalls();
+TEST(ThreadPoolExecutorTest, EDFObserver) {
+  testObserver<EDFThreadPoolExecutor>();
 }
 
 TEST(ThreadPoolExecutorTest, AddWithPriority) {
@@ -434,6 +464,10 @@ TEST(ThreadPoolExecutorTest, AddWithPriority) {
   IOThreadPoolExecutor ioExe(10);
   EXPECT_THROW(ioExe.addWithPriority(f, 0), std::runtime_error);
 
+  // EDF exe doesn't support priorities
+  EDFThreadPoolExecutor edfExe(10);
+  EXPECT_THROW(edfExe.addWithPriority(f, 0), std::runtime_error);
+
   CPUThreadPoolExecutor cpuExe(10, 3);
   cpuExe.addWithPriority(f, -1);
   cpuExe.addWithPriority(f, 0);
@@ -715,6 +749,10 @@ TEST(ThreadPoolExecutorTest, RemoveThreadTestCPU) {
   removeThreadTest<CPUThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, RemoveThreadTestEDF) {
+  removeThreadTest<EDFThreadPoolExecutor>();
+}
+
 template <typename TPE>
 static void resizeThreadWhileExecutingTest() {
   TPE tpe(10);
@@ -746,6 +784,10 @@ TEST(ThreadPoolExecutorTest, resizeThreadWhileExecutingTestCPU) {
   resizeThreadWhileExecutingTest<CPUThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, resizeThreadWhileExecutingTestEDF) {
+  resizeThreadWhileExecutingTest<EDFThreadPoolExecutor>();
+}
+
 template <typename TPE>
 void keepAliveTest() {
   auto executor = std::make_unique<TPE>(4);
@@ -770,6 +812,10 @@ TEST(ThreadPoolExecutorTest, KeepAliveTestCPU) {
   keepAliveTest<CPUThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, KeepAliveTestEDF) {
+  keepAliveTest<EDFThreadPoolExecutor>();
+}
+
 int getNumThreadPoolExecutors() {
   int count = 0;
   ThreadPoolExecutor::withAll([&count](ThreadPoolExecutor&) { count++; });
@@ -799,6 +845,10 @@ TEST(ThreadPoolExecutorTest, registersToExecutorListTestCPU) {
   registersToExecutorListTest<CPUThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, registersToExecutorListTestEDF) {
+  registersToExecutorListTest<EDFThreadPoolExecutor>();
+}
+
 template <typename TPE>
 static void testUsesNameFromNamedThreadFactory() {
   auto ntf = std::make_shared<NamedThreadFactory>("my_executor");
@@ -814,6 +864,10 @@ TEST(ThreadPoolExecutorTest, testUsesNameFromNamedThreadFactoryCPU) {
   testUsesNameFromNamedThreadFactory<CPUThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, testUsesNameFromNamedThreadFactoryEDF) {
+  testUsesNameFromNamedThreadFactory<EDFThreadPoolExecutor>();
+}
+
 TEST(ThreadPoolExecutorTest, DynamicThreadsTest) {
   boost::barrier barrier{3};
   auto twice_waiting_task = [&] { barrier.wait(), barrier.wait(); };
@@ -933,6 +987,10 @@ TEST(ThreadPoolExecutorTest, WeakRefTestCPU) {
   WeakRefTest<CPUThreadPoolExecutor>();
 }
 
+TEST(ThreadPoolExecutorTest, WeakRefTestEDF) {
+  WeakRefTest<EDFThreadPoolExecutor>();
+}
+
 TEST(ThreadPoolExecutorTest, VirtualExecutorTestIO) {
   virtualExecutorTest<IOThreadPoolExecutor>();
 }
@@ -940,3 +998,7 @@ TEST(ThreadPoolExecutorTest, VirtualExecutorTestIO) {
 TEST(ThreadPoolExecutorTest, VirtualExecutorTestCPU) {
   virtualExecutorTest<CPUThreadPoolExecutor>();
 }
+
+TEST(ThreadPoolExecutorTest, VirtualExecutorTestEDF) {
+  virtualExecutorTest<EDFThreadPoolExecutor>();
+}