Trace FeedSim's converged-QPS window instead of startup

packages/feedsim/run-feedsim-multi.sh

@@ -370,7 +370,8 @@ function analyze_and_print_results() {
     echo "    \"successful_instances\": ${successful_insts},"
     echo "    \"min_qps\": ${min_qps},"
     echo "    \"max_qps\": ${max_qps},"
-    echo "    \"is_fixed_qps\": ${IS_FIXED_QPS}"
+    echo "    \"is_fixed_qps\": ${IS_FIXED_QPS},"
+    echo "    \"dr_trace\": ${DCPERF_DR_TRACE:-0}"
     echo "}"
     if [[ "$(echo "${min_qps} < 0.8 * ${max_qps}" | bc)" = "1" ]]; then
         # ceil(max_req_qps)
@@ -391,8 +392,10 @@ else
     is_unstable_run="$?"
 fi
 
-# rerun this program with fixed qps if detecting high variance
-if [[ "$is_unstable_run" = 1 ]] && [[ "$IS_FIXED_QPS" = 0 ]] && [[ -z "$IS_RERUN" ]]; then
+# rerun this program with fixed qps if detecting high variance.
+# Variance check is skipped when tracing is on.
+if [[ "$is_unstable_run" = 1 ]] && [[ "$IS_FIXED_QPS" = 0 ]] && [[ -z "$IS_RERUN" ]] \
+   && [[ "${DCPERF_DR_TRACE:-}" != 1 ]]; then
     max_req_qps="$(cat /tmp/max_req_qps)"
     echo "Detected unstable run - rerunning with fixed QPS at ${max_req_qps}..."
     # shellcheck disable=SC2068

packages/feedsim/run.sh

@@ -897,6 +897,15 @@ main() {
     # libtorch may use) read OMP_NUM_THREADS directly. Without this,
     # each ThriftSrv.IO worker calling forward() spawns nproc OMP
     # threads, accumulating to nproc^2 GlobalCPUThread-named threads.
+    # search_qps.sh mirrors LeafNodeRank.cc's dr_trace_enabled_for_port() to
+    # decide which instance to trigger, and needs this instance's data port.
+    export FEEDSIM_PORT="$port"
+    if [ "${DCPERF_DR_TRACE:-}" = 1 ]; then
+        # trace_init()'s mkdir is not recursive; a missing parent means mkfifo
+        # fails and the run silently produces no trace.
+        mkdir -p "${DR_TRACE_OUTDIR:-/tmp/drmemtrace_out}"
+    fi
+
     # shellcheck disable=SC2086
     env $preload_env OMP_NUM_THREADS=1 MALLOC_CONF=narenas:20,dirty_decay_ms:5000 build/workloads/ranking/LeafNodeRank \
         --port="$port" \
@@ -1080,6 +1089,38 @@ main() {
     fi
 
     log_postprocessing_start "$BREAKDOWN_FOLDER" "$$"
+    if [ "${DCPERF_DR_TRACE:-}" = 1 ]; then
+        # Wait for the outdir to stop growing instead.
+        dr_out="${DR_TRACE_OUTDIR:-/tmp/drmemtrace_out}"
+        prev_size=-1
+        stable_for=0
+        waited=0
+        while [ "$waited" -lt "${DR_TRACE_FLUSH_TIMEOUT_SECONDS:-300}" ]; do
+            cur_size="$(du -sb "$dr_out" 2>/dev/null | cut -f1)"
+            if [ -z "$cur_size" ]; then
+                # du failed (outdir not created yet): not stable, keep waiting
+                # rather than exiting early and reporting an empty size.
+                stable_for=0
+            elif [ "$cur_size" = "$prev_size" ]; then
+                stable_for=$((stable_for + 5))
+                [ "$stable_for" -ge 15 ] && break
+            else
+                stable_for=0
+            fi
+            prev_size="$cur_size"
+            sleep 5
+            waited=$((waited + 5))
+        done
+        if [ "$stable_for" -ge 15 ]; then
+            echo "dr_trace: outdir stable at ${prev_size} bytes after ${waited}s"
+        else
+            last_size="$prev_size"
+            if [ -z "$last_size" ] || [ "$last_size" = "-1" ]; then
+                last_size="unreadable"
+            fi
+            echo "dr_trace: WARNING outdir still changing after ${waited}s (last size ${last_size}); trace may be truncated"
+        fi
+    fi
     sleep "$queue_drain_time" # wait for queue to drain
     kill -SIGINT $LEAF_PID || true > /dev/null # SIGINT so exits cleanly
     log_postprocessing_end "$BREAKDOWN_FOLDER" "$$"

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

@@ -311,6 +311,56 @@ collect_perf_record() {
     perf record -a -g -- sleep 5 >> /tmp/perf-record.log 2>&1
 }
 
+# Setup the pipe for the traced instance
+dr_trace_enabled_for_port() {
+    local port="${FEEDSIM_PORT:-}"
+    local multi_instance=0
+    if [ -n "${IS_AUTOSCALE_RUN:-}" ] && [ "${IS_AUTOSCALE_RUN}" -gt 1 ] 2>/dev/null; then
+        multi_instance=1
+    fi
+
+    if [ -n "${DR_TRACE_PORT:-}" ]; then
+        if [ "${DR_TRACE_PORT}" = "all" ]; then
+            [ "$multi_instance" = 0 ]
+            return
+        fi
+        [ "${DR_TRACE_PORT}" = "$port" ]
+        return
+    fi
+    if [ "$multi_instance" = 0 ]; then
+        return 0
+    fi
+    # Multi-instance with no port to check
+    [ -n "$port" ] && [ "$port" = "21212" ]
+}
+
+# Start the trace inside the converged-QPS window.
+trigger_dr_trace() {
+    [ "${DCPERF_DR_TRACE:-}" = 1 ] || return 0
+    if ! dr_trace_enabled_for_port; then
+        benchreps_tell_state "dr_trace: port ${FEEDSIM_PORT:-unknown} not selected, no trigger"
+        return 0
+    fi
+    local settle="${DR_TRACE_SETTLE_SECONDS:-30}"
+    local duration="${DR_TRACE_DURATION_SECONDS:-10}"
+    local fifo="${DR_TRACE_OUTDIR:-/tmp/drmemtrace_out}/dr_trace_trigger"
+    if [ $((settle + duration)) -ge "$experiment_time" ]; then
+        benchreps_tell_state "dr_trace: WARNING settle+duration ($settle+$duration) >= window $experiment_time"
+    fi
+    sleep "$settle"
+    if [ ! -p "$fifo" ]; then
+        benchreps_tell_state "dr_trace: no trigger fifo at $fifo, skipping"
+        return 0
+    fi
+    benchreps_tell_state "dr_trace: requesting ${duration}s trace via $fifo"
+    if timeout "${DR_TRACE_TRIGGER_TIMEOUT_SECONDS:-60}" \
+            bash -c 'printf "%s\n" "$1" > "$2"' _ "$duration" "$fifo"; then
+        benchreps_tell_state "dr_trace: trigger delivered"
+    else
+        benchreps_tell_state "dr_trace: trigger write timed out (no reader), continuing"
+    fi
+}
+
 # Initialize our own variables:
 experiment_time=120
 wait_time=5
@@ -537,6 +587,7 @@ if [[ -n "$fixed_qps" ]]; then
     if [ "${DCPERF_PERF_RECORD}" = 1 ] && ! [ -f "perf.data" ]; then
         collect_perf_record &
     fi
+    trigger_dr_trace &
     # $main_operation_name is defined in runtime_breakdown_utils.sh
     run_loadtest measured_qps measured_latency $fixed_qps "$main_operation_name"
 
@@ -712,7 +763,7 @@ if [[ -n "$IS_AUTOSCALE_RUN" ]] && [[ "$IS_AUTOSCALE_RUN" -gt 1 ]]; then
     num_ready_inst=$(grep --count "after gap_qps" $BREPS_LFILE)
     if [[ $num_ready_inst -lt $NUM_INSTANCES ]]; then
         result_filename=$(basename "$output_csv_file")
-        current_inst_num=$(echo "$result_filename" | sed -E "s/feedsim_results_([0-9]+).txt/\1/g")
+        current_inst_num=$(echo "$result_filename" | sed -E "s/feedsim_results_(fixqps-)?([0-9]+)\.txt/\2/")
         benchreps_tell_state "[Instance $current_inst_num] Waiting for other instances to finish \"gap_qps\" stage."
         while [[ $num_ready_inst -lt $NUM_INSTANCES ]]; do
             sleep 1
@@ -731,6 +782,7 @@ experiment_time=$final_experiment_time
 if [ "${DCPERF_PERF_RECORD}" = 1 ] && ! [ -f "perf.data" ]; then
     collect_perf_record &
 fi
+trigger_dr_trace &
 run_loadtest measured_qps measured_latency $cur_qps "$main_operation_name"
 printf "final requested_qps = %.2f, measured_qps = %.2f, latency = %.2f\n" $cur_qps $measured_qps $measured_latency
 

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

@@ -83,11 +83,26 @@ static constexpr uint32_t kDrTraceDurationSeconds = 10;
 static constexpr int kFeedsimBasePort = 21212;
 
 static bool dr_trace_enabled_for_port(int port) {
+  const char* autoscale = std::getenv("IS_AUTOSCALE_RUN");
+  const bool multi_instance = autoscale != nullptr && std::atoi(autoscale) > 1;
   const char* only_port = std::getenv("DR_TRACE_PORT");
-  if (only_port != nullptr) {
-    return std::strcmp(only_port, "all") == 0 || std::atoi(only_port) == port;
+  // Empty-but-set is treated as unset, matching the shell mirror's [ -n ].
+  if (only_port != nullptr && only_port[0] != '\0') {
+    if (std::strcmp(only_port, "all") == 0) {
+      // Each instance derives the trigger pipe path from the shared outdir, so
+      // tracing all of them would have them fight over one pipe.
+      return !multi_instance;
+    }
+    return std::atoi(only_port) == port;
   }
-  return std::getenv("IS_AUTOSCALE_RUN") == nullptr || port == kFeedsimBasePort;
+  return !multi_instance || port == kFeedsimBasePort;
+}
+
+// Trigger mode. "pipe" (default) blocks on $DR_TRACE_OUTDIR/dr_trace_trigger so
+// search_qps.sh can start the trace during the converged-QPS window
+static const char* dr_trace_trigger_mode() {
+  const char* mode = std::getenv("DR_TRACE_TRIGGER");
+  return mode != nullptr ? mode : "pipe";
 }
 
 // Force -raw_compress none for LeafNodeRank only.
@@ -2995,12 +3010,20 @@ int main(int argc, char** argv) {
   debug_dump_thread.detach();
 
 #ifdef DR_TRACE_INCLUDED
+  // Establish the pipe for tracing
   if (dr_trace_enabled_for_port(args.port_arg)) {
-    dr_trace_disable_raw_compression();
-    trace_configure_env();
-    std::thread(
-        trace_execution_delay<kDrTraceDelaySeconds, kDrTraceDurationSeconds>)
-        .detach();
+    const char* trigger_mode = dr_trace_trigger_mode();
+    if (std::strcmp(trigger_mode, "off") != 0) {
+      dr_trace_disable_raw_compression();
+      trace_configure_env();
+      if (std::strcmp(trigger_mode, "delay") == 0) {
+        std::thread(
+            trace_execution_delay<kDrTraceDelaySeconds, kDrTraceDurationSeconds>)
+            .detach();
+      } else {
+        std::thread(trace_execution_pipe).detach();
+      }
+    }
   }
 #endif