[AMD][AgentX] Tune MI300X Qwen3.5 MORI disaggregation with srt-slurm

benchmarks/benchmark_lib.sh

@@ -1891,7 +1891,11 @@ run_lm_eval() {
         export INFERENCEX_LM_EVAL_RUNTIME_READY=true
     fi
 
-    local openai_server_base="http://0.0.0.0:${port}"
+    # Most launchers run eval beside the API process and keep the historical
+    # 0.0.0.0 default. Orchestrators such as srt-slurm can place the benchmark
+    # client on a different node, so allow them to provide the routed host.
+    local openai_server_host="${EVAL_SERVER_HOST:-0.0.0.0}"
+    local openai_server_base="http://${openai_server_host}:${port}"
     local openai_chat_base="${openai_server_base}/v1/chat/completions"
     export OPENAI_API_KEY=${OPENAI_API_KEY:-EMPTY}
     MODEL_NAME=${MODEL_NAME:-$MODEL} # Prefer MODEL_NAME, else MODEL

benchmarks/multi_node/agentic_srt.sh

@@ -60,24 +60,102 @@ if [[ "${EVAL_ONLY:-false}" == "true" ]]; then
     _wait_for_openai_chat_route --port "$PORT"
 fi
 
+# Preserve the legacy DP-attention replay contract. The SGLang router uses this
+# header to keep every request in one AgentX correlation tree on a stable DP
+# route, which is important for both session continuity and prefix-cache reuse.
+if [[ "${PREFILL_DP_ATTN:-false}" == "true" ]]; then
+    export AIPERF_HTTP_X_SMG_ROUTING_KEY_FROM_CORRELATION_ID=true
+fi
+
+# Reset every advertised SGLang worker before each concurrency point, matching
+# the retired amd_utils trace replay. /flush_cache covers GPU radix + host
+# HiCache; the storage-backend endpoint is best-effort because L3 is optional.
+clear_agentic_worker_caches() {
+    local timeout_seconds="${FLUSH_DRAIN_TIMEOUT:-120}"
+    local metrics_csv="${AIPERF_SERVER_METRICS_URLS:-}"
+    if [[ -z "$metrics_csv" ]]; then
+        if [[ "${AIPERF_DRAIN_BACKEND:-dynamo}" == "sglang" ]]; then
+            echo "[clear_caches] ERROR: native SGLang requires all worker metrics URLs" >&2
+            return 1
+        fi
+        echo "[clear_caches] WARN: AIPERF_SERVER_METRICS_URLS unset; skipping cache flush" >&2
+        return 0
+    fi
+
+    local -a metrics_urls
+    if [[ "${AIPERF_DRAIN_BACKEND:-dynamo}" == "sglang" && \
+          ( "$metrics_csv" == ,* || "$metrics_csv" == *, || "$metrics_csv" == *,,* ) ]]; then
+        echo "[clear_caches] ERROR: empty worker metrics URL" >&2
+        return 1
+    fi
+    IFS=',' read -r -a metrics_urls <<< "$metrics_csv"
+    local metrics_url base_url start response code flushed
+    for metrics_url in "${metrics_urls[@]}"; do
+        [[ -n "$metrics_url" ]] || continue
+        base_url="${metrics_url%/metrics}"
+        start=$(date +%s)
+        flushed=0
+        response=""
+        while :; do
+            response=$(curl -sf -m 10 -X POST "${base_url}/flush_cache" 2>/dev/null || true)
+            if grep -qi "Cache flushed" <<< "$response"; then
+                flushed=1
+                break
+            fi
+            if (( $(date +%s) - start >= timeout_seconds )); then
+                break
+            fi
+            sleep 3
+        done
+        if (( flushed )); then
+            echo "[clear_caches] ${base_url}: L1+L2 flushed"
+        else
+            if [[ "${AIPERF_DRAIN_BACKEND:-dynamo}" == "sglang" ]]; then
+                echo "[clear_caches] ERROR ${base_url}: L1+L2 flush not confirmed after ${timeout_seconds}s" >&2
+                return 1
+            fi
+            echo "[clear_caches] WARN ${base_url}: L1+L2 flush not confirmed after ${timeout_seconds}s" >&2
+        fi
+
+        code=$(curl -s -m 60 -o /dev/null -w '%{http_code}' \
+            -X POST "${base_url}/hicache/storage-backend/clear" 2>/dev/null || true)
+        if [[ "$code" == "200" ]]; then
+            echo "[clear_caches] ${base_url}: L3 store cleared"
+        else
+            echo "[clear_caches] ${base_url}: L3 clear http=${code:-000} (optional backend unavailable)"
+        fi
+    done
+}
+
 wait_for_agentic_servers_idle() {
     local timeout_seconds="${AIPERF_DRAIN_TIMEOUT_SECONDS:-1800}"
     local poll_seconds="${AIPERF_DRAIN_POLL_SECONDS:-10}"
-    local frontend_metrics_url="${AIPERF_SERVER_URL%/}/metrics"
+    local frontend_metrics_url="${AIPERF_FRONTEND_METRICS_URL:-${AIPERF_SERVER_URL:-http://${SRT_FRONTEND_HOST:-localhost}:${PORT}}/metrics}"
 
     "$AIPERF_PYTHON" - \
         "$timeout_seconds" \
         "$poll_seconds" \
         "$frontend_metrics_url" \
-        "${AIPERF_SERVER_METRICS_URLS:-}" <<'PY'
+        "${AIPERF_SERVER_METRICS_URLS:-}" \
+        "${AIPERF_DRAIN_BACKEND:-dynamo}" \
+        "${DISAGG:-false}" <<'PY'
+import math
 import sys
 import time
 import urllib.request
 
 timeout_seconds = int(sys.argv[1])
 poll_seconds = int(sys.argv[2])
 frontend_url = sys.argv[3]
-worker_urls = [url for url in sys.argv[4].split(",") if url]
+worker_urls = [url.strip() for url in sys.argv[4].split(",")]
+backend = sys.argv[5]
+disagg = sys.argv[6].lower() == "true"
+if backend not in {"sglang", "dynamo"}:
+    raise SystemExit(f"Unsupported AIPERF_DRAIN_BACKEND: {backend}")
+if not all(worker_urls):
+    raise SystemExit("Agentic drain requires non-empty worker metrics URLs")
+if timeout_seconds <= 0 or poll_seconds <= 0:
+    raise SystemExit("Agentic drain timeout and poll interval must be positive")
 deadline = time.monotonic() + timeout_seconds
 idle_polls = 0
 
@@ -87,32 +165,67 @@ def fetch_metrics(url: str) -> str:
         return response.read().decode("utf-8")
 
 
-def metric_sum(metrics: str, name: str) -> float:
-    total = 0.0
+def metric_sum(metrics: str, name: str, *, required: bool = True) -> float:
+    values = []
     for line in metrics.splitlines():
         if not line or line.startswith("#"):
             continue
         fields = line.split()
         if len(fields) < 2 or fields[0].split("{", 1)[0] != name:
             continue
-        total += float(fields[1])
-    return total
+        value = float(fields[1])
+        if not math.isfinite(value) or value < 0:
+            raise ValueError(f"Invalid request gauge {name}: {value}")
+        values.append(value)
+    if required and not values:
+        raise ValueError(f"Missing required request gauge: {name}")
+    return sum(values)
+
+
+def worker_requests(metrics: str) -> float:
+    # SGLang's scheduler also holds requests outside running/waiting while KV
+    # is bootstrapping or transferring. These queues must drain before flush.
+    # Names match sglang/srt/observability/metrics_collector.py.
+    is_sglang = backend == "sglang" or "sglang:num_running_reqs" in metrics
+    if not is_sglang:
+        if "trtllm_num_requests_running" in metrics:
+            return metric_sum(metrics, "trtllm_num_requests_running") + metric_sum(
+                metrics, "trtllm_num_requests_waiting"
+            )
+        return metric_sum(metrics, "vllm:num_requests_running") + metric_sum(
+            metrics, "vllm:num_requests_waiting"
+        )
+    active = metric_sum(metrics, "sglang:num_running_reqs") + metric_sum(
+        metrics, "sglang:num_queue_reqs"
+    )
+    active += metric_sum(metrics, "sglang:num_grammar_queue_reqs", required=False)
+    for name in (
+        "sglang:num_prefill_bootstrap_queue_reqs",
+        "sglang:num_prefill_inflight_queue_reqs",
+        "sglang:num_decode_prealloc_queue_reqs",
+        "sglang:num_decode_transfer_queue_reqs",
+    ):
+        active += metric_sum(metrics, name, required=disagg)
+    return active
 
 
 while time.monotonic() < deadline:
     try:
-        frontend_metrics = fetch_metrics(frontend_url)
-        frontend_active = metric_sum(frontend_metrics, "dynamo_frontend_active_requests")
+        frontend_active = 0.0
+        # Native SGLang's router does not expose Dynamo's active-request gauge;
+        # its worker schedulers (including all PD queues) are authoritative.
+        if backend == "dynamo":
+            frontend_active = metric_sum(
+                fetch_metrics(frontend_url), "dynamo_frontend_active_requests"
+            )
         worker_active = 0.0
         for worker_url in worker_urls:
             worker_metrics = fetch_metrics(worker_url)
-            worker_active += metric_sum(worker_metrics, "vllm:num_requests_running")
-            worker_active += metric_sum(worker_metrics, "vllm:num_requests_waiting")
-            worker_active += metric_sum(worker_metrics, "trtllm_num_requests_running")
-            worker_active += metric_sum(worker_metrics, "trtllm_num_requests_waiting")
+            worker_active += worker_requests(worker_metrics)
+        frontend_status = f"{frontend_active:g}" if backend == "dynamo" else "n/a"
         print(
-            f"Agentic drain status: frontend_active={frontend_active:g} "
-            f"worker_running_or_waiting={worker_active:g}",
+            f"Agentic drain status: backend={backend} frontend_active={frontend_status} "
+            f"worker_pending_requests={worker_active:g}",
             flush=True,
         )
         if frontend_active == 0 and worker_active == 0:
@@ -145,6 +258,9 @@ for index in "${!CONCURRENCIES[@]}"; do
     mkdir -p "$RESULT_DIR"
 
     echo "Running agentic concurrency $concurrency of: ${CONCURRENCIES[*]}"
+    if [[ "${CLEAR_CACHE_BETWEEN_CONC:-0}" == "1" ]]; then
+        clear_agentic_worker_caches
+    fi
     build_replay_cmd "$RESULT_DIR"
     run_agentic_replay_and_write_outputs "$RESULT_DIR"
 

benchmarks/multi_node/srt-slurm-recipes/cluster-configs/mi300x-amd.yaml

@@ -0,0 +1,24 @@
+# MI300X AMD fleet: shared NFS source/results, node-local images and HF cache.
+# Slurm selects available nodes from compute; drained nodes remain excluded.
+cluster: mi300x-amd
+default_partition: compute
+default_time_limit: "04:00:00"
+gpus_per_node: 8
+accelerator_vendor: amd
+network_interface: ens50f1np1
+gpu_sbatch_directive: gres
+use_segment_sbatch_directive: false
+use_exclusive_sbatch_directive: true
+runtime_config_transport: shared-filesystem
+record_launch_plan: true
+default_sbatch_directives:
+  # Full-node allocation: all 128 logical CPUs and all RAM on this fleet.
+  cpus-per-task: "128"
+  mem: "0"
+default_bash_preamble: "ulimit -l unlimited; ulimit -c 0"
+default_mounts:
+  /dev/kfd: /dev/kfd
+  /dev/dri: /dev/dri
+  /dev/infiniband: /dev/infiniband
+  /raid/inferencex/models: /hf-cache
+nginx_raise_ulimit: false