Skip to content

Commit ed98993

Browse files
excelle08meta-codesync[bot]
authored andcommitted
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
1 parent 65cd072 commit ed98993

2 files changed

Lines changed: 79 additions & 10 deletions

File tree

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

Lines changed: 77 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -28,10 +28,12 @@
2828
#include <folly/Range.h>
2929
#include <folly/compression/Compression.h>
3030
#include <folly/executors/CPUThreadPoolExecutor.h>
31+
#include <folly/executors/GlobalExecutor.h>
3132
#include <folly/executors/IOThreadPoolExecutor.h>
3233
#include <folly/futures/Future.h>
3334
#include <folly/futures/Promise.h>
3435
#include <folly/init/Init.h>
36+
#include <folly/system/HardwareConcurrency.h>
3537

3638
#include <thrift/lib/cpp2/protocol/CompactProtocol.h>
3739
#include <thrift/lib/cpp2/protocol/Serializer.h>
@@ -86,10 +88,17 @@ enum class IOLatencyDistType {
8688
};
8789

8890
struct ThreadData {
89-
std::shared_ptr<folly::CPUThreadPoolExecutor> cpuThreadPool;
91+
// Phase 4 thread-pool aliases (names kept for callsite stability):
92+
// cpuThreadPool -> folly::getGlobalCPUExecutor() ("GlobalCPUThread")
93+
// srvCPUThreadPool -> RANKER pool (NamedThreadFactory("RANKER"))
94+
// srvIOThreadPool -> legacy compression pool (kept until Phase 6)
95+
// ioThreadPool -> ThriftSrv.IO pool (NamedThreadFactory("ThriftSrv.IO"))
96+
// srEventBasePool -> NEW outbound-RPC EventBase pool (idle in Phase 4)
97+
std::shared_ptr<folly::Executor> cpuThreadPool;
9098
std::shared_ptr<folly::CPUThreadPoolExecutor> srvCPUThreadPool;
9199
std::shared_ptr<folly::CPUThreadPoolExecutor> srvIOThreadPool;
92100
std::shared_ptr<folly::IOThreadPoolExecutor> ioThreadPool;
101+
std::shared_ptr<folly::IOThreadPoolExecutor> srEventBasePool;
93102
std::shared_ptr<ranking::TimekeeperPool> timekeeperPool;
94103
std::unique_ptr<ranking::dwarfs::PageRank> page_ranker;
95104
#ifdef FEEDSIM_USE_DLRM
@@ -171,17 +180,19 @@ void ThreadStartup(
171180
int thread_id,
172181
std::vector<ThreadData>& thread_data,
173182
ranking::dwarfs::PageRankParams& params,
174-
const std::shared_ptr<folly::CPUThreadPoolExecutor>& cpuThreadPool,
183+
const std::shared_ptr<folly::Executor>& cpuThreadPool,
175184
const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvCPUThreadPool,
176185
const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvIOThreadPool,
177186
const std::shared_ptr<folly::IOThreadPoolExecutor>& ioThreadPool,
187+
const std::shared_ptr<folly::IOThreadPoolExecutor>& srEventBasePool,
178188
const std::shared_ptr<ranking::TimekeeperPool>& timekeeperPool,
179189
const std::shared_ptr<ranking::dwarfs::DLRM>& shared_dlrm_ranker) {
180190
auto& this_thread = thread_data[thread_id];
181191
this_thread.cpuThreadPool = cpuThreadPool;
182192
this_thread.srvCPUThreadPool = srvCPUThreadPool;
183193
this_thread.srvIOThreadPool = srvIOThreadPool;
184194
this_thread.ioThreadPool = ioThreadPool;
195+
this_thread.srEventBasePool = srEventBasePool;
185196
this_thread.timekeeperPool = timekeeperPool;
186197

187198
// Store shared DLRM ranker
@@ -283,17 +294,19 @@ void ThreadStartup(
283294
int thread_id,
284295
std::vector<ThreadData>& thread_data,
285296
ranking::dwarfs::PageRankParams& params,
286-
const std::shared_ptr<folly::CPUThreadPoolExecutor>& cpuThreadPool,
297+
const std::shared_ptr<folly::Executor>& cpuThreadPool,
287298
const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvCPUThreadPool,
288299
const std::shared_ptr<folly::CPUThreadPoolExecutor>& srvIOThreadPool,
289300
const std::shared_ptr<folly::IOThreadPoolExecutor>& ioThreadPool,
301+
const std::shared_ptr<folly::IOThreadPoolExecutor>& srEventBasePool,
290302
const std::shared_ptr<ranking::TimekeeperPool>& timekeeperPool) {
291303
auto& this_thread = thread_data[thread_id];
292304
auto graph = params.makeGraphCopy(g_shared_graph);
293305
this_thread.cpuThreadPool = cpuThreadPool;
294306
this_thread.srvCPUThreadPool = srvCPUThreadPool;
295307
this_thread.srvIOThreadPool = srvIOThreadPool;
296308
this_thread.ioThreadPool = ioThreadPool;
309+
this_thread.srEventBasePool = srEventBasePool;
297310
this_thread.timekeeperPool = timekeeperPool;
298311
unsigned noderank_seed;
299312
if (args.node_rank_seed_given) {
@@ -1136,19 +1149,61 @@ int main(int argc, char** argv) {
11361149
char* fake_argv[2] = {const_cast<char*>("./LeafNodeRank"), nullptr};
11371150
char** sargv = static_cast<char**>(fake_argv);
11381151
folly::init(&fake_argc, &sargv);
1139-
auto cpuThreadPool =
1140-
std::make_shared<folly::CPUThreadPoolExecutor>(args.cpu_threads_arg);
11411152

1153+
// Phase 4: production-shaped thread pools. Names (visible in
1154+
// /proc/$pid/task/*/comm and Strobelight) match the multifeed_aggregator
1155+
// prod profile: ThriftSrv.IO, RANKER, SREventBase, GlobalCPUThread.
1156+
const unsigned int nproc = folly::available_concurrency();
1157+
1158+
// GlobalCPUThread: shared folly singleton. DLRM inference, feature
1159+
// extraction, and compression all dispatch here. DO NOT construct a
1160+
// second CPUThreadPoolExecutor named "GlobalCPUThreadPool" — folly's
1161+
// global executor (folly/executors/GlobalExecutor.cpp) already has that
1162+
// name and is sized to nproc by default.
1163+
auto globalCpuKa = folly::getGlobalCPUExecutor();
1164+
folly::Executor* globalCpuRaw = globalCpuKa.get();
1165+
auto globalCpuKaPtr =
1166+
std::make_shared<folly::Executor::KeepAlive<>>(std::move(globalCpuKa));
1167+
// Aliasing shared_ptr: holds the KeepAlive alive, exposes raw Executor*.
1168+
std::shared_ptr<folly::Executor> cpuThreadPool(globalCpuKaPtr, globalCpuRaw);
1169+
1170+
// RANKER: ranking-orchestration / response-generation pool. Sized to
1171+
// nproc/2 by default (matches CPL prod: 26 RANKER threads on a
1172+
// 52-logical-core host). Replaces the legacy "srvCPUThread" pool.
1173+
const int rankerThreads = (args.ranker_threads_arg > 0)
1174+
? args.ranker_threads_arg
1175+
: std::max(1, static_cast<int>(nproc) / 2);
11421176
auto srvCPUThreadPool = std::make_shared<folly::CPUThreadPoolExecutor>(
1143-
args.srv_threads_arg,
1144-
std::make_shared<folly::NamedThreadFactory>("srvCPUThread"));
1177+
rankerThreads,
1178+
std::make_shared<folly::NamedThreadFactory>("RANKER"));
11451179

1180+
// Legacy pool kept only for compression callsites until Phase 6.
11461181
auto srvIOThreadPool = std::make_shared<folly::CPUThreadPoolExecutor>(
11471182
args.srv_io_threads_arg,
11481183
std::make_shared<folly::NamedThreadFactory>("srvIOThread"));
11491184

1150-
auto ioThreadPool =
1151-
std::make_shared<folly::IOThreadPoolExecutor>(args.io_threads_arg);
1185+
// ThriftSrv.IO: inbound RPC IO loop. Renamed (was anonymous folly default).
1186+
auto ioThreadPool = std::make_shared<folly::IOThreadPoolExecutor>(
1187+
args.io_threads_arg,
1188+
std::make_shared<folly::NamedThreadFactory>("ThriftSrv.IO"));
1189+
1190+
// SREventBase: outbound-RPC EventBase pool. Idle in Phase 4 (Phase 5
1191+
// wires mock_services fanout to it). Sized 0.7 * nproc by default
1192+
// (matches CPL prod: 39 SREventBase threads on a 52-logical-core host).
1193+
const int srEventBaseThreads = (args.sr_event_base_threads_arg > 0)
1194+
? args.sr_event_base_threads_arg
1195+
: std::max(1, (static_cast<int>(nproc) * 7) / 10);
1196+
auto srEventBasePool = std::make_shared<folly::IOThreadPoolExecutor>(
1197+
srEventBaseThreads,
1198+
std::make_shared<folly::NamedThreadFactory>("SREventBase"));
1199+
1200+
std::cout << "Thread pools (nproc=" << nproc << "): "
1201+
<< "GlobalCPUThread (folly singleton), "
1202+
<< "RANKER=" << rankerThreads << ", "
1203+
<< "SREventBase=" << srEventBaseThreads << " (idle in Phase 4), "
1204+
<< "ThriftSrv.IO=" << args.io_threads_arg << ", "
1205+
<< "srvIOThread (legacy compression)=" << args.srv_io_threads_arg
1206+
<< std::endl;
11521207

11531208
auto timekeeperPool =
11541209
std::make_shared<ranking::TimekeeperPool>(args.timekeeper_threads_arg);
@@ -1159,7 +1214,7 @@ int main(int argc, char** argv) {
11591214
{
11601215
const int warmup_tasks = 100; // Run multiple tasks to ensure all threads are active
11611216

1162-
// Warm up CPU thread pool
1217+
// Warm up CPU thread pool (= folly global CPU executor).
11631218
std::vector<folly::Future<int>> cpuFutures;
11641219
for (int i = 0; i < warmup_tasks; i++) {
11651220
cpuFutures.push_back(folly::via(cpuThreadPool.get(), []() {
@@ -1170,6 +1225,16 @@ int main(int argc, char** argv) {
11701225
}
11711226
folly::collectAll(std::move(cpuFutures)).get();
11721227

1228+
// Warm up SREventBase pool so threads spawn and Strobelight sees them
1229+
// even when nothing is dispatched there in Phase 4.
1230+
std::vector<folly::Future<int>> srEbFutures;
1231+
for (int i = 0; i < warmup_tasks; i++) {
1232+
srEbFutures.push_back(folly::via(srEventBasePool.get(), []() {
1233+
return 1;
1234+
}));
1235+
}
1236+
folly::collectAll(std::move(srEbFutures)).get();
1237+
11731238
// Warm up srvCPU thread pool
11741239
std::vector<folly::Future<int>> srvCPUFutures;
11751240
for (int i = 0; i < warmup_tasks; i++) {
@@ -1308,6 +1373,7 @@ int main(int argc, char** argv) {
13081373
srvCPUThreadPool,
13091374
srvIOThreadPool,
13101375
ioThreadPool,
1376+
srEventBasePool,
13111377
timekeeperPool,
13121378
shared_dlrm_ranker);
13131379
#else
@@ -1319,6 +1385,7 @@ int main(int argc, char** argv) {
13191385
srvCPUThreadPool,
13201386
srvIOThreadPool,
13211387
ioThreadPool,
1388+
srEventBasePool,
13221389
timekeeperPool);
13231390
#endif
13241391
});

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,8 @@ option "cpu_threads" - "Number of threads to use for computation." int default="
4646
option "srv_threads" - "Number of threads for srv computation." int default="1"
4747
option "srv_io_threads" - "Number of threads for srv IO computation." int default="1"
4848
option "io_threads" - "Number of threads to use for IO." int default="1"
49+
option "ranker_threads" - "Number of threads in the RANKER orchestration pool. Default 0 = auto = max(1, nproc/2)." int default="0"
50+
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"
4951
option "timekeeper_threads" - "Number of threads to use for timekeepers." int default="1"
5052
option "port" - "Port to run server on." int default="11222"
5153
option "monitor_port" - "Port to run monitoring server on." int default="8888"

0 commit comments

Comments
 (0)