Name thread pools to match production: ThriftSrv.IO, SREventBase, RANKER, GlobalCPUThread

Summary:
LeafNodeRank's existing thread pools were anonymous from Strobelight's point of view, so the prod-vs-bench thread-pool breakdown landed almost entirely in the `Unknown` / framework-noise bucket on CPL and BGM. Per the multifeed_aggregator prod profile (~/feedsim_v2/profiles/multifeed_aggregator_main_prod/), the four hot pools are `ThriftSrv.IO`, `SREventBase{N}`, `RANKER-{N}`, and `GlobalCPUThread` (see ~/feedsim_v2/docs/phase4_researcher_notes.md section 1).

This diff is the Programmer-A half of Phase 4 (thread pools only). It renames the four existing pools to match the prod names that Strobelight categorizes, and adds two new pools (`SREventBase`, `RANKER`) that Phase 5 will start dispatching outbound RPC fanout onto. Programmer-B's diff (thrift schema + 5 new method registrations) lands as a sibling commit; via() callsites stay on their existing pool aliases so this rename is a no-op behaviorally.

Changes:
- `cpuThreadPool` is now backed by `folly::getGlobalCPUExecutor()` (the folly singleton already exposes its threads as `GlobalCPUThread` via `NamedThreadFactory("GlobalCPUThreadPool")` in `folly/executors/GlobalExecutor.cpp:65`). DO NOT instantiate a second CPU pool — that would double-count the prod `GlobalCPUThread` category. The `ThreadData::cpuThreadPool` field type changed from `shared_ptr<CPUThreadPoolExecutor>` to `shared_ptr<folly::Executor>` so the same field can hold the global singleton via an aliasing shared_ptr that owns a `folly::Executor::KeepAlive<>`. All existing `folly::via(this_thread.cpuThreadPool.get(), ...)` callsites continue to compile because `folly::via` accepts an `Executor*`.
- `srvCPUThreadPool` is now `NamedThreadFactory("RANKER")`, sized `max(1, nproc/2)` by default (matches CPL prod: 26 RANKER threads on a 52-logical-core host). Tunable via the new `--ranker_threads` flag.
- `ioThreadPool` is now `NamedThreadFactory("ThriftSrv.IO")`, sized `--io_threads`. Was previously anonymous (the kernel just labeled the threads with the executable name).
- New `srEventBasePool` (`folly::IOThreadPoolExecutor`, `NamedThreadFactory("SREventBase")`), sized `max(1, nproc * 7 / 10)` (matches CPL prod: 39 SREventBase threads on a 52-logical-core host). Tunable via the new `--sr_event_base_threads` flag. Idle in Phase 4 — Phase 5 wires the outbound mock_services fanout onto it. Threads are warmed up at server start so Strobelight sees them in steady state.
- The legacy `srvIOThreadPool` (compression dispatch) is preserved for now and retired in Phase 6 once compression callsites move onto `GlobalCPUThread` per the researcher notes.
- New CLI flags `--ranker_threads` and `--sr_event_base_threads` in `LeafNodeRankCmdline.ggo`. Default `0` means "auto-compute from `folly::available_concurrency()` per the formulas above".
- Includes added: `<folly/executors/GlobalExecutor.h>` and `<folly/system/HardwareConcurrency.h>`. Both are already on `${FOLLY_INCLUDE_DIR}` in the existing CMake target so no `CMakeLists.txt` edits were required.

Differential Revision: D103766488

Author: excelle08

packages/feedsim/third_party/src/workloads/ranking/LeafNodeRank.cc

@@ -28,10 +28,12 @@
 #include <folly/Range.h>
 #include <folly/compression/Compression.h>
 #include <folly/executors/CPUThreadPoolExecutor.h>
+#include <folly/executors/GlobalExecutor.h>
 #include <folly/executors/IOThreadPoolExecutor.h>
 #include <folly/futures/Future.h>
 #include <folly/futures/Promise.h>
 #include <folly/init/Init.h>
+#include <folly/system/HardwareConcurrency.h>
 
 #include <thrift/lib/cpp2/protocol/CompactProtocol.h>
 #include <thrift/lib/cpp2/protocol/Serializer.h>
@@ -86,10 +88,17 @@ enum class IOLatencyDistType {
 };
 
 struct ThreadData {
-  std::shared_ptr<folly::CPUThreadPoolExecutor> cpuThreadPool;
+  // Phase 4 thread-pool aliases (names kept for callsite stability):
+  //   cpuThreadPool       -> folly::getGlobalCPUExecutor() ("GlobalCPUThread")
+  //   srvCPUThreadPool    -> RANKER pool (NamedThreadFactory("RANKER"))
+  //   srvIOThreadPool     -> legacy compression pool (kept until Phase 6)
+  //   ioThreadPool        -> ThriftSrv.IO pool (NamedThreadFactory("ThriftSrv.IO"))
+  //   srEventBasePool     -> NEW outbound-RPC EventBase pool (idle in Phase 4)
+  std::shared_ptr<folly::Executor> cpuThreadPool;
   std::shared_ptr<folly::CPUThreadPoolExecutor> srvCPUThreadPool;
   std::shared_ptr<folly::CPUThreadPoolExecutor> srvIOThreadPool;
   std::shared_ptr<folly::IOThreadPoolExecutor> ioThreadPool;
+  std::shared_ptr<folly::IOThreadPoolExecutor> srEventBasePool;
   std::shared_ptr<ranking::TimekeeperPool> timekeeperPool;
   std::unique_ptr<ranking::dwarfs::PageRank> page_ranker;
 #ifdef FEEDSIM_USE_DLRM
@@ -171,17 +180,19 @@ void ThreadStartup(
     int thread_id,
     std::vector<ThreadData>& thread_data,
     ranking::dwarfs::PageRankParams& params,
-    const std::shared_ptr<folly::CPUThreadPoolExecutor>& cpuThreadPool,
+    const std::shared_ptr<folly::Executor>& cpuThreadPool,
     const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvCPUThreadPool,
     const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvIOThreadPool,
     const std::shared_ptr<folly::IOThreadPoolExecutor>& ioThreadPool,
+    const std::shared_ptr<folly::IOThreadPoolExecutor>& srEventBasePool,
     const std::shared_ptr<ranking::TimekeeperPool>& timekeeperPool,
     const std::shared_ptr<ranking::dwarfs::DLRM>& shared_dlrm_ranker) {
   auto& this_thread = thread_data[thread_id];
   this_thread.cpuThreadPool = cpuThreadPool;
   this_thread.srvCPUThreadPool = srvCPUThreadPool;
   this_thread.srvIOThreadPool = srvIOThreadPool;
   this_thread.ioThreadPool = ioThreadPool;
+  this_thread.srEventBasePool = srEventBasePool;
   this_thread.timekeeperPool = timekeeperPool;
 
   // Store shared DLRM ranker
@@ -283,17 +294,19 @@ void ThreadStartup(
     int thread_id,
     std::vector<ThreadData>& thread_data,
     ranking::dwarfs::PageRankParams& params,
-    const std::shared_ptr<folly::CPUThreadPoolExecutor>& cpuThreadPool,
+    const std::shared_ptr<folly::Executor>& cpuThreadPool,
     const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvCPUThreadPool,
     const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvIOThreadPool,
     const std::shared_ptr<folly::IOThreadPoolExecutor>& ioThreadPool,
+    const std::shared_ptr<folly::IOThreadPoolExecutor>& srEventBasePool,
     const std::shared_ptr<ranking::TimekeeperPool>& timekeeperPool) {
   auto& this_thread = thread_data[thread_id];
   auto graph = params.makeGraphCopy(g_shared_graph);
   this_thread.cpuThreadPool = cpuThreadPool;
   this_thread.srvCPUThreadPool = srvCPUThreadPool;
   this_thread.srvIOThreadPool = srvIOThreadPool;
   this_thread.ioThreadPool = ioThreadPool;
+  this_thread.srEventBasePool = srEventBasePool;
   this_thread.timekeeperPool = timekeeperPool;
   unsigned noderank_seed;
   if (args.node_rank_seed_given) {
@@ -1136,19 +1149,61 @@ int main(int argc, char** argv) {
   char* fake_argv[2] = {const_cast<char*>("./LeafNodeRank"), nullptr};
   char** sargv = static_cast<char**>(fake_argv);
   folly::init(&fake_argc, &sargv);
-  auto cpuThreadPool =
-      std::make_shared<folly::CPUThreadPoolExecutor>(args.cpu_threads_arg);
 
+  // Phase 4: production-shaped thread pools. Names (visible in
+  // /proc/$pid/task/*/comm and Strobelight) match the multifeed_aggregator
+  // prod profile: ThriftSrv.IO, RANKER, SREventBase, GlobalCPUThread.
+  const unsigned int nproc = folly::available_concurrency();
+
+  // GlobalCPUThread: shared folly singleton. DLRM inference, feature
+  // extraction, and compression all dispatch here. DO NOT construct a
+  // second CPUThreadPoolExecutor named "GlobalCPUThreadPool" — folly's
+  // global executor (folly/executors/GlobalExecutor.cpp) already has that
+  // name and is sized to nproc by default.
+  auto globalCpuKa = folly::getGlobalCPUExecutor();
+  folly::Executor* globalCpuRaw = globalCpuKa.get();
+  auto globalCpuKaPtr =
+      std::make_shared<folly::Executor::KeepAlive<>>(std::move(globalCpuKa));
+  // Aliasing shared_ptr: holds the KeepAlive alive, exposes raw Executor*.
+  std::shared_ptr<folly::Executor> cpuThreadPool(globalCpuKaPtr, globalCpuRaw);
+
+  // RANKER: ranking-orchestration / response-generation pool. Sized to
+  // nproc/2 by default (matches CPL prod: 26 RANKER threads on a
+  // 52-logical-core host). Replaces the legacy "srvCPUThread" pool.
+  const int rankerThreads = (args.ranker_threads_arg > 0)
+      ? args.ranker_threads_arg
+      : std::max(1, static_cast<int>(nproc) / 2);
   auto srvCPUThreadPool = std::make_shared<folly::CPUThreadPoolExecutor>(
-      args.srv_threads_arg,
-      std::make_shared<folly::NamedThreadFactory>("srvCPUThread"));
+      rankerThreads,
+      std::make_shared<folly::NamedThreadFactory>("RANKER"));
 
+  // Legacy pool kept only for compression callsites until Phase 6.
   auto srvIOThreadPool = std::make_shared<folly::CPUThreadPoolExecutor>(
       args.srv_io_threads_arg,
       std::make_shared<folly::NamedThreadFactory>("srvIOThread"));
 
-  auto ioThreadPool =
-      std::make_shared<folly::IOThreadPoolExecutor>(args.io_threads_arg);
+  // ThriftSrv.IO: inbound RPC IO loop. Renamed (was anonymous folly default).
+  auto ioThreadPool = std::make_shared<folly::IOThreadPoolExecutor>(
+      args.io_threads_arg,
+      std::make_shared<folly::NamedThreadFactory>("ThriftSrv.IO"));
+
+  // SREventBase: outbound-RPC EventBase pool. Idle in Phase 4 (Phase 5
+  // wires mock_services fanout to it). Sized 0.7 * nproc by default
+  // (matches CPL prod: 39 SREventBase threads on a 52-logical-core host).
+  const int srEventBaseThreads = (args.sr_event_base_threads_arg > 0)
+      ? args.sr_event_base_threads_arg
+      : std::max(1, (static_cast<int>(nproc) * 7) / 10);
+  auto srEventBasePool = std::make_shared<folly::IOThreadPoolExecutor>(
+      srEventBaseThreads,
+      std::make_shared<folly::NamedThreadFactory>("SREventBase"));
+
+  std::cout << "Thread pools (nproc=" << nproc << "): "
+            << "GlobalCPUThread (folly singleton), "
+            << "RANKER=" << rankerThreads << ", "
+            << "SREventBase=" << srEventBaseThreads << " (idle in Phase 4), "
+            << "ThriftSrv.IO=" << args.io_threads_arg << ", "
+            << "srvIOThread (legacy compression)=" << args.srv_io_threads_arg
+            << std::endl;
 
   auto timekeeperPool =
       std::make_shared<ranking::TimekeeperPool>(args.timekeeper_threads_arg);
@@ -1159,7 +1214,7 @@ int main(int argc, char** argv) {
   {
     const int warmup_tasks = 100;  // Run multiple tasks to ensure all threads are active
 
-    // Warm up CPU thread pool
+    // Warm up CPU thread pool (= folly global CPU executor).
     std::vector<folly::Future<int>> cpuFutures;
     for (int i = 0; i < warmup_tasks; i++) {
       cpuFutures.push_back(folly::via(cpuThreadPool.get(), []() {
@@ -1170,6 +1225,16 @@ int main(int argc, char** argv) {
     }
     folly::collectAll(std::move(cpuFutures)).get();
 
+    // Warm up SREventBase pool so threads spawn and Strobelight sees them
+    // even when nothing is dispatched there in Phase 4.
+    std::vector<folly::Future<int>> srEbFutures;
+    for (int i = 0; i < warmup_tasks; i++) {
+      srEbFutures.push_back(folly::via(srEventBasePool.get(), []() {
+        return 1;
+      }));
+    }
+    folly::collectAll(std::move(srEbFutures)).get();
+
     // Warm up srvCPU thread pool
     std::vector<folly::Future<int>> srvCPUFutures;
     for (int i = 0; i < warmup_tasks; i++) {
@@ -1308,6 +1373,7 @@ int main(int argc, char** argv) {
         srvCPUThreadPool,
         srvIOThreadPool,
         ioThreadPool,
+        srEventBasePool,
         timekeeperPool,
         shared_dlrm_ranker);
 #else
@@ -1319,6 +1385,7 @@ int main(int argc, char** argv) {
         srvCPUThreadPool,
         srvIOThreadPool,
         ioThreadPool,
+        srEventBasePool,
         timekeeperPool);
 #endif
   });

packages/feedsim/third_party/src/workloads/ranking/LeafNodeRankCmdline.ggo

@@ -46,6 +46,8 @@ option "cpu_threads" - "Number of threads to use for computation." int default="
 option "srv_threads" - "Number of threads for srv computation." int default="1"
 option "srv_io_threads" - "Number of threads for srv IO computation." int default="1"
 option "io_threads" - "Number of threads to use for IO." int default="1"
+option "ranker_threads" - "Number of threads in the RANKER orchestration pool. Default 0 = auto = max(1, nproc/2)." int default="0"
+option "sr_event_base_threads" - "Number of threads in the SREventBase outbound EventBase pool. Default 0 = auto = max(1, (nproc * 7) / 10)." int default="0"
 option "timekeeper_threads" - "Number of threads to use for timekeepers." int default="1"
 option "port" - "Port to run server on." int default="11222"
 option "monitor_port" - "Port to run monitoring server on." int default="8888"