Channel keepalive eliminates cold-channel latency cliff (facebookresearch#740)

Summary:
Pull Request resolved: facebookresearch#740

Adds a per-MockServicesClient keepalive timer that fires a fire-and-forget `getStatus()` probe RPC every N milliseconds, keeping Rocket channels warm between sparse session bursts.

After D105903218 distributed outbound RPCs across one MockServicesClient per SREventBase (123 channels on BGM), the t25 QPS-latency sweep surfaced a severe cold-channel anti-pattern: p95 latency at low offered load was up to 14× *worse* than at peak load on BGM (9,488 ms at q=5/inst vs 684 ms at q=35/inst), with throughput collapsing to 31% of requested (1.55 of 5 QPS). Grace and CPL showed milder 2.5-2.8× cliffs.

Root cause (confirmed via perf-record diff at q=5 vs q=35 on BGM): at low offered load, ≈1.5 sessions in-flight × K=16 fanout = 24 RPCs spread across 123 channels — most channels idle for 100-300 ms between bursts. Each cold RPC then pays three compounding penalties: Rocket channel re-arm (+2.4pp Folly, +1.9pp RPC-AsyncIO at q=5), deep C-state wake on idle SREventBase cores (Zen4c C6 wake ≈100 µs; `poll_idle` + `acpi_processor_ffh_cstate_enter` in top 14 hot functions), and cold allocator/JIT caches (+4.1pp MemAlloc, +2.0pp JIT-Unresolved). All three penalties share the same proximate cause: channel idle time > 100 ms.

Touching every channel every 150 ms via a cheap getStatus() probe eliminates all three penalties simultaneously. The probe runs on the channel's own EventBase (thread-affine) and is fire-and-forget (drops the returned SemiFuture); transport errors are swallowed silently to avoid keepalive failures propagating to the application path.

Bandwidth cost: ~13K pings/sec/instance × ~100 bytes round-trip = ~1.3 MB/s — negligible. mock_services CPU cost ≈0.07 cores per instance.

Knobs:
- New CLI flag `--mock_keepalive_interval_ms` on LeafNodeRank (default 0 = disabled so the anti-pattern stays observable for regression testing).
- New `MOCK_KEEPALIVE_INTERVAL_MS` env override in run.sh so sweep scripts can A/B without rebuilding.
- Recommended starting value: 150 ms (validated). 300-500 ms likely sufficient; tunable.

Reviewed By: YifanYuan3

Differential Revision: D106558423

Author: excelle08

packages/feedsim/run.sh

@@ -719,6 +719,16 @@ main() {
         mock_services_opts="$mock_services_opts --use_legacy_sleep"
         echo "RPC fanout: forced OFF via LEAFNODE_USE_LEGACY_SLEEP=1 (legacy sleep path)"
     fi
+    # t25 mitigation knob: per-MockServicesClient keepalive ping. When
+    # MOCK_KEEPALIVE_INTERVAL_MS is set and > 0, each channel issues a
+    # 1-byte getStatus() probe every N ms to defeat the cold-channel
+    # anti-pattern observed at low QPS (BGM saw 14x p95 cliff at q=5).
+    # Recommended starting value: 150-500 ms. 0 / unset = disabled
+    # (anti-pattern stays observable).
+    if [ -n "${MOCK_KEEPALIVE_INTERVAL_MS:-}" ] && [ "${MOCK_KEEPALIVE_INTERVAL_MS}" != "0" ]; then
+        mock_services_opts="$mock_services_opts --mock_keepalive_interval_ms=${MOCK_KEEPALIVE_INTERVAL_MS}"
+        echo "MockServicesClient keepalive: ENABLED (interval=${MOCK_KEEPALIVE_INTERVAL_MS} ms)"
+    fi
 
     # OMP_NUM_THREADS=1: cap PyTorch's OpenMP parallel backend pool to
     # 1 thread. at::set_num_threads(1) only affects libtorch's native

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

@@ -321,7 +321,9 @@ void ThreadStartup(
             std::make_unique<ranking::MockServicesClient>(
                 evbs[i].get(),
                 args.mock_services_host_arg,
-                static_cast<uint16_t>(args.mock_services_port_arg)));
+                static_cast<uint16_t>(args.mock_services_port_arg),
+                std::chrono::milliseconds(
+                    args.mock_keepalive_interval_ms_arg)));
       } catch (const std::exception& e) {
         std::cerr << "Failed to connect to mock_services on "
                   << args.mock_services_host_arg << ":"
@@ -468,7 +470,9 @@ void ThreadStartup(
             std::make_unique<ranking::MockServicesClient>(
                 evbs[i].get(),
                 args.mock_services_host_arg,
-                static_cast<uint16_t>(args.mock_services_port_arg)));
+                static_cast<uint16_t>(args.mock_services_port_arg),
+                std::chrono::milliseconds(
+                    args.mock_keepalive_interval_ms_arg)));
       } catch (const std::exception& e) {
         std::cerr << "Failed to connect to mock_services on "
                   << args.mock_services_host_arg << ":"
@@ -2357,7 +2361,9 @@ int main(int argc, char** argv) {
     std::cout << "RPC fanout: enabled (target "
               << args.mock_services_host_arg << ":"
               << args.mock_services_port_arg
-              << ", scale=" << args.rpc_fanout_scale_arg << ")"
+              << ", scale=" << args.rpc_fanout_scale_arg
+              << ", keepalive_ms=" << args.mock_keepalive_interval_ms_arg
+              << ")"
               << std::endl;
   } else {
     std::cout << "RPC fanout: disabled (no --rpc_dist_path); using legacy"

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

@@ -96,3 +96,13 @@ option "rpc_fanout_scale" - "Scale factor applied to per-session fanout counts.
 # mock_services attempted). Lets later diffs in the stack run end-to-end
 # integration tests without depending on the mock_services side process.
 option "use_legacy_sleep" - "Force legacy folly::futures::sleep path; skip MockServicesClient construction and outbound RPC fanout." flag off
+
+# Mitigation knob for the t25 cold-channel anti-pattern: when offered load is
+# low, MockServicesClient channels go idle between sparse session bursts.
+# Per-channel keepalive issues a 1-byte getStatus() ping every N ms to keep
+# the Rocket channel + EventBase warm, defeating the cold-wake + deep-C-state
+# penalty observed at low QPS on BGM (p95 cliff: 14x at q=5). 0 = disabled
+# (default = keepalive off so the anti-pattern stays observable for regression
+# tests). Recommended: 150-500ms. See ~/.claude/plans/ancient-launching-pizza.md
+# section "2026-05-27 -- t25" for the root-cause analysis.
+option "mock_keepalive_interval_ms" - "Interval (ms) between per-MockServicesClient keepalive pings. 0 = disabled." int default="0"

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

@@ -29,8 +29,68 @@
 
 namespace ranking {
 
+// KeepaliveTimer fires a fire-and-forget getStatus() RPC every
+// keepalive_interval_ms to keep the underlying Rocket channel warm.
+//
+// Why this exists: t25 (2026-05-27) measured a 14x latency cliff on BGM at
+// low QPS, traced to cold MockServicesClient channels. After D105903218
+// each leaf thread owns one MockServicesClient per SREventBase (~123 on
+// BGM); at qps=5 most channels see no traffic for 100-300ms between
+// session bursts, then pay re-arm + deep C-state wake costs on the next
+// RPC. Per-channel keepalive defeats this by ensuring every channel sees
+// traffic at least every keepalive_interval_ms.
+//
+// Thread-affinity: scheduled on the same EventBase as the parent
+// MockServicesClient, so the callback runs on the right thread to call
+// dispatchByEnum() (which is thread-affine). Drops the resulting
+// SemiFuture immediately (fire-and-forget); transport errors are
+// swallowed silently — keepalive failures should not propagate to the
+// application path.
+class MockServicesClient::KeepaliveTimer : public folly::AsyncTimeout {
+ public:
+  KeepaliveTimer(
+      folly::EventBase* evb,
+      MockServicesClient* parent,
+      std::chrono::milliseconds interval)
+      : folly::AsyncTimeout(evb), parent_(parent), interval_(interval) {}
+
+  // Caller must invoke from the EventBase thread (or via
+  // runInEventBaseThread). scheduleTimeout itself is thread-affine.
+  void start() { scheduleTimeout(interval_); }
+
+  void timeoutExpired() noexcept override {
+    // Reschedule first so we don't drop the next tick if dispatch throws.
+    scheduleTimeout(interval_);
+
+    // Fire-and-forget getStatus() with latency_us=0 (lightest possible
+    // server-side handler). The 4-byte response header is the wire-
+    // contract minimum. We don't await — the returned SemiFuture goes
+    // out of scope but the underlying RPC remains in flight on the
+    // channel until completion, achieving the warming effect.
+    try {
+      std::string probe(4, '\0');
+      writeBigEndianResponseSize(probe.data(), 4);
+      parent_->client_->semifuture_getStatus(probe, /*latency_us=*/0)
+          .via(parent_->evb_)
+          .thenValue([](std::string&&) {})
+          .thenError(folly::tag_t<std::exception>{}, [](const std::exception&) {
+            // Swallow transport errors silently. The next iter will retry.
+          });
+    } catch (...) {
+      // Defensive — never let the timer callback propagate.
+    }
+  }
+
+ private:
+  MockServicesClient* parent_;
+  std::chrono::milliseconds interval_;
+};
+
 MockServicesClient::MockServicesClient(
-    folly::EventBase* evb, const std::string& host, uint16_t port)
+    folly::EventBase* evb,
+    const std::string& host,
+    uint16_t port,
+    std::chrono::milliseconds keepalive_interval)
     : evb_(evb) {
   if (evb_ == nullptr) {
     throw std::invalid_argument(
@@ -88,15 +148,34 @@ MockServicesClient::MockServicesClient(
     evb_->runInEventBaseThreadAndWait([this]() { client_.reset(); });
     std::rethrow_exception(result.exception().to_exception_ptr());
   }
+
+  // Start keepalive timer if requested. Must be constructed and started
+  // on the EventBase thread because AsyncTimeout is thread-affine.
+  if (keepalive_interval.count() > 0) {
+    evb_->runInEventBaseThreadAndWait([this, keepalive_interval]() {
+      keepalive_ =
+          std::make_unique<KeepaliveTimer>(evb_, this, keepalive_interval);
+      keepalive_->start();
+    });
+  }
 }
 
 MockServicesClient::~MockServicesClient() {
-  // The AsyncClient and its channel must be destroyed on the EventBase
-  // thread. Use runInEventBaseThreadAndWait to guarantee that even when
-  // the destructor runs from a different thread (e.g., during process
-  // shutdown).
-  if (evb_ != nullptr && client_) {
-    evb_->runInEventBaseThreadAndWait([this]() { client_.reset(); });
+  // The AsyncClient, its channel, and the keepalive timer are all
+  // thread-affine to evb_. Destroy them on that thread to guarantee
+  // correct cleanup even when the destructor runs from a different
+  // thread (e.g., during process shutdown). cancelTimeout must precede
+  // destruction or AsyncTimeout will assert.
+  if (evb_ != nullptr) {
+    evb_->runInEventBaseThreadAndWait([this]() {
+      if (keepalive_) {
+        keepalive_->cancelTimeout();
+        keepalive_.reset();
+      }
+      if (client_) {
+        client_.reset();
+      }
+    });
   }
 }
 

packages/feedsim/third_party/src/workloads/ranking/MockServicesClient.h

@@ -14,12 +14,14 @@
 
 #pragma once
 
+#include <chrono>
 #include <cstdint>
 #include <memory>
 #include <string>
 
 #include <folly/SocketAddress.h>
 #include <folly/futures/Future.h>
+#include <folly/io/async/AsyncTimeout.h>
 #include <folly/io/async/EventBase.h>
 
 #include "RpcDistRegistry.h"
@@ -66,7 +68,9 @@ class MockServicesClient {
   MockServicesClient(
       folly::EventBase* evb,
       const std::string& host,
-      uint16_t port);
+      uint16_t port,
+      std::chrono::milliseconds keepalive_interval =
+          std::chrono::milliseconds(0));
 
   ~MockServicesClient();
 
@@ -97,8 +101,15 @@ class MockServicesClient {
   folly::EventBase* getEventBase() const { return evb_; }
 
  private:
+  // KeepaliveTimer (defined in .cc) fires a fire-and-forget getStatus()
+  // probe RPC every keepalive_interval_ms to keep the underlying Rocket
+  // channel + SREventBase warm. Forward declared here so the unique_ptr
+  // member doesn't pull AsyncTimeout into the header's public surface.
+  class KeepaliveTimer;
+
   folly::EventBase* evb_;
   std::unique_ptr<mock_services::MockServiceAsyncClient> client_;
+  std::unique_ptr<KeepaliveTimer> keepalive_;
 };
 
 /**