[None][feat] cache transceiver test in Perf sanity

examples/disaggregated/slurm/cache_transceiver_test/report.py

@@ -227,22 +227,34 @@ def _parse_cpp_recv_csvs(csv_dir):
 
 
 def _parse_python_csvs(csv_dir):
-    """Return {rid: [per_rank_GBps, ...]} from perf_logger py_*_*.csv files.
+    """Return {rid: [per_rank_GBps, ...]} from Python perf_logger task CSVs.
+
+    PerfLogManager names its per-task CSV by priority of the env vars it sees:
+    with TRTLLM_KVCACHE_TIME_OUTPUT_PATH set (which this harness always sets
+    for the C++ transceiver), files are "{dir}/{instanceUuid}_{rank}.csv";
+    only the legacy TLLM_KV_TRANSFER_PERF_LOG_FILE fallback produces the old
+    "py_{instanceUuid}_{rank}.csv" prefix. So identify Python task CSVs by
+    their header columns (unique_rid + throughput_mbs) rather than by file
+    name -- C++ send/recv CSVs and the gen_transfer_summary CSV have neither
+    column and are skipped.
 
     `throughput_mbs` is recorded for send tasks in MiB/s (perf_logger divides by
     1024*1024). We convert MiB/s -> GB/s (`* 1024^2 / 1e9`) so it matches the
     (converted) C++ bandwidth, and return the per-rank list (NOT a sum) -- see
     `_parse_cpp_recv_csvs`.
     """
     per_rid = {}  # rid -> list[per-rank throughput GB/s]
-    for path in glob.glob(os.path.join(csv_dir, "py_*_*.csv")):
+    for path in glob.glob(os.path.join(csv_dir, "*.csv")):
         with open(path) as f:
             reader = csv.DictReader(f)
+            fields = reader.fieldnames or []
+            if "unique_rid" not in fields or "throughput_mbs" not in fields:
+                continue  # not a Python perf_logger task CSV (e.g. C++ send/recv)
             for row in reader:
-                if "unique_rid" not in row or "throughput_mbs" not in row:
-                    continue
-                if "Send" not in row.get("task_type", ""):
-                    continue  # throughput is meaningful only for send tasks
+                if row.get("task_type") != "KVSendTask":
+                    continue  # only KV sends carry meaningful throughput;
+                    # AuxSendTask rows are tiny metadata sends that would
+                    # drag the median down, KVRecvTask rows have none
                 try:
                     rid = int(float(row["unique_rid"]))
                     mbs = float(row["throughput_mbs"])

examples/disaggregated/slurm/cache_transceiver_test/run_cache_transceiver_test.py

@@ -25,7 +25,8 @@
 received blocks regenerate to the same pattern. Bandwidth is emitted by the
 transceivers themselves into per-rank CSVs (parsed later by report.py):
   C++  -> TRTLLM_KVCACHE_TIME_OUTPUT_PATH (<instanceId>_*_send.csv / <instanceId>_*_recv.csv)
-  Py   -> TLLM_KV_TRANSFER_PERF_LOG_FILE  (py_*_*.csv, throughput_mbs)
+  Py   -> same env var (PerfLogManager gives it top priority):
+          <instanceUuid>_<rank>.csv, throughput_mbs column
 
 This driver mirrors the single-process test (tests/unittest/others/
 test_kv_cache_transceiver.py) and the multi-process Python test
@@ -487,12 +488,12 @@ def main():
     work_dir = cfg["environment"]["work_dir"]
     csv_dir = os.path.join(work_dir, "csv", str(sweep), role)
     os.makedirs(csv_dir, exist_ok=True)
+    # One env var drives both transceivers: C++ caches it on first read, and
+    # Python's PerfLogManager gives it top priority (enabling perf logging and
+    # writing task CSVs as <instanceUuid>_<rank>.csv in csv_dir; the
+    # per-transceiver UUID avoids cross-combination collisions, and report.py
+    # identifies these files by header columns, not name).
     os.environ["TRTLLM_KVCACHE_TIME_OUTPUT_PATH"] = csv_dir
-    # Python perf logging (singleton reads these once; C++ ignores them). Set at
-    # startup so it is enabled regardless of combination ordering. Per-transceiver UUID
-    # filenames (py_<uuid>_<rank>.csv) avoid cross-combination collisions.
-    os.environ["TLLM_ENABLE_CACHE_TRANSFER_PERF_INFO"] = "1"
-    os.environ["TLLM_KV_TRANSFER_PERF_LOG_FILE"] = os.path.join(csv_dir, "py")
 
     status_dir = os.path.join(work_dir, "status")
     os.makedirs(status_dir, exist_ok=True)

jenkins/L0_Test.groovy

@@ -2043,6 +2043,8 @@ def runLLMTestlistWithSbatch(pipeline, platform, testList, config=VANILLA_CONFIG
                         echo "Pytest succeed in Slurm job \$jobId"
                     else
                         echo "Pytest failed in Slurm job \$jobId"
+                        echo "Full test output (logs not shown above) is uploaded after stage teardown to:"
+                        echo "  https://urm.nvidia.com/artifactory/${UPLOAD_PATH}/test-results/results-${stageName}${postTag}.tar.gz"
                     fi
                     echo "Status: \$STATUS | Exit_code \$EXIT_CODE"
                     exit 0

jenkins/scripts/perf/disaggregated/slurm_ct_precheck_gate.sh

@@ -0,0 +1,180 @@
+# Cache-transceiver precheck gate for the disaggregated perf-sanity launch
+# script. submit.py splices this file into the generated launch script ahead
+# of slurm_launch_draft.sh, which calls run_cache_transceiver_precheck after
+# computing the per-server node slices. Kept as functions in a separate file
+# so the gate logic can be sourced and exercised standalone.
+#
+# Expects the launch-script globals: srunArgs, numGenServers, numCtxServers,
+# nodesPerGenServer/nodesPerCtxServer, gpusPerNodePerGenServer/
+# gpusPerNodePerCtxServer, genNodeLists/ctxNodeLists, testOutputDir,
+# jobWorkspace, pytestCommandCTXPrecheck/pytestCommandGENPrecheck,
+# precheckRunScript, ctPrecheckEnabled, ctPrecheckTimeout, stageName,
+# and the cleanup_on_failure function.
+
+# Escape text for embedding in XML; junit parsers also choke on raw control
+# bytes that MPI/UCX logs may contain, so strip those too.
+ct_xml_escape() {
+    sed -e 's/&/\&amp;/g' -e 's/</\&lt;/g' -e 's/>/\&gt;/g' -e 's/"/\&quot;/g' \
+        | tr -d '\000-\010\013\014\016-\037'
+}
+
+# First few root-cause-shaped lines of a step log — the tail alone can miss
+# the real error when it happened early and retry spam follows.
+ct_first_errors() {
+    grep -m 5 -nE "Traceback \(most recent call last\)|MPI_ABORT|MPIR_Err|srun: error|Segmentation fault|CUDA error|RuntimeError|AssertionError|INIT_ERROR|TRANSFER_ERROR" \
+        "$1" 2>/dev/null || true
+}
+
+# Shared verdict predicate + failing-log excerpt, so the console summary and
+# the junit xml cannot drift apart.
+ct_step_passed() {
+    local statusFile="$precheckDir/status/$1.status"
+    [ -f "$statusFile" ] && grep -q "^PASS" "$statusFile"
+}
+
+ct_step_excerpt() {
+    local stepLog="$precheckDir/logs/$1.log"
+    echo "First error lines (line-numbered):"
+    ct_first_errors "$stepLog"
+    echo "Log tail ($stepLog):"
+    tail -n 60 "$stepLog" 2>/dev/null || true
+}
+
+# Console summary for a failed precheck: per-instance verdicts, first error
+# lines + tail of each failing step log, and UCX red-flag lines.
+# Uses: precheckDir, precheckNames.
+ct_print_failure_summary() {
+    echo "===================================================================="
+    echo "CACHE TRANSCEIVER PRECHECK FAILED - the disaggregated test will NOT run"
+    echo "Instance verdicts:"
+    cat "$precheckDir"/status/*.status 2>/dev/null \
+        || echo "(no status files - steps died before transceiver setup; see logs below)"
+    echo ""
+    echo "Failing step logs:"
+    for k in "${!precheckNames[@]}"; do
+        ct_step_passed "${precheckNames[$k]}" && continue
+        echo "----- ${precheckNames[$k]} -----"
+        ct_step_excerpt "${precheckNames[$k]}"
+    done
+    echo ""
+    echo "UCX red flags (host-staged tcp fallback / UCX errors), if any:"
+    grep -hE "sw-emul|UCX +(ERROR|WARN)" "$precheckDir"/logs/*.log 2>/dev/null | sort -u | head -20 || true
+    echo "Full artifacts: $precheckDir (status/*.json, logs/, csv/)"
+    echo "===================================================================="
+}
+
+# Synthetic junit result so the failure shows up as a test entry in the
+# Jenkins test report: uploadResults scps $jobWorkspace/results*.xml back and
+# junit() ingests them. Best-effort — callers must not let this mask the real
+# failure path. Uses: precheckDir, precheckNames, jobWorkspace, stageName.
+ct_write_junit_xml() {
+    local junitXml="$jobWorkspace/results-ct-precheck.xml"
+    local suiteName
+    suiteName="$(printf '%s' "${stageName:-${SLURM_JOB_NAME:-disagg_perf_sanity}}" | ct_xml_escape)"
+    local junitFailures=0
+    local junitCases=""
+    local name verdict detail
+    for k in "${!precheckNames[@]}"; do
+        name="${precheckNames[$k]}"
+        if ct_step_passed "$name"; then
+            junitCases+="<testcase name=\"cache_transceiver_precheck[$name]\" classname=\"$suiteName\" time=\"0\"/>"$'\n'
+            continue
+        fi
+        junitFailures=$((junitFailures + 1))
+        verdict="$( (head -n 1 "$precheckDir/status/$name.status" 2>/dev/null \
+            || echo "NO_STATUS: step died before writing a verdict (see log)") | ct_xml_escape)"
+        detail="$(ct_step_excerpt "$name" | ct_xml_escape)"
+        junitCases+="<testcase name=\"cache_transceiver_precheck[$name]\" classname=\"$suiteName\" time=\"0\">"
+        junitCases+="<failure message=\"$verdict\">$detail</failure></testcase>"$'\n'
+    done
+    if {
+        echo '<?xml version="1.0" encoding="UTF-8"?>'
+        echo "<testsuites><testsuite name=\"$suiteName\" tests=\"${#precheckNames[@]}\" failures=\"$junitFailures\" errors=\"0\" skipped=\"0\" time=\"0\">"
+        printf '%s' "$junitCases"
+        echo "</testsuite></testsuites>"
+    } > "$junitXml" 2>/dev/null; then
+        echo "Synthetic junit result written to $junitXml (will appear in the Jenkins test report)"
+    else
+        echo "WARNING: could not write synthetic junit xml to $junitXml (non-fatal)"
+    fi
+}
+
+# Run one precheck srun per ctx/gen server instance with the same node
+# slices, MPI topology, and UCX env as the real server steps
+# (pytestCommand{CTX,GEN}Precheck embed the same ucx_tls_cmd + worker env var
+# strings). On failure: console summary + synthetic junit, then the stage
+# aborts via cleanup_on_failure — before any model bring-up.
+run_cache_transceiver_precheck() {
+    if [ "${ctPrecheckEnabled:-0}" != "1" ] || [ "${TRTLLM_DISAGG_BENCHMARK_GEN_ONLY:-0}" = "1" ]; then
+        return 0
+    fi
+    echo "Starting cache transceiver precheck..."
+    precheckDir="$testOutputDir/cache_transceiver_precheck"
+    mkdir -p "$precheckDir/logs"
+    # A reused work dir (Slurm requeue reruns this batch script with the same
+    # directories) may hold a previous run's rendezvous/status/csv/abort files:
+    # stale addr files would point gen leaders at dead ports, stale status files
+    # would pollute the verdict aggregation below, a stale precheck.abort would
+    # fail-fast-skip the whole rerun (a requeued job keeps its SLURM_JOB_ID, so
+    # the driver's job-id stamp cannot tell it apart), and stale bandwidth CSVs
+    # (the Python transceiver's perf_<uuid>_<rank>.csv are per-run and appended)
+    # would make parse_python_bandwidth_gbps median over two runs' samples. The
+    # driver also job-id-stamps addr files as a second line of defense.
+    rm -f "$precheckDir"/rendezvous/*.addr "$precheckDir"/status/*.status \
+        "$precheckDir"/status/*.json "$precheckDir"/precheck.abort 2>/dev/null || true
+    rm -rf "$precheckDir"/csv 2>/dev/null || true
+    precheckPids=()
+    precheckNames=()
+    # ct_launch_step <role> <idx> <nodes> <gpusPerNode> <nodeList> <pytestCmd>
+    ct_launch_step() {
+        local role=$1 i=$2 nodes=$3 gpusPerNode=$4 nodeList=$5 pytestCmd=$6
+        export DISAGG_SERVING_TYPE="${role^^}_PRECHECK_$i"
+        export pytestCommand="$pytestCmd --server-idx $i"
+        timeout -k 60 "${ctPrecheckTimeout:-900}" \
+            srun "${srunArgs[@]}" --mpi=pmix --kill-on-bad-exit=1 \
+            -N "$nodes" \
+            -w "$nodeList" \
+            --ntasks=$((nodes * gpusPerNode)) \
+            --ntasks-per-node="$gpusPerNode" \
+            bash $precheckRunScript &> "$precheckDir/logs/${role}_$i.log" &
+        precheckPids+=($!)
+        precheckNames+=("${role}_$i")
+        sleep 5  # Wait for pyxis container namespace initialization to avoid race condition
+    }
+    local i
+    for i in $(seq 0 $((numGenServers - 1))); do
+        ct_launch_step gen "$i" "$nodesPerGenServer" "$gpusPerNodePerGenServer" \
+            "${genNodeLists[$i]}" "$pytestCommandGENPrecheck"
+    done
+    for i in $(seq 0 $((numCtxServers - 1))); do
+        ct_launch_step ctx "$i" "$nodesPerCtxServer" "$gpusPerNodePerCtxServer" \
+            "${ctxNodeLists[$i]}" "$pytestCommandCTXPrecheck"
+    done
+
+    local precheckFailed=0 k rc
+    for k in "${!precheckPids[@]}"; do
+        if wait "${precheckPids[$k]}"; then
+            echo "Precheck step ${precheckNames[$k]} passed"
+        else
+            rc=$?
+            echo "Precheck step ${precheckNames[$k]} FAILED (exit $rc; 124 = external timeout)"
+            precheckFailed=1
+        fi
+    done
+
+    if [ "$precheckFailed" -eq 1 ]; then
+        ct_print_failure_summary
+        ct_write_junit_xml
+        cleanup_on_failure "Cache transceiver precheck failed. See summary above and $precheckDir"
+    fi
+    # No status files means every step took the skip path (e.g. the yaml has no
+    # cache_transceiver_config.backend): the run wrote no verdicts, so report it
+    # as SKIPPED rather than an empty "PASSED" that reads like real validation.
+    if ls "$precheckDir"/status/*.status >/dev/null 2>&1; then
+        echo "Cache transceiver precheck PASSED:"
+        cat "$precheckDir"/status/*.status 2>/dev/null || true
+    else
+        echo "Cache transceiver precheck SKIPPED: not applicable for this config" \
+            "(no verdicts written; see $precheckDir/logs)"
+    fi
+}

jenkins/scripts/perf/disaggregated/slurm_launch_draft.sh

@@ -17,6 +17,38 @@ if ! srun "${srunArgs[@]}" $installScript &> $jobWorkspace/install.log; then
 fi
 echo "Installation completed on all nodes"
 
+# Deterministic node slices per server: gen servers take the first nodes,
+# then ctx servers (same order the steps are started in). Both the cache
+# transceiver precheck and the real server steps pin to these slices with
+# `srun -w`, so the precheck exercises exactly the node pairs / NICs the
+# real disaggregated test will use.
+mapfile -t allNodes < <(scontrol show hostnames "$SLURM_JOB_NODELIST")
+nodeCursor=0
+genNodeLists=()
+for i in $(seq 0 $((numGenServers - 1))); do
+    slice=("${allNodes[@]:$nodeCursor:$nodesPerGenServer}")
+    genNodeLists+=("$(IFS=,; echo "${slice[*]}")")
+    nodeCursor=$((nodeCursor + nodesPerGenServer))
+done
+ctxNodeLists=()
+if [ "${TRTLLM_DISAGG_BENCHMARK_GEN_ONLY:-0}" != "1" ]; then
+    for i in $(seq 0 $((numCtxServers - 1))); do
+        slice=("${allNodes[@]:$nodeCursor:$nodesPerCtxServer}")
+        ctxNodeLists+=("$(IFS=,; echo "${slice[*]}")")
+        nodeCursor=$((nodeCursor + nodesPerCtxServer))
+    done
+fi
+if [ "$nodeCursor" -gt "${#allNodes[@]}" ]; then
+    cleanup_on_failure "Node slicing needs $nodeCursor nodes but the job only has ${#allNodes[@]} ($SLURM_JOB_NODELIST)"
+fi
+
+# Cache transceiver network precheck: same instance count / node slices /
+# MPI topology / UCX env as the real ctx+gen server steps. On failure the
+# stage aborts HERE, with per-instance verdicts + a synthetic junit entry,
+# before any model bring-up. Functions come from slurm_ct_precheck_gate.sh,
+# spliced in above this draft by submit.py. No-op unless ctPrecheckEnabled=1.
+run_cache_transceiver_precheck
+
 # Start gen servers
 echo "Starting gen servers..."
 for i in $(seq 0 $((numGenServers - 1))); do
@@ -25,10 +57,11 @@ for i in $(seq 0 $((numGenServers - 1))); do
     export pytestCommand="$pytestCommandGENWorker"
     srun "${srunArgs[@]}" --mpi=pmix --kill-on-bad-exit=1 \
         -N $nodesPerGenServer \
+        -w "${genNodeLists[$i]}" \
         --ntasks=$gen_world_size \
         --ntasks-per-node=$gpusPerNodePerGenServer \
         $runScript &> $testOutputDir/gen_server_$i.log &
-    echo "Started gen server $i"
+    echo "Started gen server $i on ${genNodeLists[$i]}"
     sleep 5  # Wait for pyxis container namespace initialization to avoid race condition
 done
 
@@ -41,10 +74,11 @@ if [ "${TRTLLM_DISAGG_BENCHMARK_GEN_ONLY:-0}" != "1" ]; then
         export pytestCommand="$pytestCommandCTXWorker"
         srun "${srunArgs[@]}" --mpi=pmix --kill-on-bad-exit=1 \
             -N $nodesPerCtxServer \
+            -w "${ctxNodeLists[$i]}" \
         --ntasks=$ctx_world_size \
         --ntasks-per-node=$gpusPerNodePerCtxServer \
             $runScript &> $testOutputDir/ctx_server_$i.log &
-        echo "Started ctx server $i"
+        echo "Started ctx server $i on ${ctxNodeLists[$i]}"
         sleep 5  # Wait for pyxis container namespace initialization to avoid race condition
     done
 else

jenkins/scripts/perf/disaggregated/slurm_precheck_run.sh

@@ -0,0 +1,39 @@
+#!/bin/bash
+
+# Per-rank container entrypoint for the disagg cache-transceiver PRECHECK
+# step (launched by slurm_launch_draft.sh BEFORE the real ctx/gen servers).
+#
+# It intentionally mirrors slurm_run.sh's runtime environment: the same
+# slurm_env_setup.sh is sourced (LD_LIBRARY_PATH, the `unset UCX_TLS=tcp`
+# fixup, PMIX_MCA_gds), and $pytestCommand carries the same
+# `unset/export UCX_TLS ...` prefix and worker env vars as the real ctx/gen
+# worker steps (built from the same strings by jenkins/scripts/perf/submit.py),
+# so a precheck PASS/FAIL is representative of the network environment the
+# real test will run in. Unlike slurm_run.sh it skips coverage/perf-report
+# handling -- the precheck is a plain MPI program, not a pytest run.
+
+set -xEeuo pipefail
+trap 'rc=$?; echo "Error in file ${BASH_SOURCE[0]} on line $LINENO: $BASH_COMMAND (exit $rc)"; exit $rc' ERR
+
+# CI (jenkins/L0_Test.groovy) exports resourcePathNode with the /tmp
+# extraction layout; the local flow (jenkins/scripts/perf/local/submit.py)
+# exports llmSrcNode pointing at the repo directly and has no
+# resourcePathNode. Honor whichever layout is present.
+if [ -n "${resourcePathNode:-}" ]; then
+    cd "$resourcePathNode"
+    llmSrcNode=$resourcePathNode/TensorRT-LLM/src
+fi
+: "${llmSrcNode:?either resourcePathNode or llmSrcNode must be exported}"
+
+source "$llmSrcNode/jenkins/scripts/slurm_env_setup.sh"
+slurm_setup_runtime_env
+
+echo "Precheck rank ${SLURM_PROCID:-?} (${DISAGG_SERVING_TYPE:-unknown}) command: $pytestCommand"
+
+set +e
+eval $pytestCommand
+precheck_exit_code=$?
+set -e
+
+echo "Rank${SLURM_PROCID:-?} cache-transceiver precheck finished with exit code $precheck_exit_code"
+exit $precheck_exit_code