config part working

benchmarks/profiler/profile_sla.py

@@ -52,12 +52,12 @@
     parser.add_argument(
         "--backend",
         type=str,
-        default="vllm_v0",
-        choices=["vllm_v0", "vllm_v1"],
-        help="backend type (currently only vllm is supported)",
+        default="vllm_v1",
+        choices=["vllm_v1"],
+        help="backend type, currently support [vllm_v1]",
     )
     parser.add_argument(
-        "--config", type=str, required=True, help="Path to the dynamo config file"
+        "--config", type=str, required=True, help="Path to the DynamoGraphDeployment config file"
     )
     parser.add_argument(
         "--example-dir",
@@ -71,6 +71,18 @@
         default="profiling_results",
         help="Path to the output results directory",
     )
+    parser.add_argument(
+        "--min-num-gpus-per-engine",
+        type=int,
+        default=1,
+        help="minimum number of GPUs per engine",
+    )
+    parser.add_argument(
+        "--max-num-gpus-per-engine",
+        type=int,
+        default=8,
+        help="maximum number of GPUs per engine",
+    )
     parser.add_argument(
         "--isl", type=int, default=3000, help="target input sequence length"
     )
@@ -121,10 +133,11 @@
     with open(args.config, "r") as f:
         config = yaml.safe_load(f)
 
-    # Get the number of available GPUs
-    available_gpus = get_available_gpu_count()
-
-    profile_tp_size = [2**i for i in range(int(math.log2(available_gpus)) + 1)]
+    profile_tp_size = [
+        2**i
+        for i in range(int(math.log2(args.max_num_gpus_per_engine)) + 1)
+        if args.min_num_gpus_per_engine <= 2**i <= args.max_num_gpus_per_engine
+    ]
     logger.info(f"Profiling TP sizes: {profile_tp_size}")
 
     os.makedirs(args.output_dir, exist_ok=True)

benchmarks/profiler/utils/config.py

@@ -14,8 +14,10 @@
 # limitations under the License.
 
 import logging
+from copy import deepcopy
 from typing import Literal
 
+from utils.defaults import DEFAULT_MODEL_NAME, DYNAMO_RUN_DEFAULT_PORT
 from dynamo.planner.defaults import WORKER_COMPONENT_NAMES
 
 logger = logging.getLogger(__name__)
@@ -28,190 +30,84 @@
 console_handler.setFormatter(formatter)
 logger.addHandler(console_handler)
 
-
-class VllmV0ConfigModifier:
+class VllmV1ConfigModifier:
     @classmethod
     def convert_config(cls, config: dict, target: Literal["prefill", "decode"]) -> dict:
-        config = config.copy()
+        config = deepcopy(config)
 
         # disable planner
-        if "Planner" in config:
-            config["Planner"]["no-operation"] = True
+        if "Planner" in config["spec"]["services"]:
+            del config["spec"]["services"]["Planner"]
 
         if target == "prefill":
-            if WORKER_COMPONENT_NAMES["vllm_v0"].prefill_worker in config:
-                # make PrefillWorker into VllmWorker
-                del config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker]
-                config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker] = config[
-                    WORKER_COMPONENT_NAMES["vllm_v0"].prefill_worker
-                ]
-                del config[WORKER_COMPONENT_NAMES["vllm_v0"].prefill_worker]
-
-            # to profile prefill, we disable prefix caching
-            config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker][
-                "enable-prefix-caching"
-            ] = False
-        elif target == "decode":
-            if WORKER_COMPONENT_NAMES["vllm_v0"].prefill_worker in config:
-                del config[WORKER_COMPONENT_NAMES["vllm_v0"].prefill_worker]
+            # convert prefill worker into decode worker
+            config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker] = config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker]
+            del config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker]
 
-            # to profile prefill, we enable prefix caching to pass the prefill stage
-            config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker][
-                "enable-prefix-caching"
-            ] = True
+            args = config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]["extraPodSpec"]["mainContainer"]["args"]
 
-        # set num workers to 1
-        config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker]["ServiceArgs"][
-            "workers"
-        ] = 1
-
-        # set PP to 1
-        if (
-            "pipeline-parallel-size"
-            in config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker]
-            and config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker][
-                "pipeline-parallel-size"
-            ]
-            > 1
-        ):
-            logger.warning("Currently we only support TP, setting PP to 1")
-            config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker][
-                "pipeline-parallel-size"
-            ] = 1
-
-        # always local prefill
-        config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker][
-            "remote-prefill"
-        ] = False
-        config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker][
-            "conditional-disagg"
-        ] = False
+            # remove --is-prefill-worker flag
+            args.remove("--is-prefill-worker")
 
-        return config
+            # disable prefix caching
+            if "--enable-prefix-caching" in args:
+                args.remove("--enable-prefix-caching")
+            if "--no-enable-prefix-caching" not in args:
+                args.append("--no-enable-prefix-caching")
 
-    @classmethod
-    def set_config_tp_size(cls, config: dict, tp_size: int):
-        config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker][
-            "tensor-parallel-size"
-        ] = tp_size
-        config[WORKER_COMPONENT_NAMES["vllm_v0"].decode_worker]["ServiceArgs"][
-            "resources"
-        ]["gpu"] = tp_size
-        return config
-
-    @classmethod
-    def get_model_name(cls, config: dict) -> str:
-        if "Common" in config and "served_model_name" in config["Common"]:
-            return config["Common"]["served_model_name"]
-        else:
-            return config["Frontend"]["served_model_name"]
-
-    @classmethod
-    def get_port(cls, config: dict) -> int:
-        if "Common" in config and "port" in config["Common"]:
-            return config["Common"]["port"]
-        else:
-            return config["Frontend"]["port"]
-
-    @classmethod
-    def get_kv_cache_size_from_dynamo_log(cls, dynamo_log_fn: str) -> int:
-        try:
-            with open(dynamo_log_fn, "r") as f:
-                for line in f:
-                    if "Maximum concurrency for" in line:
-                        line = line.strip().split("Maximum concurrency for ")[1]
-                        token_count = int(line.split(" tokens per request: ")[0])
-                        concurrency = float(line.split(" tokens per request: ")[1][:-1])
-
-                        logger.info(
-                            f"Found KV cache info: {token_count} x {concurrency} = {int(token_count * concurrency)}"
-                        )
-                        return int(token_count * concurrency)
-        except Exception as e:
-            logger.warning(
-                f"Failed to parse KV cache size from line: {line}. Error: {e}"
-            )
-        return 0
-
-
-class VllmV1ConfigModifier:
-    @classmethod
-    def convert_config(cls, config: dict, target: Literal["prefill", "decode"]) -> dict:
-        config = config.copy()
-
-        # disable planner
-        if "Planner" in config:
-            config["Planner"]["no-operation"] = True
-
-        # turn-off disagg
-        config["SimpleLoadBalancer"]["enable_disagg"] = False
-
-        if target == "prefill":
-            if WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker in config:
-                # make VllmPrefillWorker into VllmDecodeWorker
-                del config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]
-                config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker] = config[
-                    WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker
-                ]
-                del config[WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker]
-
-            # to profile prefill, we disable prefix caching
-            config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker][
-                "enable-prefix-caching"
-            ] = False
         elif target == "decode":
-            if WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker in config:
-                del config[WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker]
+            # delete prefill worker
+            del config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].prefill_worker]
 
-            # to profile prefill, we enable prefix caching to pass the prefill stage
-            config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker][
-                "enable-prefix-caching"
-            ] = True
+            args = config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]["extraPodSpec"]["mainContainer"]["args"]
+
+            # enable prefix caching
+            if "--enable-prefix-caching" not in args:
+                args.append("--enable-prefix-caching")
+            if "--no-enable-prefix-caching" in args:
+                args.remove("--no-enable-prefix-caching")
 
         # set num workers to 1
-        config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]["ServiceArgs"][
-            "workers"
-        ] = 1
-
-        # set PP to 1
-        if (
-            "pipeline-parallel-size"
-            in config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]
-            and config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker][
-                "pipeline-parallel-size"
-            ]
-            > 1
-        ):
-            logger.warning("Currently we only support TP, setting PP to 1")
-            config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker][
-                "pipeline-parallel-size"
-            ] = 1
+        decode_worker_config = config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]
+        decode_worker_config["replicas"] = 1
 
         return config
 
     @classmethod
     def set_config_tp_size(cls, config: dict, tp_size: int):
-        config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker][
-            "tensor-parallel-size"
-        ] = tp_size
-        config[WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]["ServiceArgs"][
-            "resources"
-        ]["gpu"] = tp_size
+        config = deepcopy(config)
+
+        args = config["spec"]["services"][WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker]["extraPodSpec"]["mainContainer"]["args"]
+
+        try:
+            idx = args.index("--tensor-parallel-size")
+            args[idx + 1] = str(tp_size)
+        except ValueError:
+            args.append("--tensor-parallel-size")
+            args.append(str(tp_size))
+
         return config
 
     @classmethod
     def get_model_name(cls, config: dict) -> str:
-        if "Common" in config and "served_model_name" in config["Common"]:
-            return config["Common"]["served_model_name"]
-        else:
-            return config["Frontend"]["served_model_name"]
+        worker_name = WORKER_COMPONENT_NAMES["vllm_v1"].decode_worker
+        args = config["spec"]["services"][worker_name]["extraPodSpec"]["mainContainer"]["args"]
+
+        for i, arg in enumerate(args):
+            if arg == "--model" and i + 1 < len(args):
+                return args[i + 1]
+
+        logger.warning(f"Model name not found in configuration args, using default model name: {DEFAULT_MODEL_NAME}")
+        return DEFAULT_MODEL_NAME
 
     @classmethod
     def get_port(cls, config: dict) -> int:
-        if "Common" in config and "port" in config["Common"]:
-            return config["Common"]["port"]
-        else:
-            return config["Frontend"]["port"]
+        args = config["spec"]["services"]["Frontend"]["extraPodSpec"]["mainContainer"]["args"]
+        for arg in args:
+            if arg.startswith("port="):
+                return int(arg.split("=")[1])
+        logger.warning(f"Port not found in configuration args, using default port: {DYNAMO_RUN_DEFAULT_PORT}")
+        return DYNAMO_RUN_DEFAULT_PORT
 
     @classmethod
     def get_kv_cache_size_from_dynamo_log(cls, dynamo_log_fn: str) -> int:
@@ -237,6 +133,5 @@ def get_kv_cache_size_from_dynamo_log(cls, dynamo_log_fn: str) -> int:
 
 
 CONFIG_MODIFIERS = {
-    "vllm_v0": VllmV0ConfigModifier,
     "vllm_v1": VllmV1ConfigModifier,
 }

benchmarks/profiler/utils/defaults.py

@@ -29,3 +29,6 @@
     450,
     500,
 ]
+
+DEFAULT_MODEL_NAME = "Qwen/Qwen3-0.6B"
+DYNAMO_RUN_DEFAULT_PORT = 8080