feat: FMA actuation path classification, hit rates, and per-path timing

config/templates/jinja/20_harness_pod.yaml.j2

@@ -124,6 +124,8 @@ spec:
       value: "{{ fma.launcherConfigurator.port }}"
     - name: LLMDBENCH_FMA_ITERATIONS
       value: "{{ fma.iterations }}"
+    - name: LLMDBENCH_FMA_MAX_INSTANCES
+      value: "{{ fma.launcher.maxInstances }}"
 {% endif %}
 {% if harness.extraEnvVars is defined and harness.extraEnvVars %}
 {% for env_entry in harness.extraEnvVars %}

config/templates/jinja/24_fma-deployment.yaml.j2

@@ -44,7 +44,7 @@ metadata:
 {% endif %}
   namespace: {{ namespace.name }}
 spec:
-  maxSleepingInstances: 3
+  maxInstances: {{ fma.launcher.maxInstances }}
   podTemplate:
     metadata:
 {% if fma.launcher.podTemplate.metadata.labels is defined %}

config/templates/values/defaults.yaml

@@ -1305,6 +1305,7 @@ fma:
     limitsMemory: 250Mi
 
   launcher:
+    maxInstances: 4
     image:
       repository: ghcr.io/llm-d-incubation/llm-d-fast-model-actuation/launcher
       tag: v0.6.0-alpha.13

llmdbenchmark/analysis/benchmark_report/native_to_br0_1.py

@@ -2323,6 +2323,7 @@ def _import_categories(cat_list: list[dict[str, Any]]) -> list[dict[str, Any]]:
                 "deploy_methods": results["scenario"]["deploy_methods"],
                 "load_format": results["scenario"]["load_format"],
                 "sleep_mode": results["scenario"]["sleep_mode"],
+                "max_instances": results["scenario"].get("max_instances", 0),
                 "gpus": results["scenario"]["gpus"],
             },
         },
@@ -2548,9 +2549,41 @@ def _import_categories(cat_list: list[dict[str, Any]]) -> list[dict[str, Any]]:
                 info["launcher_endpoint"] = launcher_info["launcher_endpoint"]
                 info["vllm_endpoint"] = launcher_info["vllm_endpoint"]
                 info["ttft"] = {"units": Units.S, "value": launcher_info["ttft"]}
+                info["launcher_creation_timestamp"] = {
+                    "units": Units.S,
+                    "value": launcher_info.get("launcher_creation_timestamp", 0.0),
+                }
+                info["launcher_node"] = launcher_info.get("launcher_node", "")
+                if launcher_info.get("t_wake") is not None:
+                    info["t_wake"] = {
+                        "units": Units.S,
+                        "value": launcher_info["t_wake"],
+                    }
+                if launcher_info.get("t_instance_create") is not None:
+                    info["t_instance_create"] = {
+                        "units": Units.S,
+                        "value": launcher_info["t_instance_create"],
+                    }
+                if launcher_info.get("t_cold_launcher") is not None:
+                    info["t_cold_launcher"] = {
+                        "units": Units.S,
+                        "value": launcher_info["t_cold_launcher"],
+                    }
                 launcher_infos.append(info)
 
             it["launcher_infos"] = launcher_infos
+            it["hot_hit_rate"] = {
+                "units": Units.COUNT,
+                "value": iteration.get("hot_hit_rate", 0.0),
+            }
+            it["warm_hit_rate"] = {
+                "units": Units.COUNT,
+                "value": iteration.get("warm_hit_rate", 0.0),
+            }
+            it["cold_launcher_hit_rate"] = {
+                "units": Units.COUNT,
+                "value": iteration.get("cold_launcher_hit_rate", 0.0),
+            }
             iterations.append(it)
 
         metadata_dict["iterations"] = iterations

llmdbenchmark/analysis/scripts/nop-analyze_results.py

@@ -121,6 +121,9 @@ def write_benchmark_scenario(file: io.TextIOWrapper, scenario: Scenario):
     file.write(f"  Harness              : {scenario.load.name}\n")
     file.write(f"  Load Format          : {scenario.metadata['load_format']}\n")
     file.write(f"  Sleep Mode On        : {scenario.metadata['sleep_mode']}\n")
+    file.write(
+        f"  Max Instances        : {scenario.metadata.get('max_instances', 'N/A')}\n"
+    )
     file.write(f"  Model                : {scenario.model.name}\n")
     for engine in scenario.platform.engine:
         file.write("  Engine\n")
@@ -293,60 +296,82 @@ def write_fma_metrics(  # pylint: disable=too-many-locals,too-many-statements
 
     file.write("\n\n")
     file.write("Actuation Conditions:\n")
-    file.write(
-        "  T_luke_warm: when new launcher created by Dual Pod Controller + new vLLM\n"
-    )
-    file.write("  T_warm: when existing launcher creates new vLLM\n")
-    file.write("  T_hot: when waking up sleeping vLLM\n\n")
+    file.write("  T_cold_launcher: DPC creates new launcher + new vLLM instance\n")
+    file.write("  T_warm: existing launcher creates new vLLM instance\n")
+    file.write("  T_hot: waking sleeping vLLM instance\n\n")
     file.write("T_actuation: Time for the Requester Pod to be ready\n")
-    file.write("TTRD: Time for the Requester Pod to have dual label set\n")
+    file.write("T_hot: Hot-start timing (upper bound)\n")
+    file.write("T_warm: Warm-start timing (upper bound)\n")
+    file.write("T_cold_launcher: Cold-start-with-launcher timing (upper bound)\n")
     file.write("T_first_token: Time for vLLM server to return first token\n")
-    file.write("TTRD + T_first_token == T_e2e\n")
     file.write("Each iteration scales ReplicaSet from 0 to 1 and then from 1 to 0\n")
-    file.write("Hit_rate (count(hot starts) / total iterations)\n")
 
     hot_starts = 0
+    warm_starts = 0
+    cold_starts = 0
     total_iterations = len(iterations)
     pandas_datas = []
     for iteration in iterations:
         for launcher_info in iteration["launcher_infos"]:
             ct = float(launcher_info["requester_info"]["creation_timestamp"]["value"])
             rt = float(launcher_info["requester_info"]["ready_timestamp"]["value"])
-            dt = float(launcher_info["requester_info"]["dual_label_timestamp"]["value"])
             ttrr = rt - ct if rt > 0.0 else 0.0
-            ttrd = dt - ct if dt > 0.0 else 0.0
             ttft = float(launcher_info["ttft"]["value"])
             actuation_condition = launcher_info["actuation_condition"]
             if actuation_condition == "T_hot":
                 hot_starts += 1
+            elif actuation_condition == "T_warm":
+                warm_starts += 1
+            elif actuation_condition == "T_cold_launcher":
+                cold_starts += 1
+
+            t_hot = launcher_info.get("t_wake", {})
+            t_hot_val = float(t_hot["value"]) if isinstance(t_hot, dict) else None
+            t_warm = launcher_info.get("t_instance_create", {})
+            t_warm_val = float(t_warm["value"]) if isinstance(t_warm, dict) else None
+            t_cold = launcher_info.get("t_cold_launcher", {})
+            t_cold_val = float(t_cold["value"]) if isinstance(t_cold, dict) else None
+            node = launcher_info.get("launcher_node", "")
 
             pandas_datas.append(
                 {
                     "Iteration": iteration["iteration"]["value"],
-                    "vLLM Name": launcher_info["name"],
+                    "Node": node,
                     "Actuation Condition": actuation_condition,
-                    "T_actuation(secs)": ttrr,
-                    "TTRD(secs)": ttrd,
-                    "T_first_token(secs)": ttft,
-                    "T_e2e": ttrd + ttft,
+                    "T_actuation(s)": ttrr,
+                    "T_hot(s)": t_hot_val,
+                    "T_warm(s)": t_warm_val,
+                    "T_cold(s)": t_cold_val,
+                    "T_first_token(s)": ttft,
                 }
             )
 
-    hit_rate = hot_starts / total_iterations if total_iterations > 0 else 0.0
+    hot_hit_rate = hot_starts / total_iterations if total_iterations > 0 else 0.0
+    warm_hit_rate = warm_starts / total_iterations if total_iterations > 0 else 0.0
+    cold_hit_rate = cold_starts / total_iterations if total_iterations > 0 else 0.0
 
     df = pd.DataFrame(pandas_datas)
 
     file.write("\n")
 
     # Float formatting
-    float_columns = ["T_actuation(secs)", "TTRD(secs)", "T_first_token(secs)", "T_e2e"]
-
-    # Compute column widths dynamically
+    float_columns = [
+        "T_actuation(s)",
+        "T_hot(s)",
+        "T_warm(s)",
+        "T_cold(s)",
+        "T_first_token(s)",
+    ]
+
+    # Compute column widths dynamically.
+    # FMA per-path columns (T_hot, T_warm, T_cold) contain None for
+    # inapplicable paths, so we handle NaN when computing widths.
     col_widths = {}
     for col in df.columns:
         if col in float_columns:
-            # max width between header and max formatted float
-            max_float_width = max(df[col].apply(lambda x: len(f"{x:.4f}")))
+            max_float_width = max(
+                df[col].apply(lambda x: len(f"{x:.4f}") if pd.notna(x) else 2)
+            )
             col_widths[col] = max(len(col), max_float_width)
         else:
             col_widths[col] = max(len(col), df[col].astype(str).apply(len).max())
@@ -365,20 +390,25 @@ def write_fma_metrics(  # pylint: disable=too-many-locals,too-many-statements
     )
     file.write(f"{' ' * left_padding}{separator}\n")
 
-    # Rows
+    # Rows -- per-path timing columns show "--" when not applicable
     for _, r in df.iterrows():
         row = []
         for col in df.columns:
             val = r[col]
             if col in float_columns:
-                row.append(f"{val:>{col_widths[col]}.4f}")  # right-align numbers
+                if pd.notna(val):
+                    row.append(f"{val:>{col_widths[col]}.4f}")  # right-align numbers
+                else:
+                    row.append(f"{'--':>{col_widths[col]}}")  # N/A for this path
             elif isinstance(val, int):
                 row.append(f"{val:>{col_widths[col]}}")
             else:
-                row.append(f"{val:<{col_widths[col]}}")  # left-align strings
+                row.append(f"{val:<{col_widths[col]}}")
         file.write(f"{' ' * left_padding}{(' ' * space_between_cols).join(row)}\n")
 
-    file.write(f"\nHit_rate: {hit_rate:8.2f}\n")
+    file.write(f"\n  Hot_hit_rate:            {hot_hit_rate:.2f}\n")
+    file.write(f"  Warm_hit_rate:           {warm_hit_rate:.2f}\n")
+    file.write(f"  Cold_launcher_hit_rate:  {cold_hit_rate:.2f}\n")
 
     file.write("\n")
 

workload/harnesses/fma_functions.py

@@ -78,6 +78,11 @@ class FMALauncherInfo:  # pylint: disable=too-many-instance-attributes
     vllm_endpoint: str = ""
     ttft: float = 0.0
     actuation_condition: FMAActuationCondition | None = None
+    launcher_creation_timestamp: float = 0.0
+    launcher_node: str = ""
+    t_wake: float | None = None
+    t_instance_create: float | None = None
+    t_cold_launcher: float | None = None
 
     def dump(self) -> dict[str, Any]:
         """Convert FMALauncherInfo to dict.
@@ -102,9 +107,9 @@ def dump(self) -> dict[str, Any]:
 class FMAActuationCondition(StrEnum):
     """Type of actuation"""
 
-    T_LUKE_WARM = "T_luke_warm"  # when new launcher created by DPC + new vllm
-    T_WARM = "T_warm"  # when existing launcher creates new vllm
-    T_HOT = "T_hot"  # when waking up sleeping vllm
+    T_COLD_LAUNCHER = "T_cold_launcher"  # DPC creates new launcher + new vllm
+    T_WARM = "T_warm"  # existing launcher creates new vllm
+    T_HOT = "T_hot"  # waking sleeping vllm
 
     def dump(self) -> str:
         """Convert FMAActuationCondition to str.
@@ -121,6 +126,9 @@ class FMAMetricsIteration:
 
     iteration: int
     launcher_infos: list[FMALauncherInfo]
+    hot_hit_rate: float = 0.0
+    warm_hit_rate: float = 0.0
+    cold_launcher_hit_rate: float = 0.0
 
     def dump(self) -> dict[str, Any]:
         """Convert FMAMetricsIteration to dict.
@@ -268,6 +276,12 @@ def get_fma_launcher_infos(  # pylint: disable=too-many-locals,too-many-argument
                 launcher_info.container_name = container.name
                 launcher_info.name = engine.name
                 launcher_info.requester_info = requester_info
+                launcher_info.launcher_creation_timestamp = (
+                    launcher_pod.metadata.creation_timestamp.astimezone(
+                        timezone.utc
+                    ).timestamp()
+                )
+                launcher_info.launcher_node = launcher_pod.spec.node_name or ""
                 launcher_info.launcher_endpoint = (
                     f"http://{launcher_pod_ip}:{fma_launcher_port}"
                 )
@@ -832,20 +846,73 @@ def benchmark_fma(  # pylint: disable=too-many-arguments,too-many-positional-arg
                                     FMAActuationCondition.T_HOT
                                 )
 
-                        # TODO: Improve the warm/luke_warm check instead of pod name
                         if launcher_info.actuation_condition is None:
-                            launcher_info.actuation_condition = (
-                                FMAActuationCondition.T_WARM
-                                if launcher_info.name.startswith("launcher-fma-")
-                                else FMAActuationCondition.T_LUKE_WARM
+                            if (
+                                launcher_info.launcher_creation_timestamp > 0.0
+                                and launcher_info.requester_info.creation_timestamp
+                                > 0.0
+                                and launcher_info.launcher_creation_timestamp
+                                < launcher_info.requester_info.creation_timestamp
+                            ):
+                                launcher_info.actuation_condition = (
+                                    FMAActuationCondition.T_WARM
+                                )
+                            else:
+                                launcher_info.actuation_condition = (
+                                    FMAActuationCondition.T_COLD_LAUNCHER
+                                )
+
+                        # Compute per-path timing (upper bound via Kube timestamps)
+                        ready_ts = launcher_info.requester_info.ready_timestamp
+                        creation_ts = launcher_info.requester_info.creation_timestamp
+                        if (
+                            launcher_info.actuation_condition
+                            == FMAActuationCondition.T_HOT
+                            and ready_ts > 0.0
+                        ):
+                            launcher_info.t_wake = ready_ts - creation_ts
+                        elif (
+                            launcher_info.actuation_condition
+                            == FMAActuationCondition.T_WARM
+                            and ready_ts > 0.0
+                        ):
+                            launcher_info.t_instance_create = ready_ts - creation_ts
+                        elif (
+                            launcher_info.actuation_condition
+                            == FMAActuationCondition.T_COLD_LAUNCHER
+                            and ready_ts > 0.0
+                            and launcher_info.launcher_creation_timestamp > 0.0
+                        ):
+                            launcher_info.t_cold_launcher = (
+                                ready_ts - launcher_info.launcher_creation_timestamp
                             )
 
                     except Exception as e:
                         raise RuntimeError(
                             f"error on benchmark FMA '{launcher_info.name}' launcher"
                         ) from e
 
-                fma_metrics_iteration = FMAMetricsIteration(iteration, launcher_infos)
+                # Compute hit rates for this iteration
+                total = len(launcher_infos)
+                hot_count = sum(
+                    li.actuation_condition == FMAActuationCondition.T_HOT
+                    for li in launcher_infos
+                )
+                warm_count = sum(
+                    li.actuation_condition == FMAActuationCondition.T_WARM
+                    for li in launcher_infos
+                )
+                cold_count = sum(
+                    li.actuation_condition == FMAActuationCondition.T_COLD_LAUNCHER
+                    for li in launcher_infos
+                )
+                fma_metrics_iteration = FMAMetricsIteration(
+                    iteration,
+                    launcher_infos,
+                    hot_hit_rate=hot_count / total if total > 0 else 0.0,
+                    warm_hit_rate=warm_count / total if total > 0 else 0.0,
+                    cold_launcher_hit_rate=cold_count / total if total > 0 else 0.0,
+                )
                 fma_metrics.iterations.append(fma_metrics_iteration)
             finally:
                 logger.info("Benchmark FMA iteration '%d' end.", iteration)

workload/harnesses/nop-llm-d-benchmark.py

@@ -82,7 +82,13 @@ def main():
             ]
         )
     if fma_enabled:
-        keys.extend(["LLMDBENCH_FMA_LAUNCHER_CONFIG_PORT", "LLMDBENCH_FMA_ITERATIONS"])
+        keys.extend(
+            [
+                "LLMDBENCH_FMA_LAUNCHER_CONFIG_PORT",
+                "LLMDBENCH_FMA_ITERATIONS",
+                "LLMDBENCH_FMA_MAX_INSTANCES",
+            ]
+        )
 
     envs.update(get_env_variables(keys))
     logger.info("Environment variables:")
@@ -106,6 +112,7 @@ def main():
 
     fma_launcher_port = envs.get("LLMDBENCH_FMA_LAUNCHER_CONFIG_PORT", "0")
     fma_iterations = int(envs.get("LLMDBENCH_FMA_ITERATIONS", "0"))
+    fma_max_instances = int(envs.get("LLMDBENCH_FMA_MAX_INSTANCES", "0"))
 
     deploy_methods = []
     if standalone_enabled:
@@ -127,6 +134,7 @@ def main():
 
     benchmark_result = BenchmarkResult()
     benchmark_result.scenario.deploy_methods = ",".join(deploy_methods)
+    benchmark_result.scenario.max_instances = fma_max_instances
     if fma_enabled:
         try:
             logger.info("Benchmark FMA launcher start...")

workload/harnesses/nop_functions.py

@@ -545,6 +545,7 @@ class BenchmarkScenario:
     deploy_methods: str = ""
     load_format: LoadFormat = LoadFormat.UNKNOWN
     sleep_mode: bool = False
+    max_instances: int = 0
     model: ModelScenario = field(default_factory=ModelScenario)
     platform: PlatformScenario = field(default_factory=PlatformScenario)
     gpus: list[GPUScenario] = field(default_factory=list[GPUScenario])