Skip to content

[https://nvbugs/6472256][fix] Fix disagg stress cluster flapping and DeepSeek R1 FP4 ctx OOM; add aiperf error-rate gate #17427

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
moraxu wants to merge 1 commit into
base: main
Choose a base branch
from
+175 −19
Open

[https://nvbugs/6472256][fix] Fix disagg stress cluster flapping and DeepSeek R1 FP4 ctx OOM; add aiperf error-rate gate #17427

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .../defs/disaggregated/test_configs/disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,13 @@ context_servers:
dtype: fp8
moe_config:
backend: TRTLLM
# Chunk the MoE forward at half of max_num_tokens: the TRTLLM-Gen FP4
# MoE workspace for a full 16640-token chunk is a 7-8 GiB transient,
# which exceeds the headroom left after weights (~107.5 GiB/GPU) and the
# KV pool on 192GB B200 and intermittently OOMs the ctx worker under
# 512-concurrency 8k prefill load (memory estimation only observes
# ~10.8 GiB dynamic peak, so the KV pool leaves no slack for it).
max_num_tokens: 8320

@brnguyen2 brnguyen2 Aug 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

disagg_config_ctxtp4_gentp4_deepseek_r1_v2_fp4_tllm_mtp.yaml has an identical context server (max_num_tokens: 16640, free_gpu_memory_fraction: 0.8, TRTLLM MoE backend, same weights) and does not get this chunking. By the analysis in the description it carries the same workspace transient, and deepseek_r1_v2_fp4_mtp_stress is not waived — so it is a live OOM candidate on B200. Either apply the same moe_config.max_num_tokens there or say in the PR why the MTP variant's headroom differs (enable_attention_dp: false and max_draft_len: 1 change the picture, but in which direction?).

cuda_graph_config: null
print_iter_log: true
cache_transceiver_config:
Expand Down
186 changes: 168 additions & 18 deletions tests/integration/defs/disaggregated/test_disaggregated.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import subprocess
import tempfile
import time
from collections import namedtuple
from collections import deque, namedtuple
from dataclasses import dataclass
from typing import Any, Optional

Expand Down Expand Up @@ -129,6 +129,49 @@ def scan_logs_for_fatal_errors(processes):
return findings


def print_first_fatal_log_context(processes, context_lines=20):
"""Print the lines around the FIRST fatal-pattern match in each log.

The last-N-lines tail printed on failure usually shows only post-crash
shutdown spam ("LLM is shutting down" storms); the root cause — e.g. the
OOM traceback with the allocation site — is at the first match, often
thousands of lines earlier. Streams each log to avoid loading multi-GB
worker logs into memory.
"""
for proc in processes:
log_path = getattr(proc, "log_path", None)
if not log_path or not os.path.exists(log_path):
continue
before = deque(maxlen=context_lines)
after = []
matched = None
try:
with open(log_path, "r", errors="replace") as f:
for lineno, line in enumerate(f, 1):
if matched is None:
pat = next(
(p for p in _FATAL_LOG_PATTERNS if p in line), None)
if pat is None:
before.append(line)
continue
matched = (lineno, pat)
after.append(line)
else:
after.append(line)
if len(after) > context_lines:
break
except OSError:
continue
if matched is None:
continue
lineno, pat = matched
logger.error(f"-------- {log_path}: first fatal pattern '{pat}' at "
f"line {lineno} (+/-{context_lines} lines) --------")
for line in [*before, *after]:
if line.strip():
logger.error(line.rstrip())


def _crashed_workers(workers):
return [
w for w in workers
Expand Down Expand Up @@ -165,6 +208,20 @@ def get_default_disagg_cluster_config():
}


# Production service-discovery timings, matching the DisaggClusterConfig
# defaults in tensorrt_llm/llmapi/disagg_utils.py. The tight 1s/2s defaults
# above keep short functional tests snappy, but at stress-level concurrency
# they leave <1s of heartbeat slack while the worker heartbeat task, the
# cluster-storage /expire handler, and the expiry sweep all share event loops
# saturated by request traffic — so workers get spuriously expired and the
# router flaps "Cluster is not ready" (nvbugs/6472256). Stress runners must
# use these production values instead.
PRODUCTION_CLUSTER_TIMINGS = {
"heartbeat_interval_sec": 5,
"inactive_timeout_sec": 10,
}


def build_worker_config(base_config: dict[str, Any],
server_type_config: dict[str, Any],
disagg_cluster: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -657,6 +714,7 @@ def setup_disagg_cluster(
startup_callback=None,
startup_tick: int = 30,
perf_metrics_output_dir: str | None = None,
disagg_cluster_overrides: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], list[ProcessWrapper], list[ProcessWrapper],
ProcessWrapper, int, str]:
"""Load config, launch workers + disagg server, wait for ready.
Expand All @@ -667,6 +725,9 @@ def setup_disagg_cluster(
env: Environment variables to pass to subprocess (workers and disagg server)
server_start_timeout: Timeout in seconds for server to become ready
schedule_style: Disagg schedule style ('context_first' or 'generation_first')
disagg_cluster_overrides: Entries merged over
get_default_disagg_cluster_config(), e.g. PRODUCTION_CLUSTER_TIMINGS
for stress runs (cluster_uri/minimal_instances are still derived below)

Returns:
tuple: (config, ctx_workers, gen_workers, disagg_server, server_port, work_dir)
Expand All @@ -689,6 +750,8 @@ def setup_disagg_cluster(
speculative_model)

disagg_cluster = get_default_disagg_cluster_config()
if disagg_cluster_overrides:
disagg_cluster.update(disagg_cluster_overrides)
server_host = config.get("hostname", "localhost")
server_port = get_free_port()
if save_log:
Expand Down Expand Up @@ -2308,6 +2371,69 @@ def get_config_for_benchmark(model_root, backend):
return serve_config


def enforce_aiperf_error_rate(artifact_dir, max_error_rate):
"""Fail if the fraction of non-cancellation request errors exceeds max_error_rate.

aiperf exits 0 and counts HTTP 500s as completed requests, so without this
gate a mid-run server error storm (e.g. "Cluster is not ready" readiness
flapping, nvbugs/6472256) passes silently. Reads aiperf's per-record export
(profile_export.jsonl in artifact_dir), where each line carries an optional
"error" object with code/type/message. Intentional client-side cancellations
(HTTP 499 / RequestCancellationError) are excluded from both the numerator
and the denominator — stress tests cancel a fraction of requests on purpose.
"""
export_path = os.path.join(artifact_dir, "profile_export.jsonl")
assert os.path.exists(export_path), (
f"aiperf per-record export not found at {export_path}; cannot enforce "
"the request error-rate gate. If this aiperf version/export level does "
"not produce it, pass max_error_rate=None explicitly.")
Comment on lines +2385 to +2389

@coderabbitai coderabbitai Bot Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Replace the assertion with an explicit exception.

Python removes assert statements when it runs with -O. A missing export then fails later at open() without the intended diagnostic. Raise FileNotFoundError explicitly.

Proposed fix 
-    assert os.path.exists(export_path), (
-        f"aiperf per-record export not found at {export_path}; cannot enforce "
-        "the request error-rate gate. If this aiperf version/export level does "
-        "not produce it, pass max_error_rate=None explicitly.")
+    if not os.path.exists(export_path):
+        raise FileNotFoundError(
+            f"aiperf per-record export not found at {export_path}; cannot enforce "
+            "the request error-rate gate. If this aiperf version/export level does "
+            "not produce it, pass max_error_rate=None explicitly.")

As per coding guidelines, “raise ValueError rather than assertions.”

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
export_path = os.path.join(artifact_dir, "profile_export.jsonl")
assert os.path.exists(export_path), (
f"aiperf per-record export not found at {export_path}; cannot enforce "
"the request error-rate gate. If this aiperf version/export level does "
"not produce it, pass max_error_rate=None explicitly.")
export_path = os.path.join(artifact_dir, "profile_export.jsonl")
if not os.path.exists(export_path):
raise FileNotFoundError(
f"aiperf per-record export not found at {export_path}; cannot enforce "
"the request error-rate gate. If this aiperf version/export level does "
"not produce it, pass max_error_rate=None explicitly.")
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/integration/defs/disaggregated/test_disaggregated.py` around lines 2385
- 2389, Replace the assert checking export_path in the disaggregated test with
an explicit FileNotFoundError, preserving the existing diagnostic message and
path context so missing exports fail immediately regardless of Python
optimization.

Source: Coding guidelines

total = 0
cancelled = 0
# (code, type) -> [count, example message]
error_counts: dict[tuple, list] = {}
with open(export_path, "r", errors="replace") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
total += 1
error = record.get("error")
if not error:
continue
code = error.get("code")
err_type = error.get("type")
if code == 499 or err_type == "RequestCancellationError":
cancelled += 1

@brnguyen2 brnguyen2 Aug 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The whole gate hinges on this classification being exhaustive: these stress runs use --request-cancellation-rate 10, so if aiperf reports a cancelled request under any other shape (a different type string, a null code, or no error object with a separate status field), ~10% of records become "errors" and every stress test trips the 5% threshold on the first run.

Please record the aiperf version this record schema was validated against in the docstring, and post the [aiperf-gate] output line from the 35k validation run so the cancelled: count can be checked against the expected ~3500.

continue
entry = error_counts.setdefault((code, err_type),
[0, error.get("message", "")])
entry[0] += 1
considered = total - cancelled
if considered <= 0:

@brnguyen2 brnguyen2 Aug 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This early return makes the gate silently pass in the cases you most want it to fail. total only counts lines that parsed as JSON, so if aiperf writes an empty export, gets killed mid-write, or changes the export path/nesting in a future version, total == 0 → considered <= 0 → clean pass. Same for the json.JSONDecodeError: continue above: a wholesale format change degrades to "no errors found" instead of an error.

Suggest counting decode failures and asserting the export is substantive before computing the rate:

assert total > 0, f"{export_path} contained no parseable records ({malformed} malformed lines)"
assert malformed <= 0.01 * (total + malformed), ...
if considered <= 0:
    raise AssertionError(f"all {total} records were cancellations")

return
errors = sum(count for count, _ in error_counts.values())
error_rate = errors / considered
print(
f"[aiperf-gate] non-cancellation errors: {errors}/{considered} "
f"({error_rate:.2%}), cancelled: {cancelled}, "
f"threshold: {max_error_rate:.2%}",
flush=True)
if error_rate > max_error_rate:
breakdown = "\n".join(
f" code={code} type={err_type}: {count} (e.g. {message[:200]})"
for (code, err_type), (count, message) in sorted(
error_counts.items(), key=lambda kv: -kv[1][0]))
raise AssertionError(
f"aiperf request error rate {error_rate:.2%} exceeds threshold "
f"{max_error_rate:.2%} ({errors} non-cancellation errors out of "
f"{considered} requests; {cancelled} intentional cancellations "
f"excluded). Error breakdown:\n{breakdown}")


def run_disaggregated_aiperf(config_file,
model_path,
server_start_timeout=1200,
Expand All @@ -2325,6 +2451,7 @@ def run_disaggregated_aiperf(config_file,
threshold=0.8,
cancellation_rate=None,
cancellation_delay=None,
max_error_rate=0.05,
env=None,
cwd=None):
"""Run disaggregated test with genai-perf for performance/stress testing.
Expand All @@ -2343,6 +2470,8 @@ def run_disaggregated_aiperf(config_file,
random_seed: Random seed for reproducibility
accuracy_test: Whether to run accuracy test
threshold: Threshold for accuracy test
max_error_rate: Fail if the fraction of non-cancellation request
errors recorded by aiperf exceeds this (None disables the gate)
env: Environment variables dict
cwd: Working directory
"""
Expand All @@ -2354,7 +2483,8 @@ def run_disaggregated_aiperf(config_file,
config, ctx_workers, gen_workers, disagg_server, server_port, work_dir = \
setup_disagg_cluster(config_file, model_name=model_path, env=run_env, cwd=cwd,
server_start_timeout=server_start_timeout,
save_log=True)
save_log=True,
disagg_cluster_overrides=PRODUCTION_CLUSTER_TIMINGS)

server_host = config.get("hostname", "localhost")
artifact_dir = os.path.join(cwd or ".", "benchmark-results")
Expand Down Expand Up @@ -2453,6 +2583,13 @@ def run_disaggregated_aiperf(config_file,
"Fatal error patterns detected in disaggregated worker/server "
f"logs:\n{summary}")

# Gate on the per-request error rate from aiperf's record export:
# aiperf exits 0 even when the server returns 500s for a large share
# of requests, so this is the only check that catches a mid-run error
# storm on this path (the fatal-pattern scan above is hang/OOM only).
if max_error_rate is not None:
enforce_aiperf_error_rate(artifact_dir, max_error_rate)

if accuracy_test:
accuracy_test_result, accuracy_value = run_accuracy_test(
model_path=model_path,
Expand Down Expand Up @@ -2489,14 +2626,17 @@ def run_disaggregated_aiperf(config_file,
f"worker/server logs after accuracy run:\n{summary}")

except Exception:
# Print tail of each captured worker/server log to aid triage.
# Print the context around the first fatal-pattern match (the root
# cause, e.g. an OOM traceback) and the tail of each captured
# worker/server log to aid triage.
print_first_fatal_log_context(
[*ctx_workers, *gen_workers, disagg_server])
for proc in [*ctx_workers, *gen_workers, disagg_server]:
log_path = getattr(proc, "log_path", None)
if not log_path or not os.path.exists(log_path):
continue
logger.error(f"-------- {log_path} (last 30 lines) --------")
try:
from collections import deque
with open(log_path, "r", errors="replace") as f:
for line in deque(f, maxlen=30):
if line.strip():
Expand Down Expand Up @@ -2676,13 +2816,20 @@ def test_llama4_long_context_kv_cache_overflow(disaggregated_test_root,
disaggregated_example_root,
os.path.dirname(__file__))

run_disaggregated_aiperf(config_file=config_file,
model_path=llama4_model_root,
server_start_timeout=1200,
input_tokens=128000,
output_tokens=100,
env=llm_venv._new_env,
cwd=llm_venv.get_working_directory())
run_disaggregated_aiperf(
config_file=config_file,
model_path=llama4_model_root,
server_start_timeout=1200,
input_tokens=128000,
output_tokens=100,
# This repro intentionally degrades the KV
# transfer path (tiny max_tokens_in_buffer vs
# 128k inputs), so sporadic request errors are
# by-design; keep the test scoped to its
# original crash/fatal-log checks.
max_error_rate=None,
env=llm_venv._new_env,
cwd=llm_venv.get_working_directory())


@skip_pre_blackwell
Expand Down Expand Up @@ -3446,9 +3593,10 @@ async def _warmup_requests(server_url: str, profiles: list, count: int,
The first request of each shape pays a one-time autotuner/compile cost
(~20s host-steps observed on B200). Running those here, before the measured
run, keeps them out of the accuracy/incomplete accounting and out of the
heartbeat-eviction path (a worker stuck in a 20s step misses the 2s cluster
heartbeat and gets evicted under a high-concurrency flood). Failures are
ignored — the only goal is to trigger the autotuner across the profile mix.
heartbeat-eviction path (a worker stuck in a 20s step exceeds even the 10s
PRODUCTION_CLUSTER_TIMINGS inactive timeout and gets evicted under a
high-concurrency flood). Failures are ignored — the only goal is to
trigger the autotuner across the profile mix.
"""
import random

Expand Down Expand Up @@ -3523,7 +3671,8 @@ def run_disaggregated_mixed_stress(example_dir: str,
config, ctx_workers, gen_workers, disagg_server, server_port, work_dir = \
setup_disagg_cluster(config_file, model_name=model_path, env=run_env,
cwd=cwd, server_start_timeout=server_start_timeout,
save_log=True, startup_callback=startup_callback)
save_log=True, startup_callback=startup_callback,
disagg_cluster_overrides=PRODUCTION_CLUSTER_TIMINGS)
print(f"[startup] cluster ready in {time.monotonic() - setup_start:.1f}s",
flush=True)

Expand All @@ -3540,9 +3689,10 @@ def run_disaggregated_mixed_stress(example_dir: str,
# Pay the one-time autotuner cost before the measured run. The first
# request of each shape triggers a ~20s autotuner host-step; left in
# the measured run at high concurrency, a worker stuck in that step
# misses the 2s cluster heartbeat and is evicted mid-run, causing
# "Cluster is not ready" 500s. Default count = ~20s at the ~6 req/s
# observed in the 5k B200 run; results are discarded.
# exceeds even the 10s PRODUCTION_CLUSTER_TIMINGS inactive timeout
# and is evicted mid-run, causing "Cluster is not ready" 500s.
# Default count = ~20s at the ~6 req/s observed in the 5k B200 run;
# results are discarded.
if warmup_request_count is None:
warmup_request_count = 120
print(
Expand Down
1 change: 0 additions & 1 deletion tests/integration/test_lists/waives.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,7 +163,6 @@ full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpu
full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpus_block_reuse[ADP4_MTP] SKIP (https://nvbugs/6525008)
full:B200/accuracy/test_llm_api_pytorch.py::TestNemotronV3Ultra::test_nvfp4_4gpus_block_reuse[TEP4] SKIP (https://nvbugs/6474894)
full:B200/accuracy/test_llm_api_pytorch.py::TestQwen3_30B_A3B::test_dummy_load_format SKIP (https://nvbugs/6525059)
full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-deepseek_r1_v2_fp4_stress] SKIP (https://nvbugs/6472256)
full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-glm5_nvfp4_tp4_ep4_dp_stress] SKIP (https://nvbugs/6544407)
full:B200/disaggregated/test_disaggregated.py::test_disaggregated_stress_test[input8k-output1k-conc512-qwen3_32b_fp8_stress] SKIP (https://nvbugs/6472256)

@brnguyen2 brnguyen2 Aug 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

qwen3_32b_fp8_stress is still waived under the same NVBug this PR fixes. The cluster-timing fix applies to it too (it runs through the same run_disaggregated_aiperf path), so either it should be unwaived alongside the DeepSeek param, or — if it fails for an unrelated reason — it needs its own bug ID. As it stands, nvbugs/6472256 can't be closed by this PR.

full:B200/llmapi/test_llm_api_pytorch_moe_lora.py::test_qwen_moe_routed_expert_multi_lora_varying_ranks[cudagraph] SKIP (https://nvbugs/6475623)
Expand Down
Loading