MB-36956: FollyExecutorPool: Implement thread priorities

Use folly::PriorityThreadFactory when creating CPUPool
(Reader/Writer/AuxIO/NonIO) threads. Use the same priorties as
CB3ExecutorPool - Writers have lowest priority, all others have
default priority.

Change-Id: Id101962c781989edecd2cbc2a3a3c9075b3bd629
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/133974
Tested-by: Build Bot <[email protected]>
Reviewed-by: Trond Norbye <[email protected]>

Author: daverigby

engines/ep/src/cb3_executorthread.cc

@@ -85,36 +85,22 @@ void CB3ExecutorThread::run() {
     // BucketAllocationGuard to account task deletion to the bucket.
     NonBucketAllocationGuard guard;
 
-    // Decrease the priority of Writer threads to lessen their impact on
-    // other threads (esp front-end workers which should be prioritized ahead
-    // of non-critical path Writer tasks (both Flusher and Compaction).
-    // TODO: Investigate if it is worth increasing the priority of Flusher tasks
-    // which _are_ on the critical path for front-end operations - i.e.
-    // SyncWrites at level=persistMajority / persistActive.
-    // This could be done by having two different priority thread pools (say
-    // Low and High IO threads and putting critical path Reader (BGFetch) and
-    // Writer (SyncWrite flushes) on the High IO thread pool; keeping
-    // non-persist SyncWrites / normal mutations & compaction on the Low IO
-    // pool.
+    // Set priority based on this thread's task type.
 #if defined(__linux__)
     // Only doing this for Linux at present:
     // - On Windows folly's getpriority() compatability function changes the
     //   priority of the entire process.
     // - On macOS setpriority(PRIO_PROCESS) affects the entire process (unlike
     //   Linux where it's only the current thread), hence calling setpriority()
     //   would be pointless.
-    if (taskType == WRITER_TASK_IDX) {
-        // Note Linux uses the range -20..19 (highest..lowest).
-        const int lowestPriority = 19;
-        if (setpriority(PRIO_PROCESS,
-                        /*Current thread*/ 0,
-                        lowestPriority)) {
-            EP_LOG_WARN("Failed to set thread {} to background priority - {}",
-                        getName(),
-                        strerror(errno));
-        } else {
-            EP_LOG_INFO("Set thread {} to background priority", getName());
-        }
+    if (setpriority(PRIO_PROCESS,
+                    /*Current thread*/ 0,
+                    ExecutorPool::getThreadPriority(taskType))) {
+        EP_LOG_WARN("Failed to set thread {} to background priority - {}",
+                    getName(),
+                    strerror(errno));
+    } else {
+        EP_LOG_INFO("Set thread {} to background priority", getName());
     }
 #endif
 

engines/ep/src/executorpool.cc

@@ -183,3 +183,39 @@ size_t ExecutorPool::calcNumNonIO(size_t threadCount) const {
     }
     return count;
 }
+
+int ExecutorPool::getThreadPriority(task_type_t taskType) {
+    // Decrease the priority of Writer threads to lessen their impact on
+    // other threads (esp front-end workers which should be prioritized ahead
+    // of non-critical path Writer tasks (both Flusher and Compaction).
+    // TODO: Investigate if it is worth increasing the priority of Flusher tasks
+    // which _are_ on the critical path for front-end operations - i.e.
+    // SyncWrites at level=persistMajority / persistActive.
+    // This could be done by having two different priority thread pools (say
+    // Low and High IO threads and putting critical path Reader (BGFetch) and
+    // Writer (SyncWrite flushes) on the High IO thread pool; keeping
+    // non-persist SyncWrites / normal mutations & compaction on the Low IO
+    // pool.
+#if defined(__linux__)
+    // Only doing this for Linux at present:
+    // - On Windows folly's getpriority() compatability function changes the
+    //   priority of the entire process.
+    // - On macOS setpriority(PRIO_PROCESS) affects the entire process (unlike
+    //   Linux where it's only the current thread), hence calling setpriority()
+    //   would be pointless.
+    switch (taskType) {
+    case WRITER_TASK_IDX:
+        // Linux uses the range -20..19 (highest..lowest).
+        return 19;
+    case READER_TASK_IDX:
+    case AUXIO_TASK_IDX:
+    case NONIO_TASK_IDX:
+        return 0;
+    case NO_TASK_TYPE:
+    case NUM_TASK_GROUPS:
+        // These are both invalid taskTypes.
+        folly::assume_unreachable();
+    }
+#endif
+    return 0;
+}

engines/ep/src/executorpool.h

@@ -18,6 +18,7 @@
 #include <memcached/engine_common.h>
 #include <memcached/thread_pool_config.h>
 
+#include "task_type.h"
 #include <atomic>
 #include <memory>
 #include <mutex>
@@ -165,6 +166,13 @@ class ExecutorPool {
 
     virtual ~ExecutorPool() = default;
 
+    /**
+     * Return the thread priority to use for threads of the given task type.
+     * @param taskType
+     * @return priority to use, as passed to setpriority().
+     */
+    static int getThreadPriority(task_type_t taskType);
+
 protected:
     ExecutorPool(size_t maxThreads);
 

engines/ep/src/folly_executorpool.cc

@@ -28,6 +28,27 @@
 
 #include <folly/executors/CPUThreadPoolExecutor.h>
 #include <folly/executors/IOThreadPoolExecutor.h>
+#include <folly/executors/thread_factory/PriorityThreadFactory.h>
+
+/**
+ * Thread factory for CPU pool threads.
+ *  - Gives each thread name based on the given prefix.
+ *  - Sets each thread to the given priority.
+ */
+class CBPriorityThreadFactory : public folly::ThreadFactory {
+public:
+    CBPriorityThreadFactory(std::string prefix, int priority)
+        : priorityThreadFactory(
+                  std::make_shared<folly::NamedThreadFactory>(prefix),
+                  priority) {
+    }
+
+    std::thread newThread(folly::Func&& func) override {
+        return priorityThreadFactory.newThread(std::move(func));
+    }
+
+    folly::PriorityThreadFactory priorityThreadFactory;
+};
 
 /**
  * Proxy object recorded for each registered (scheduled) Task. Inherits from
@@ -433,19 +454,37 @@ FollyExecutorPool::FollyExecutorPool(size_t maxThreads,
       maxWriters(calcNumWriters(maxWriters_)),
       maxAuxIO(calcNumAuxIO(maxAuxIO_)),
       maxNonIO(calcNumNonIO(maxNonIO_)) {
+    /*
+     * Define a function to create thread factory with a given prefix,
+     * and priority where supported.
+     *
+     * Only setting priority for Linux at present:
+     *  - On Windows folly's getpriority() compatibility function changes the
+     *    priority of the entire process.
+     *  - On macOS setpriority(PRIO_PROCESS) affects the entire process (unlike
+     *    Linux where it's only the current thread), hence calling
+     *    setpriority() would be pointless.
+     */
+    auto makeThreadFactory = [](std::string prefix, task_type_t taskType) {
+#if defined(__linux__)
+        return std::make_shared<CBPriorityThreadFactory>(
+                prefix, ExecutorPool::getThreadPriority(taskType));
+#else
+        return std::make_shared<folly::NamedThreadFactory>(prefix);
+#endif
+    };
+
     futurePool = std::make_unique<folly::IOThreadPoolExecutor>(
             1, std::make_shared<folly::NamedThreadFactory>("SchedulerPool"));
 
     readerPool = std::make_unique<folly::CPUThreadPoolExecutor>(
-            maxReaders,
-            std::make_shared<folly::NamedThreadFactory>("ReaderPool")),
+            maxReaders, makeThreadFactory("ReaderPool", READER_TASK_IDX));
     writerPool = std::make_unique<folly::CPUThreadPoolExecutor>(
-            maxWriters,
-            std::make_shared<folly::NamedThreadFactory>("WriterPool"));
+            maxWriters, makeThreadFactory("WriterPool", WRITER_TASK_IDX));
     auxPool = std::make_unique<folly::CPUThreadPoolExecutor>(
-            maxAuxIO, std::make_shared<folly::NamedThreadFactory>("AuxIoPool"));
+            maxAuxIO, makeThreadFactory("AuxIoPool", AUXIO_TASK_IDX));
     nonIoPool = std::make_unique<folly::CPUThreadPoolExecutor>(
-            maxNonIO, std::make_shared<folly::NamedThreadFactory>("NonIoPool"));
+            maxNonIO, makeThreadFactory("NonIoPool", NONIO_TASK_IDX));
 }
 
 FollyExecutorPool::~FollyExecutorPool() {

engines/ep/tests/module_tests/executorpool_test.cc

@@ -440,11 +440,6 @@ TYPED_TEST(ExecutorPoolTest, increase_workers) {
 // Verifies the priority of the different thread types. On Windows and Linux
 // the Writer threads should be low priority.
 TYPED_TEST(ExecutorPoolTest, ThreadPriorities) {
-    if (typeid(TypeParam) == typeid(FollyExecutorPool)) {
-        // Not yet implemented for FollyExecutorPool.
-        GTEST_SKIP();
-    }
-
     // Create test pool and register a (mock) taskable to start all threads.
     this->makePool(10);