[https://nvbugs/6510284][fix] Cap gen-only benchmark queue size

jenkins/scripts/perf/local/submit.py

@@ -324,6 +324,25 @@ def get_benchmark_config(config, benchmark_mode):
     }
 
 
+def get_benchmark_request_queue_size(config, concurrency):
+    """Cap the gen-only fill target to the GEN executor's active capacity."""
+    gen_config = (config.get("worker_config", {}) or {}).get("gen", {}) or {}
+    concurrency = int(concurrency)
+    max_batch_size = int(gen_config.get("max_batch_size", concurrency))
+    enable_attention_dp = gen_config.get("enable_attention_dp", False)
+    tp_size = int(gen_config.get("tensor_parallel_size", 1))
+    max_capacity = max_batch_size * tp_size if enable_attention_dp else max_batch_size
+    queue_size = min(max_capacity, concurrency)
+    if queue_size < concurrency:
+        print(
+            "[WARNING] TLLM_BENCHMARK_REQ_QUEUES_SIZE capped to "
+            f"{queue_size} (max_batch_size={max_batch_size}, tp_size={tp_size}, "
+            f"attention_dp={enable_attention_dp}) instead of concurrency={concurrency}. "
+            "The fill loop cannot reach a target above the GEN executor capacity."
+        )
+    return queue_size
+
+
 def partition_has_gpu_gres(partition):
     """Return True if the Slurm partition reports GPU GRES (e.g. 'gpu:4'), False if null/absent."""
     try:
@@ -835,12 +854,13 @@ def main():
             srun_args_lines.append("--container-env=TRTLLM_DISAGG_BENCHMARK_GEN_ONLY")
         elif "gen_only" in bm_config.get("mode", ""):
             concurrency = bm_config.get("concurrency", 1)
+            queue_size = get_benchmark_request_queue_size(config, concurrency)
             ctx_worker_env_vars = (
                 f"TRTLLM_DISABLE_KV_CACHE_TRANSFER_OVERLAP=1 {ctx_worker_env_vars}"
             )
             gen_worker_env_vars = (
                 f"TRTLLM_DISABLE_KV_CACHE_TRANSFER_OVERLAP=1 "
-                f"TLLM_BENCHMARK_REQ_QUEUES_SIZE={concurrency} {gen_worker_env_vars}"
+                f"TLLM_BENCHMARK_REQ_QUEUES_SIZE={queue_size} {gen_worker_env_vars}"
             )
 
         if is_gb300:

jenkins/scripts/perf/submit.py

@@ -289,6 +289,25 @@ def get_benchmark_config(config):
     }
 
 
+def get_benchmark_request_queue_size(config, concurrency):
+    """Cap the gen-only fill target to the GEN executor's active capacity."""
+    gen_config = (config.get("worker_config", {}) or {}).get("gen", {}) or {}
+    concurrency = int(concurrency)
+    max_batch_size = int(gen_config.get("max_batch_size", concurrency))
+    enable_attention_dp = gen_config.get("enable_attention_dp", False)
+    tp_size = int(gen_config.get("tensor_parallel_size", 1))
+    max_capacity = max_batch_size * tp_size if enable_attention_dp else max_batch_size
+    queue_size = min(max_capacity, concurrency)
+    if queue_size < concurrency:
+        print(
+            "[WARNING] TLLM_BENCHMARK_REQ_QUEUES_SIZE capped to "
+            f"{queue_size} (max_batch_size={max_batch_size}, tp_size={tp_size}, "
+            f"attention_dp={enable_attention_dp}) instead of concurrency={concurrency}. "
+            "The fill loop cannot reach a target above the GEN executor capacity."
+        )
+    return queue_size
+
+
 # --------------------------------------------------------------------------- #
 # pytestCommand splitting
 # --------------------------------------------------------------------------- #
@@ -554,12 +573,13 @@ def main():
             srun_args_lines.append("--container-env=TRTLLM_DISAGG_BENCHMARK_GEN_ONLY")
         elif benchmark_mode == "gen_only":
             concurrency = benchmark_config.get("concurrency", 1)
+            queue_size = get_benchmark_request_queue_size(config, concurrency)
             ctx_worker_env_vars = (
                 f"TRTLLM_DISABLE_KV_CACHE_TRANSFER_OVERLAP=1 {ctx_worker_env_vars}"
             )
             gen_worker_env_vars = (
                 f"TRTLLM_DISABLE_KV_CACHE_TRANSFER_OVERLAP=1 "
-                f"TLLM_BENCHMARK_REQ_QUEUES_SIZE={concurrency} {gen_worker_env_vars}"
+                f"TLLM_BENCHMARK_REQ_QUEUES_SIZE={queue_size} {gen_worker_env_vars}"
             )
 
         if is_gb300:

tests/unittest/scripts/test_perf_submit.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+import yaml
+
+REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent
+SUBMIT_PATHS = (
+    REPO_ROOT / "jenkins" / "scripts" / "perf" / "submit.py",
+    REPO_ROOT / "jenkins" / "scripts" / "perf" / "local" / "submit.py",
+)
+DISAGG_CONFIG_DIR = REPO_ROOT / "tests" / "scripts" / "perf-sanity" / "disaggregated"
+
+
+@pytest.fixture(params=SUBMIT_PATHS, ids=("ci", "local"))
+def submit_module(request):
+    spec = importlib.util.spec_from_file_location(
+        f"perf_submit_{request.param.parent.name}", request.param
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.mark.parametrize(
+    ("config_name", "expected_queue_size"),
+    (
+        ("gb300_deepseek-v4-pro-fp4_8k1k_con8_ctx1_dep4_gen4_tep8_eplb0_mtp3_ccb-NIXL", 1),
+        (
+            "gb300_deepseek-v4-pro-fp4_8k1k_con180_ctx3_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL",
+            128,
+        ),
+        (
+            "gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx6_dep4_gen1_dep16_eplb384_mtp3_ccb-NIXL",
+            512,
+        ),
+        (
+            "gb300_deepseek-v4-pro-fp4_8k1k_con4301_ctx12_dep4_gen1_dep8_eplb384_mtp1_ccb-NIXL",
+            4096,
+        ),
+    ),
+)
+def test_gen_only_queue_size_does_not_exceed_executor_capacity(
+    submit_module, config_name, expected_queue_size
+):
+    with open(DISAGG_CONFIG_DIR / f"{config_name}.yaml") as config_file:
+        config = yaml.safe_load(config_file)
+
+    concurrency = config["benchmark"]["concurrency_list"]
+
+    assert (
+        submit_module.get_benchmark_request_queue_size(config, concurrency) == expected_queue_size
+    )
+
+
+def test_gen_only_queue_size_preserves_reachable_concurrency(submit_module):
+    config = {
+        "worker_config": {
+            "gen": {
+                "max_batch_size": 8,
+                "tensor_parallel_size": 4,
+                "enable_attention_dp": True,
+            }
+        }
+    }
+
+    assert submit_module.get_benchmark_request_queue_size(config, 16) == 16