Give Glean derive methods a dedicated HIGH-priority thread pool

Summary:
High-level goal: run multiple indexing jobs on the same Glean write server.
Right now, when two indexing jobs run concurrently, the first job that reaches the derive stage will have its derive requests waiting in thrift queues behind many write requests, making the sandcastle jobs possibly retry and issuing duplicate derive requests. To fix that, the goal of this diff is to assign higher priority to derive requests.

To that end:

- Tag `deriveStored`/`deriveStoredV2` with
  `thrift.Priority{level = thrift.RpcPriority.HIGH}` in `glean.thrift`.
- Size the HIGH pool to 10 threads on the Glean server instead of the default 2.

Sizing the pool needs a way to configure the priority thread manager on an
hsthrift `CppServer`. This adds a first-class `cpuPriorityPoolSizes` field to
hsthrift's `ServerOptions`: a list of per-priority pool-size overrides keyed off
the canonical `Thrift.Protocol.RpcOptions.Types.Priority` enum. Priorities not
listed keep the fbthrift default. The overrides are marshalled to
`c_create_cpp_server`, which applies them via `setThreadManagerPoolSizes`.

To avoid duplicating fbthrift's default pool sizing, `PriorityThreadManager` now
exposes `defaultThreadCounts(normalThreadsCount)` as the single source of those
defaults; `newPriorityThreadManager(size_t)` delegates to it, and hsthrift
starts from it and applies only the requested overrides.

Notes:
- Priority pools are a `CppServer`-only feature, so the Glean override is
  compiled out of the HTTP (OSS) build.
- Existing `ServerOptions` consumers are unaffected: the new field defaults to
  `[]` (fbthrift's normal sizing) and they build options via
  `defaultOptions { ... }`.

Reviewed By: bochko

Differential Revision: D114073952

fbshipit-source-id: a0e6fca4b8fcb828bdd1bdbd999e1958ec51873c

Author: Catherine Gasnier

server/Thrift/Server/CppServer.hs

@@ -102,14 +102,25 @@ withBackgroundServer' handler postProcess ServerOptions{..} action =
     priorities =
       fromIntegral . fromThriftEnum . methodPriority <$> Map.elems infos
 
+    -- CPU worker pool size overrides, split into parallel arrays of priority
+    -- ordinal and size for the C++ side. Priorities not listed keep the
+    -- fbthrift default.
+    poolOverridePriorities =
+      fromIntegral . fromThriftEnum . fst <$> cpuPriorityPoolSizes
+    poolOverrideSizes = fromIntegral . snd <$> cpuPriorityPoolSizes
+    numPoolOverrides = fromIntegral (length cpuPriorityPoolSizes)
+
     -- Creates a PServer to run `act` on
     withCServer cb act =
       useTextsAsCStringLens names $ \names_ptr names_sizes names_len ->
       withArray priorities $ \priorities_ptr ->
-      withArray oneways $ \oneways_pts -> do
+      withArray oneways $ \oneways_pts ->
+      withArray poolOverridePriorities $ \pool_prios_ptr ->
+      withArray poolOverrideSizes $ \pool_sizes_ptr -> do
         let
           alloc =
             let create = create_cpp_server cb factoryFn cPort cNumWorkers
+                  pool_prios_ptr pool_sizes_ptr numPoolOverrides
                   priorities_ptr oneways_pts
                   names_ptr names_sizes names_len
             in
@@ -150,6 +161,9 @@ foreign import ccall safe "c_create_cpp_server"
                     -> FactoryFunction
                     -> CInt -- ^ port
                     -> CInt -- ^ workers, or 0 to use the default
+                    -> Ptr CInt -- ^ pool size override priorities
+                    -> Ptr CSize -- ^ pool size override sizes
+                    -> CSize -- ^ number of pool size overrides
                     -> Ptr CInt -- ^ method priorities array
                     -> Ptr Bool -- ^ one way methods
                     -> Ptr CString -- ^ method names array

server/Thrift/Server/Types.hs

@@ -19,6 +19,8 @@ import Control.Exception hiding (handle)
 import Data.Text (Text)
 import Foreign.Ptr
 
+import Thrift.Protocol.RpcOptions.Types (Priority)
+
 -- -----------------------------------------------------------------------------
 -- Factory function
 
@@ -47,6 +49,11 @@ data ServerOptions = ServerOptions
   , customFactoryFn :: Maybe FactoryFunction
       -- ^ whether a custom factory should be used
   , customModifyFn :: Maybe ModifyFunction
+  , cpuPriorityPoolSizes :: [(Priority, Int)]
+      -- ^ CPU worker pool size overrides, per Thrift RPC priority. Priorities
+      -- not listed keep the fbthrift default (see
+      -- @PriorityThreadManager::defaultThreadCounts@): the number of CPU cores
+      -- for NORMAL and an implementation-defined number for the rest.
   }
 
 -- | Takes the `onewayFunctions'` from your thrift Service instance
@@ -56,6 +63,7 @@ defaultOptions = ServerOptions
   , numWorkerThreads = Nothing
   , customFactoryFn = Nothing
   , customModifyFn = Nothing
+  , cpuPriorityPoolSizes = []
   }
 
 -- -----------------------------------------------------------------------------

server/cpp/CppServer.cpp

@@ -14,6 +14,7 @@
 #include <folly/Memory.h>
 #include <glog/logging.h>
 
+#include <thrift/lib/cpp/concurrency/ThreadManager.h>
 #include <thrift/lib/cpp2/server/ThriftServer.h>
 
 #include "cpp/Destructible.h"
@@ -138,6 +139,9 @@ CreateCppServerResult* c_create_cpp_server(
     apache::thrift::TFactory factoryFn,
     int desiredPort,
     int workers,
+    const int* poolSizeOverridePriorities,
+    const size_t* poolSizeOverrideSizes,
+    size_t numPoolSizeOverrides,
     const apache::thrift::concurrency::PRIORITY* methodPriorities,
     const bool* methodOneways,
     const char** methodNames,
@@ -173,6 +177,25 @@ CreateCppServerResult* c_create_cpp_server(
       cppServer->setNumCPUWorkerThreads(workers);
     }
 
+    if (numPoolSizeOverrides > 0) {
+      // Start from PriorityThreadManager's own defaults and override only the
+      // priorities the caller asked for, so we never duplicate or drift from
+      // the default pool sizing.
+      auto poolSizes = apache::thrift::concurrency::PriorityThreadManager::
+          defaultThreadCounts(cppServer->getNumCPUWorkerThreads());
+      for (size_t i = 0; i < numPoolSizeOverrides; i++) {
+        const int priority = poolSizeOverridePriorities[i];
+        if (priority >= 0 &&
+            priority <
+                static_cast<int>(apache::thrift::concurrency::N_PRIORITIES)) {
+          poolSizes[priority] = poolSizeOverrideSizes[i];
+        }
+      }
+      cppServer->setThreadManagerType(
+          apache::thrift::ThriftServer::ThreadManagerType::PRIORITY);
+      cppServer->setThreadManagerPoolSizes(poolSizes);
+    }
+
     return new CreateCppServerResult(HsLeft, std::move(cppServer));
   } catch (const std::exception& e) {
     auto exStr = e.what();

server/test/ServerTest.hs

@@ -132,6 +132,7 @@ portAlreadyBoundTest pname protId =
       , numWorkerThreads = Just 2
       , customFactoryFn = Nothing
       , customModifyFn = Nothing
+      , cpuPriorityPoolSizes = []
       }
     headerConfig :: HeaderConfig t
     headerConfig = mkHeaderConfig port protId