From 34972ac6734bac7c665c16d056ad5e76f8600464 Mon Sep 17 00:00:00 2001 From: Kaiwei Tu Date: Tue, 18 Aug 2026 12:08:03 -0700 Subject: [PATCH] Trace FeedSim's converged-QPS window instead of startup Reviewed By: suyashmahar Differential Revision: D115961975 --- packages/feedsim/run-feedsim-multi.sh | 9 ++-- packages/feedsim/run.sh | 41 ++++++++++++++ .../third_party/src/scripts/search_qps.sh | 54 ++++++++++++++++++- .../src/workloads/ranking/LeafNodeRank.cc | 39 +++++++++++--- 4 files changed, 131 insertions(+), 12 deletions(-) diff --git a/packages/feedsim/run-feedsim-multi.sh b/packages/feedsim/run-feedsim-multi.sh index e7ac693cb..cfa97f581 100755 --- a/packages/feedsim/run-feedsim-multi.sh +++ b/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 diff --git a/packages/feedsim/run.sh b/packages/feedsim/run.sh index 561b669e8..c716fc460 100644 --- a/packages/feedsim/run.sh +++ b/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" "$$" diff --git a/packages/feedsim/third_party/src/scripts/search_qps.sh b/packages/feedsim/third_party/src/scripts/search_qps.sh index d6111b4d9..64582a26f 100755 --- a/packages/feedsim/third_party/src/scripts/search_qps.sh +++ b/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 diff --git a/packages/feedsim/third_party/src/workloads/ranking/LeafNodeRank.cc b/packages/feedsim/third_party/src/workloads/ranking/LeafNodeRank.cc index d78c2ef85..762f8afed 100644 --- a/packages/feedsim/third_party/src/workloads/ranking/LeafNodeRank.cc +++ b/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) - .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) + .detach(); + } else { + std::thread(trace_execution_pipe).detach(); + } + } } #endif