Adaptive driver depth in search_qps (fleet default) to fix the throughput anti-pattern

Reviewed By: YifanYuan3

Differential Revision: D112694748

Author: excelle08

packages/feedsim/run.sh

@@ -834,7 +834,11 @@ main() {
     # client is launched through search_qps.sh, which does not reliably forward
     # the parent's exported env to the driver process, so FEEDSIM_DRIVER_TLS is
     # also injected directly on the driver command line via `driver_bin`.
-    driver_bin="build/workloads/ranking/DriverNodeRank"
+    # Per-run driver knobs are injected on the DriverNodeRank command line via
+    # `env VAR=val` (accumulated in driver_env), because search_qps.sh launches
+    # the driver as a bare `$command &` that does not reliably inherit the
+    # parent shell's exported env.
+    driver_env=""
     if [ "${FEEDSIM_DRIVER_TLS:-0}" = "1" ]; then
         driver_cert_dir="${FEEDSIM_ROOT}/certs"
         if [ ! -r "${driver_cert_dir}/example.crt" ] || [ ! -r "${driver_cert_dir}/example.key" ]; then
@@ -844,9 +848,22 @@ main() {
         export FEEDSIM_TLS_CERT="${driver_cert_dir}/example.crt"
         export FEEDSIM_TLS_KEY="${driver_cert_dir}/example.key"
         export FEEDSIM_DRIVER_TLS=1
-        driver_bin="env FEEDSIM_DRIVER_TLS=1 build/workloads/ranking/DriverNodeRank"
+        driver_env="${driver_env} FEEDSIM_DRIVER_TLS=1"
         echo "Driver↔Leaf TLS: ENABLED (cert=${driver_cert_dir}/example.crt, AES-GCM)"
     fi
+    # FEEDSIM_STATS_WARMUP_SECS: drop the first N seconds of latency/throughput
+    # samples in each search_qps probe (DriverNodeRank resets its stats N secs
+    # in) so cold-start transients don't inflate the tail and make the search
+    # back off QPS prematurely. Default unset/0 (no warmup).
+    if [ -n "${FEEDSIM_STATS_WARMUP_SECS:-}" ] && [ "${FEEDSIM_STATS_WARMUP_SECS}" != "0" ]; then
+        driver_env="${driver_env} FEEDSIM_STATS_WARMUP_SECS=${FEEDSIM_STATS_WARMUP_SECS}"
+        echo "Driver stats warmup: ${FEEDSIM_STATS_WARMUP_SECS}s (dropping cold-start samples per probe)"
+    fi
+    if [ -n "$driver_env" ]; then
+        driver_bin="env${driver_env} build/workloads/ranking/DriverNodeRank"
+    else
+        driver_bin="build/workloads/ranking/DriverNodeRank"
+    fi
     # MOCK_ZSTD_FRAC env consumed by MockServicesClient::resolveZstdFraction.
     # Always export so the t43 c7 default 0.75 reaches the binary.
     export MOCK_ZSTD_FRAC="$mock_zstd_frac"
@@ -979,16 +996,30 @@ main() {
     # Override via --sla-p95-ms CLI flag (handled in arg parsing above).
     sla_arg="95p:${sla_p95_ms}"
 
+    # Adaptive driver depth (fleet default): search_qps raises the driver's
+    # pipeline --depth in the peak phase until the server saturates (system
+    # CPU>=95% or p95>=SLA), giving each platform just enough offered concurrency
+    # to reach a real bound instead of capping on driver concurrency (the t19
+    # anti-pattern where big boxes sat at ~80% CPU with p95 far below SLA).
+    # Enabled by default up to depth 8; set FEEDSIM_ADAPTIVE_DEPTH_MAX=0 to
+    # disable and fall back to the fixed FEEDSIM_DRIVER_DEPTH (default 1).
+    sqps_adaptive_arg=""
+    adaptive_depth_max="${FEEDSIM_ADAPTIVE_DEPTH_MAX:-8}"
+    if [ "$adaptive_depth_max" != "0" ]; then
+        sqps_adaptive_arg="-D ${adaptive_depth_max}"
+    fi
+
     if [ -z "$fixed_qps" ] && [ "$auto_driver_threads" != "1" ]; then
         benchreps_tell_state "before search_qps"
         echo "search_qps SLA: ${sla_arg}"
         # shellcheck disable=SC2086
-        scripts/search_qps.sh -w 15 -f 300 -s "$sla_arg" -P "$LEAF_PID" -B "$BREAKDOWN_FOLDER" $qps_threshold_args $no_retry_args -o "${FEEDSIM_ROOT}/${result_filename}" -- \
+        scripts/search_qps.sh -t "${FEEDSIM_EXPERIMENT_TIME:-120}" -w 15 -f 300 -s "$sla_arg" $sqps_adaptive_arg -P "$LEAF_PID" -B "$BREAKDOWN_FOLDER" $qps_threshold_args $no_retry_args -o "${FEEDSIM_ROOT}/${result_filename}" -- \
             $driver_bin \
                 --server "0.0.0.0:$port" \
                 --monitor_port "$client_monitor_port" \
                 --threads="${driver_threads}" \
                 --connections=4 \
+                --depth="${FEEDSIM_DRIVER_DEPTH:-1}" \
                 $client_feature_opts \
                 $silesia_opts \
                 $req_size_opts
@@ -997,10 +1028,11 @@ main() {
         benchreps_tell_state "before search_qps"
         echo "search_qps SLA: ${sla_arg}"
         # shellcheck disable=SC2086
-        scripts/search_qps.sh -a -w 15 -f 300 -s "$sla_arg" -P "$LEAF_PID" -B "$BREAKDOWN_FOLDER" $qps_threshold_args $no_retry_args -o "${FEEDSIM_ROOT}/${result_filename}" -- \
+        scripts/search_qps.sh -a -t "${FEEDSIM_EXPERIMENT_TIME:-120}" -w 15 -f 300 -s "$sla_arg" $sqps_adaptive_arg -P "$LEAF_PID" -B "$BREAKDOWN_FOLDER" $qps_threshold_args $no_retry_args -o "${FEEDSIM_ROOT}/${result_filename}" -- \
             $driver_bin \
                 --monitor_port "$client_monitor_port" \
                 --server "0.0.0.0:$port" \
+                --depth="${FEEDSIM_DRIVER_DEPTH:-1}" \
                 $client_feature_opts \
                 $silesia_opts \
                 $req_size_opts
@@ -1029,6 +1061,7 @@ main() {
                 --monitor_port "$client_monitor_port" \
                 --threads="${num_workers}" \
                 --connections="${num_connections}" \
+                --depth="${FEEDSIM_DRIVER_DEPTH:-1}" \
                 $client_feature_opts \
                 $silesia_opts \
                 $req_size_opts

packages/feedsim/third_party/src/scripts/search_qps.sh

@@ -31,6 +31,26 @@ echo "${SCRIPT_NAME}: DCPERF_PERF_RECORD=${DCPERF_PERF_RECORD}"
 function benchreps_tell_state () {
     date +"%Y-%m-%d_%T ${1}" >> $BREPS_LFILE
 }
+
+# ─── CPU utilization helpers (used by adaptive depth) ────────────────────────
+# Read /proc/stat's aggregate cpu line and echo "total idle_all" jiffies.
+cpu_snapshot() {
+  local cpu u n s idle iow irq sirq st rest
+  read -r cpu u n s idle iow irq sirq st rest < /proc/stat
+  local idle_all=$((idle + iow))
+  local total=$((u + n + s + idle + iow + irq + sirq + st))
+  echo "$total $idle_all"
+}
+# System-wide CPU busy% (100 - idle%) measured over the next $1 seconds.
+cpu_busy_over() {
+  local secs="$1" s1 s2 t1 i1 t2 i2 dt di
+  s1=$(cpu_snapshot); t1=${s1% *}; i1=${s1#* }
+  sleep "$secs"
+  s2=$(cpu_snapshot); t2=${s2% *}; i2=${s2#* }
+  dt=$((t2 - t1)); di=$((i2 - i1))
+  if [ "$dt" -le 0 ]; then echo "0"; return; fi
+  echo "scale=1; (($dt - $di) * 100) / $dt" | bc
+}
 # Source runtime breakdown utilities if they exist
 if [ -f "${BENCHPRESS_ROOT}/packages/common/runtime_breakdown_utils.sh" ]; then
     source "${BENCHPRESS_ROOT}/packages/common/runtime_breakdown_utils.sh"
@@ -99,6 +119,11 @@ mutilate (EuroSys \'14) [https://github.com/leverich/mutilate]
                 without retrying. Optional
     -P          PID of the process to log runtime breakdowns. Optional
     -B          Folder to log runtime breakdowns. Optional
+    -D          Adaptive depth: max driver pipeline depth. When set, the peak
+                phase raises the driver's --depth from 1 until the server is
+                saturated (system CPU >= 95% OR achieved p95 >= SLA), then holds
+                that depth for the QPS search. Overrides any --depth in the
+                driver command. Optional.
 EOF
 }
 
@@ -145,7 +170,7 @@ run_loadtest() {
   for r in $(seq 1 $load_test_retries); do
     # run the command, saving result to tmpfile
     local tmp_file=$(mktemp)
-    $command $threads_arg $qps_arg &>$tmp_file &
+    $command $threads_arg $qps_arg $adaptive_depth_arg &>$tmp_file &
     LOADTEST_PID=$!
 
     if [ "$no_retry_mode" = "1" ]; then
@@ -301,14 +326,19 @@ max_warmup_iterations=10
 no_retry_mode=""
 breakdown_pid=""
 breakdown_folder=""
+adaptive_depth_max=""   # -D: when set, search_qps raises driver --depth in the
+adaptive_depth_arg=""   # peak phase until the server saturates (CPU>=95% or p95>=SLA)
 
 OPTIND=1 # Reset is necessary if getopts was used previously in the script.  It is a good idea to make this local in a function.
-while getopts "ht:f:w:m:s:q:ao:r:x:NP:B:" opt; do
+while getopts "ht:f:w:m:s:q:ao:r:x:NP:B:D:" opt; do
   case "$opt" in
     h)
       show_help
       exit 0
       ;;
+    D)
+      adaptive_depth_max=$OPTARG
+      ;;
     t)
       experiment_time=$OPTARG
       ;;
@@ -364,6 +394,14 @@ fi
 # remaining argument is loadtest command
 command=$@
 
+# In adaptive-depth mode, search_qps owns the driver's --depth: strip any fixed
+# --depth from the command so our per-attempt --depth is the only one, and start
+# the peak search at depth=1.
+if [ -n "$adaptive_depth_max" ]; then
+  command=$(echo "$command" | sed -E 's/[[:space:]]*--depth=[0-9]+//g')
+  adaptive_depth_arg="--depth=1"
+fi
+
 # make sure latency_type and latency_target are specified
 if [[ -z "$fixed_qps" ]] && ( [[ $latency_type = "" ]] || [[ $latency_target = "" ]] ); then
   echo 'error: -s metric:target must be specified' >&2; exit 1
@@ -513,7 +551,40 @@ fi
 
 # find peak QPS
 benchreps_tell_state "before peak_qps"
-run_loadtest peak_qps measured_latency "" ""
+if [ -n "$adaptive_depth_max" ]; then
+  # Adaptive depth: the peak load test offers at most threads*connections*depth
+  # concurrent requests. Starting at depth=1, keep raising depth (and re-running
+  # peak) until the server saturates — system CPU >= 95% OR achieved p95 >= SLA
+  # — so platforms that need more offered concurrency reach a real bound instead
+  # of capping on driver concurrency (the t19 anti-pattern). The selected depth
+  # is then held for the QPS search / tuning / final phases.
+  cur_depth=1
+  while : ; do
+    adaptive_depth_arg="--depth=$cur_depth"
+    # Sample system CPU busy% over a mid-run window while the peak load runs.
+    cpu_busy_file="/tmp/adaptive_cpu_busy_$$"
+    ( sleep 20; cpu_busy_over 40 > "$cpu_busy_file" ) &
+    cpu_sampler_pid=$!
+    run_loadtest peak_qps measured_latency "" ""
+    wait "$cpu_sampler_pid" 2>/dev/null
+    cpu_busy=$(cat "$cpu_busy_file" 2>/dev/null || echo 0)
+    rm -f "$cpu_busy_file"
+    cpu_sat=$(echo "${cpu_busy:-0} >= 95" | bc 2>/dev/null || echo 0)
+    lat_sat=$(echo "$measured_latency >= $latency_target" | bc 2>/dev/null || echo 0)
+    printf "adaptive-depth: depth=%d peak_qps=%.2f p95=%.2f cpu_busy=%s%% cpu_sat=%s lat_sat=%s\n" \
+      "$cur_depth" "$peak_qps" "$measured_latency" "${cpu_busy:-0}" "$cpu_sat" "$lat_sat"
+    echo "adaptive-depth: depth=$cur_depth peak_qps=$peak_qps p95=$measured_latency cpu_busy=${cpu_busy}% cpu_sat=$cpu_sat lat_sat=$lat_sat" >> $BREPS_LFILE
+    if [ "$cpu_sat" -eq 1 ] || [ "$lat_sat" -eq 1 ] || [ "$cur_depth" -ge "$adaptive_depth_max" ]; then
+      break
+    fi
+    cur_depth=$((cur_depth + 1))
+    sleep "$wait_time"
+  done
+  echo "adaptive-depth: SELECTED depth=$cur_depth (cpu_busy=${cpu_busy}%, p95=$measured_latency, sla=$latency_target)" >> $BREPS_LFILE
+  printf "adaptive-depth: selected depth=%d (cpu_busy=%s%%, p95=%.2f)\n" "$cur_depth" "${cpu_busy:-0}" "$measured_latency"
+else
+  run_loadtest peak_qps measured_latency "" ""
+fi
 printf "peak qps = %.2f, latency = %.2f\n" $peak_qps $measured_latency
 benchreps_tell_state "after peak_qps"
 

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

@@ -864,6 +864,17 @@ void FeedSimDriver::enableMonitoring(uint16_t port) {
   impl_->monitor_port = port;
 }
 
+namespace {
+// One-shot timer callback used by FEEDSIM_STATS_WARMUP_SECS: drops all
+// latency/throughput samples collected so far so the final stats reflect only
+// the post-warmup (steady-state) window. Runs on the DriverThread's own event
+// base, so it never races that thread's logRequest()/logResponse() writes to
+// the same DriverStats.
+void resetDriverStatsCb(evutil_socket_t, short, void* arg) {
+  reinterpret_cast<DriverStats*>(arg)->reset();
+}
+} // namespace
+
 void FeedSimDriver::run(uint32_t num_threads, bool thread_pinning,
                         uint32_t num_connections_per_thread,
                         uint32_t max_connection_depth) {
@@ -876,6 +887,23 @@ void FeedSimDriver::run(uint32_t num_threads, bool thread_pinning,
   impl_->running = true;
   impl_->total_stats = std::make_unique<DriverStats>(1000);
 
+  // FEEDSIM_STATS_WARMUP_SECS: when >0, each DriverThread drops the samples it
+  // collected in the first N seconds of the measurement window (a per-thread
+  // one-shot timer calls DriverStats::reset()). This excludes cold-start
+  // transients (connection ramp, cache/JIT warmup, first-touch faults) from
+  // the reported latency distribution, giving search_qps a more stable p95 so
+  // it doesn't back off QPS on a noisy tail (the t19 anti-pattern).
+  int stats_warmup_secs = 0;
+  {
+    const char* e = std::getenv("FEEDSIM_STATS_WARMUP_SECS");
+    if (e != nullptr && e[0] != '\0') {
+      stats_warmup_secs = std::atoi(e);
+      if (stats_warmup_secs < 0) {
+        stats_warmup_secs = 0;
+      }
+    }
+  }
+
   // Barrier for thread init synchronization
   pthread_barrier_t init_barrier;
   pthread_barrier_init(&init_barrier, nullptr, num_threads + 1);
@@ -921,7 +949,7 @@ void FeedSimDriver::run(uint32_t num_threads, bool thread_pinning,
 
     // Start thread
     dt->thread = std::thread([this, &dt_ref = *dt, &init_barrier,
-                              thread_pinning, i]() {
+                              thread_pinning, i, stats_warmup_secs]() {
       // CPU affinity
       if (thread_pinning) {
         cpu_set_t mask;
@@ -953,6 +981,18 @@ void FeedSimDriver::run(uint32_t num_threads, bool thread_pinning,
       // Start making requests
       TestDriver::Impl::makeRequests(*dt_ref.driver);
 
+      // Schedule the cold-start stats reset on this thread's own base (added
+      // here, before dispatch, so it fires from this thread — no data race with
+      // the stats writes). Fires once ~stats_warmup_secs into steady traffic.
+      if (stats_warmup_secs > 0) {
+        struct timeval warmup_tv {
+          stats_warmup_secs, 0
+        };
+        event_base_once(
+            dt_ref.base, -1, EV_TIMEOUT, resetDriverStatsCb,
+            &dt_ref.driver->impl_->current_stats, &warmup_tv);
+      }
+
       // Run event loop
       event_base_dispatch(dt_ref.base);
     });
@@ -1070,7 +1110,14 @@ void FeedSimDriver::run(uint32_t num_threads, bool thread_pinning,
   }
 
   double end_time = getTimeSec();
-  double elapsed = end_time - start_time;
+  // With a stats warmup, samples were reset ~stats_warmup_secs into the run, so
+  // the throughput denominator must be the post-warmup window (else QPS would
+  // be understated by counting post-warmup queries over the full duration).
+  double meas_start = start_time;
+  if (stats_warmup_secs > 0) {
+    meas_start = start_time + stats_warmup_secs;
+  }
+  double elapsed = end_time - meas_start;
 
   // Aggregate stats from all threads
   for (auto& dt : impl_->threads) {