diff --git a/jenkins/scripts/perf/local/submit.py b/jenkins/scripts/perf/local/submit.py index c12798aa847e..0d56b3cfa068 100755 --- a/jenkins/scripts/perf/local/submit.py +++ b/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: diff --git a/jenkins/scripts/perf/submit.py b/jenkins/scripts/perf/submit.py index d68e776f036e..da49367c1fdc 100755 --- a/jenkins/scripts/perf/submit.py +++ b/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: diff --git a/tests/unittest/scripts/test_perf_submit.py b/tests/unittest/scripts/test_perf_submit.py new file mode 100644 index 000000000000..ceba38c3bed3 --- /dev/null +++ b/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