Move runFeatureExtraction off the dispatcher; enable for dlrm_mini

Summary:
DLRMRequestHandler used to call runFeatureExtraction synchronously on the dispatcher (ThriftSrv.IO) thread, blocking the IO loop for the duration of feature extraction. Worse, feedsim_autoscale_dlrm_mini didn't pass --feature-extractors at all, so on the fixed-QPS variant the call was a no-op anyway and FeatureExtraction's CPU share dropped to ~0% (vs production's 30-35%). The recent higher-QPS sweeps showed BGM at 80% idle past saturation — there's plenty of CPU room to host real feature extraction work, and getting the share up is required to match the production multifeed_aggregator microarchitectural profile.

This commit reshapes DLRMRequestHandler's pipeline:
* runFeatureExtraction is dispatched async via the same RANKER -> GlobalCPUThread pattern that GetStoriesUncompressedRequestHandler already uses (folly::via(srvCPUThreadPool, lambda { return folly::via(cpuThreadPool, runFeatureExtraction); })). RANKER orchestrates; GlobalCPUThread runs the generated vc_* extractor functions. The dispatcher returns immediately.
* Feature extraction starts at the head of the async chain in parallel with the DLRM inference future that was kicked off above. Both run on cpuThreadPool (folly::CPUThreadPoolExecutor is multi-threaded so the two stages overlap). After feature extraction completes, the chain waits for inference, then proceeds to simulateIoOrFanout / pointer chase / response generation.
* feedsim_autoscale_dlrm_mini now passes --feature-extractors --feature-complexity=5 --num-stories=100 --extractors-per-story=50 by default. That's 5000 extractor calls per request, the count tuned during Phase 2 to match production's instruction mix. If the resulting CPU share doesn't reach the 30-35% target, the knobs are tunable per-run via -i.

Differential Revision: D105119220

Author: excelle08

benchpress/config/jobs.yml

@@ -805,6 +805,10 @@
     - '--client-feature-seed={client_feature_seed}'
     - '--client-num-dense={client_num_dense}'
     - '--client-num-sparse={client_num_sparse}'
+    - '--feature-extractors'
+    - '--feature-complexity={feature_complexity}'
+    - '--num-stories={num_stories}'
+    - '--extractors-per-story={extractors_per_story}'
     - '-q {fixed_qps}'
     - '-d {fixed_qps_duration}'
     - '-w {warmup_time}'
@@ -831,6 +835,9 @@
     - 'client_feature_seed=42'
     - 'client_num_dense=13'
     - 'client_num_sparse=26'
+    - 'feature_complexity=5'
+    - 'num_stories=100'
+    - 'extractors_per_story=50'
     - 'fixed_qps=100000'
     - 'fixed_qps_duration=10'
     - 'warmup_time=5'

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

@@ -1192,10 +1192,10 @@ void DLRMRequestHandler(
     }
   }
 
-  // Run feature extraction if enabled (with story content if available)
-  runFeatureExtraction(this_thread, story_contents);
-
-  // Populate CopyContext structData from story content when available
+  // Populate CopyContext structData from story content when available.
+  // This is small (KB-scale memcpy + hashing) so it stays on the
+  // dispatcher; the heavy ranking-side feature extraction runs async
+  // below.
   if (!story_contents.empty()) {
     std::vector<float> story_dense;
     std::vector<int64_t> story_sparse;
@@ -1254,10 +1254,17 @@ void DLRMRequestHandler(
       ? (num_io_stages * io_stage_latency_ms)
       : io_latency_ms;
 
-  // Pipeline: DLRM inference -> I/O sleep (or RPC fanout) -> pointer
-  // chase -> generate+send response. Everything chains via futures so
-  // the handler thread returns immediately and the request is processed
-  // entirely off the dispatcher thread.
+  // Pipeline:
+  //   - DLRM inference (already started above, runs on cpuThreadPool)
+  //   - Feature extraction in parallel: orchestrated by RANKER, the
+  //     actual extractor work delegated to GlobalCPUThread (matches the
+  //     prod thread-pool layout: ranking orchestration on RANKER, heavy
+  //     CPU extraction on GlobalCPUThread). Production multifeed
+  //     aggregator spends ~30-35% of CPU here, so we keep it in the
+  //     hot path.
+  //   - collectAll the two, then proceed to I/O fanout, pointer chase,
+  //     response generation. Everything chains via futures so the
+  //     dispatcher thread returns immediately.
   //
   // Phase 5: when --rpc_dist_path is set, the I/O sleep is replaced by
   // an outbound RPC fanout to mock_services on srEventBasePool (see
@@ -1268,7 +1275,33 @@ void DLRMRequestHandler(
   // now redundant.
   auto timekeeper = timekeeperPool->getTimekeeper();
   ThreadData* this_thread_ptr = &this_thread;
-  std::move(inference_future)
+  auto cpuPool = this_thread.cpuThreadPool;
+
+  // Stage A: hop to RANKER, then delegate feature extraction to
+  // GlobalCPUThread. Production multifeed_aggregator spends ~30-35% of
+  // CPU here, so we keep it on the hot path. Runs in parallel with
+  // the DLRM inference that was already kicked off above (inference
+  // executes on cpuThreadPool too, but folly's CPUThreadPoolExecutor
+  // is multi-threaded so the two stages overlap).
+  // story_contents is captured by value (copy) so its lifetime
+  // survives into the deferred GlobalCPUThread task.
+  folly::via(srvCPUThreadPool.get(),
+             [this_thread_ptr, story_contents, cpuPool]() {
+    return folly::via(cpuPool.get(),
+                      [this_thread_ptr, story_contents]() {
+      runFeatureExtraction(*this_thread_ptr, story_contents);
+      return folly::unit;
+    });
+  })
+      // Stage B: wait for DLRM inference to complete. inference_future
+      // was started before this chain so the wait is usually trivial
+      // (extraction is heavier). Hop onto ioThreadPool so the
+      // continuation that runs simulateIoOrFanout is on the IO loop.
+      .via(ioThreadPool.get())
+      .thenValue([inference_future = std::move(inference_future)](
+                     folly::Unit) mutable {
+        return std::move(inference_future);
+      })
       .via(ioThreadPool.get())
       .thenValue([this_thread_ptr, total_io_latency_ms, timekeeper, ioThreadPool](
                      int prediction_result) {