diff --git a/AGENTS.md b/AGENTS.md index deaf00b..c379dda 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -282,7 +282,8 @@ simulation-runs/sim-2026-04-02_14_30_45/ "scale_latency": 40, "max_workers": 50, "policy": "MaxPrefix", // RoundRobin, LeastLoaded, Random, MaxPrefix, Balanced - "periodic_infra_update_collection_time": 30, + "latency_percentile": "p95", // p90, p95, p99 + "latency_window": 50, // last N completed requests for TTFT/ITL SLOs "max_event_batch_size": 64 } }, diff --git a/configs/defaults.json b/configs/defaults.json index f3d13e8..c7ea97b 100644 --- a/configs/defaults.json +++ b/configs/defaults.json @@ -21,7 +21,8 @@ "scale_latency": 40, "max_workers": 50, "policy": "MaxPrefix", - "periodic_infra_update_collection_time": 30, + "latency_percentile": "p95", + "latency_window": 50, "max_event_batch_size": 64 } }, @@ -56,7 +57,7 @@ "worker_params":{ "worker_local_queue_capacity": 1, - "periodic_infra_update_time": 30, + "periodic_infra_update_time": 5, "kvcevent_coalesce_time": 30 }, diff --git a/configs/defaults_otel.json b/configs/defaults_otel.json index f3dbe59..9fbfebf 100644 --- a/configs/defaults_otel.json +++ b/configs/defaults_otel.json @@ -21,7 +21,8 @@ "scale_latency": 40, "max_workers": 50, "policy": "MaxPrefix", - "periodic_infra_update_collection_time": 30, + "latency_percentile": "p95", + "latency_window": 50, "max_event_batch_size": 64 } }, @@ -59,7 +60,7 @@ "worker_params":{ "worker_local_queue_capacity": 1, - "periodic_infra_update_time": 30, + "periodic_infra_update_time": 5, "kvcevent_coalesce_time": 30 }, diff --git a/configs/hf.json b/configs/hf.json index 5702cae..2a18c4d 100644 --- a/configs/hf.json +++ b/configs/hf.json @@ -20,7 +20,8 @@ "scale_latency": 40, "max_workers": 50, "policy": "MaxPrefix", - "periodic_infra_update_collection_time": 30, + "latency_percentile": "p95", + "latency_window": 50, "max_event_batch_size": 64 } }, @@ -55,7 +56,7 @@ "worker_params":{ "worker_local_queue_capacity": 1, - "periodic_infra_update_time": 30, + "periodic_infra_update_time": 5, "kvcevent_coalesce_time": 30 }, diff --git a/opal/core/events.py b/opal/core/events.py index 94c1717..ef8f9d7 100644 --- a/opal/core/events.py +++ b/opal/core/events.py @@ -34,5 +34,8 @@ class SystemEvent(OpalInfraEvent): # 0 = min, 1 = max load: float ingress_queue_occupancy: float - mem_used: float gpu_utilization: float + # kvc_utilization: fraction of GPU KV-cache blocks in use (1 - free/total), in [0, 1]. + kvc_utilization: float = 0.0 + # queue_depth: absolute in-flight request count on the worker (waiting + running). + queue_depth: int = 0 diff --git a/opal/router/router.py b/opal/router/router.py index 1824e80..b591378 100644 --- a/opal/router/router.py +++ b/opal/router/router.py @@ -9,6 +9,7 @@ from opal.core.events import OpalInfraEvent, KVCEvent, SystemEvent from opal.kvcache.kvbm import KVBM from opal.core.request import LLMRequest +from opal.stats.metrics import MetricsSnapshot from opal.worker.vllm_worker import LLMWorkerVLLMScheduler from opal.utils.util import parse_bool, safe_process @@ -39,6 +40,19 @@ def _get_policy_func(self): f"{policy} no such routing policy. Supported policies are: RoundRobin, LeastLoaded, Random, MaxPrefix, Balanced" ) + def _get_latency_percentile(self) -> int: + name = str(self.opalConfig["router"]["router_params"]["latency_percentile"]).lower() + mapping = {"p90": 90, "p95": 95, "p99": 99} + if name not in mapping: + raise Exception(f"{name} is not a supported latency percentile. Supported: p90, p95, p99") + return mapping[name] + + def _get_latency_window(self) -> int: + window = int(self.opalConfig["router"]["router_params"]["latency_window"]) + if window <= 0: + raise Exception(f"latency_window must be a positive integer, got {window}") + return window + def __init__(self, opal_env, opal_config): self.opal_env = opal_env self.opalConfig = opal_config @@ -51,12 +65,16 @@ def __init__(self, opal_env, opal_config): self.results_queue = simpy.Store(self.sim_env) # leave infinite capacity for this self._event_queue = simpy.Store(self.sim_env) - self.periodic_infra_update_collection_time = self.opalConfig["router"]["router_params"][ - "periodic_infra_update_collection_time" - ] self._kvbm = KVBM(self.opal_env) self._worker_stats = None self._policy_func = self._get_policy_func() + self._latency_percentile = self._get_latency_percentile() + self._latency_window = self._get_latency_window() + + # Last SystemEvent reported by each worker and the latest + # MetricsSnapshot built from them. + self._latest_worker_metrics: dict[int, SystemEvent] = {} + self._latest_metrics: MetricsSnapshot | None = None self.num_workers = self.opalConfig["simulation"]["num_workers"] self._worker_cls = LLMWorkerVLLMScheduler @@ -149,6 +167,25 @@ def _per_second_stats(self): stats.add_per_unit_gpu_utilization(utilization) self.log.debug(f"Breaking the per second stats loop at {self.sim_env.now}") + def _pool_metrics_snapshot(self, stats): + """Create a MetricsSnapshot from the last SystemEvent each worker pushed, + plus SLO percentiles over the configured sliding window of completions. + + Called when process_events() drains a batch that contains SystemEvents. + """ + ttft, itl = stats.recent_ttft_itl(self._latency_percentile, self._latency_window) + snapshot = MetricsSnapshot( + timestamp=self.sim_env.now, + queue_depth_per_worker={wid: ev.queue_depth for wid, ev in self._latest_worker_metrics.items()}, + kvc_util_per_worker={wid: ev.kvc_utilization for wid, ev in self._latest_worker_metrics.items()}, + percentile=self._latency_percentile, + window=self._latency_window, + ttft_secs=ttft, + itl_secs=itl, + ) + self._latest_metrics = snapshot + stats.add_metrics_snapshot(snapshot.to_dict()) + def _policy_leastloaded(self, req: LLMRequest): queue_size = min(self._outstanding_requests_per_worker.values()) worker = next((k for k, v in self._outstanding_requests_per_worker.items() if v == queue_size), None) @@ -245,6 +282,16 @@ def queue_events(self, elist: list[OpalInfraEvent], delay: float = 0): for elem in elist: yield self._event_queue.put(elem) + def _ingest_event(self, e: OpalInfraEvent, kvbm_events: list[KVCEvent], systems_events: list[SystemEvent]): + if isinstance(e, KVCEvent): + kvbm_events.append(e) + elif isinstance(e, SystemEvent): + systems_events.append(e) + # Store last reported SystemEvent from this worker. Used to create telemetry snapshot + self._latest_worker_metrics[e.worker_id] = e + else: + raise Exception(f"Unknown event type {type(e)}") + def process_events(self): max_event_batch_size = self.opalConfig["router"]["router_params"]["max_event_batch_size"] @@ -252,31 +299,22 @@ def process_events(self): kvbm_events = [] systems_events = [] - # Collect events up to batch size + e = yield self._event_queue.get() + self._ingest_event(e, kvbm_events, systems_events) + while len(self._event_queue.items) > 0 and not self.opal_env.are_we_done(): - # Stop if either batch is full if len(kvbm_events) >= max_event_batch_size or len(systems_events) >= max_event_batch_size: break - e = yield self._event_queue.get() - if isinstance(e, KVCEvent): - kvbm_events.append(e) - elif isinstance(e, SystemEvent): - systems_events.append(e) - else: - raise Exception(f"Unknown event type {type(e)}") - - # Process batches if we have any events + self._ingest_event(e, kvbm_events, systems_events) + if kvbm_events or systems_events: self._kvbm.process_kvc_events(kvbm_events) self._kvbm.process_system_events(systems_events) - # If queue still has items, continue immediately - if len(self._event_queue.items) > 0: - continue - - # Otherwise sleep until next periodic check - yield self.sim_env.timeout(self.periodic_infra_update_collection_time) + if systems_events: + stats = self.opal_env.workload_orchestrator.get_active_stage_stats() + self._pool_metrics_snapshot(stats) def shutdown(self): self._stats_request_allocated_per_worker = dict(sorted(self._stats_request_allocated_per_worker.items())) diff --git a/opal/stats/metrics.py b/opal/stats/metrics.py new file mode 100644 index 0000000..c598df5 --- /dev/null +++ b/opal/stats/metrics.py @@ -0,0 +1,48 @@ +# SPDX-License-Identifier: Apache-2.0 +from __future__ import annotations +from dataclasses import dataclass, field + + +@dataclass +class MetricsSnapshot: + """A point-in-time view of cluster telemetry as seen by the router. + + Built when workers push SystemEvent telemetry. + Worker fields are the last report from each worker (delayed by + periodic_infra_update_time). TTFT/ITL are the configured percentile over a + sliding window of recent completed requests. + + Units: + - queue_depth_per_worker: absolute in-flight request count per worker. + - kvc_util_per_worker: fraction in [0, 1]. + - ttft_secs / itl_secs: seconds; -1 until `window` requests have completed. + """ + + timestamp: float = 0.0 + queue_depth_per_worker: dict[int, int] = field(default_factory=dict) + kvc_util_per_worker: dict[int, float] = field(default_factory=dict) + percentile: int = 95 # 90, 95, or 99 + window: int = 50 # last N completed requests used for ttft/itl + ttft_secs: float = -1.0 + itl_secs: float = -1.0 + + @property + def max_queue_depth(self) -> int: + return max(self.queue_depth_per_worker.values(), default=0) + + @property + def max_kvc_util(self) -> float: + return max(self.kvc_util_per_worker.values(), default=0.0) + + def to_dict(self) -> dict: + return { + "timestamp": self.timestamp, + "queue_depth_per_worker": dict(self.queue_depth_per_worker), + "kvc_util_per_worker": dict(self.kvc_util_per_worker), + "percentile": self.percentile, + "window": self.window, + "ttft_secs": self.ttft_secs, + "itl_secs": self.itl_secs, + "max_queue_depth": self.max_queue_depth, + "max_kvc_util": self.max_kvc_util, + } diff --git a/opal/stats/stage_statistics.py b/opal/stats/stage_statistics.py index e86951c..ba0ee07 100644 --- a/opal/stats/stage_statistics.py +++ b/opal/stats/stage_statistics.py @@ -32,6 +32,8 @@ def __init__(self): self.per_unit_req_done = [] self.per_unit_workers = [] self.per_unit_gpu_utilization = [] + # List of all MetricsSnapshot objects recorded during the simulation. + self.metrics_snapshots = [] while val < max_bin: self.bins.append(val) val *= 2 @@ -59,6 +61,7 @@ def to_dict(self): "per_unit_throughput": self.per_unit_req_done, "per_unit_workers": self.per_unit_workers, "per_unit_gpu_utilization": self.per_unit_gpu_utilization, + "metrics_snapshots": self.metrics_snapshots, # numpy array -> list "latencies": self.latencies.tolist(), "stage_time_start": self.stage_time_start, @@ -85,6 +88,7 @@ def from_dict(cls, data): obj.per_unit_req_done = data["per_unit_throughput"] obj.per_unit_workers = data["per_unit_workers"] obj.per_unit_gpu_utilization = data["per_unit_gpu_utilization"] + obj.metrics_snapshots = data.get("metrics_snapshots", []) obj.stage_time_end = data["stage_time_end"] obj.stage_time_start = data["stage_time_start"] obj.kvc_tier_tokens = defaultdict(int, data.get("kvc_tier_tokens", {})) @@ -226,6 +230,25 @@ def add_per_unit_workers(self, worker_count: int): def add_per_unit_workdone(self, done: int): self.per_unit_req_done.append(done) + def add_metrics_snapshot(self, snapshot: dict): + self.metrics_snapshots.append(snapshot) + + def recent_ttft_itl(self, percentile: int, window: int) -> Tuple[float, float]: + """Return (ttft_secs, itl_secs) at `percentile` over the last `window` completed requests. + + Returns (-1.0, -1.0) until at least `window` requests have finished. + """ + if window <= 0: + raise ValueError(f"window must be a positive integer, got {window}") + if len(self.raw_ttft_values) < window: + return -1.0, -1.0 + recent_ttft = self.raw_ttft_values[-window:] + ttft = float(np.percentile(recent_ttft, percentile)) + recent_decode = self.raw_decode_values[-window:] + _, all_itls = self._calculate_itl_tpot(recent_decode) + itl = float(np.percentile(all_itls, percentile)) if len(all_itls) > 0 else -1.0 + return ttft, itl + def sample_workdone_per_K(self, k: int = 1): return sample_series_K(self.per_unit_req_done, k) diff --git a/opal/webserver/index.html b/opal/webserver/index.html index 4d7ca12..bea5c12 100644 --- a/opal/webserver/index.html +++ b/opal/webserver/index.html @@ -325,9 +325,19 @@

OPAL config builder

- - -
Virtual seconds between router-side infra-event drains (KV events from workers).
+ + +
Percentile used for snapshot TTFT/ITL SLOs. Computed over latency_window recent completions.
+
+ +
+ + +
Number of most recent completed requests used for TTFT/ITL percentiles. Snapshots store -1 until this many requests have finished.
@@ -366,8 +376,8 @@

worker_params

- -
Virtual seconds between worker→router status pushes (gpu_util, queue depth, mem_used).
+ +
Virtual seconds between worker→router status pushes (queue depth, kvc_util). Snapshots are taken when these arrive.
diff --git a/opal/worker/vllm_worker.py b/opal/worker/vllm_worker.py index 5efb453..d635c6f 100644 --- a/opal/worker/vllm_worker.py +++ b/opal/worker/vllm_worker.py @@ -7,23 +7,17 @@ """ TODO(atr) - Known issues to address: -1. _periodic_infra_updates is defined but never started in _run(). - The router never receives load/utilization updates from this worker. - -2. Dead code in _async_kvc_retrieve: the assert on line ~1508 guarantees +1. Dead code in _async_kvc_retrieve: the assert on line ~1508 guarantees actual_kvc_blocks == estimated_kvc_blocks, making the subsequent if-block unreachable. -3. Starvation risk: preempted requests reset to prompt_processed=0 and get +2. Starvation risk: preempted requests reset to prompt_processed=0 and get inserted at the front of waiting_requests, but under sustained memory pressure they can be repeatedly preempted (up to cap of 3) with no further recourse or priority escalation. -4. Leaked request_tokens entries: Phase 3 preemption removes requests from +3. Leaked request_tokens entries: Phase 3 preemption removes requests from batch lists but never does `del batch.request_tokens[req.request_id]`. - -5. Unused variable: total_requests in _periodic_infra_updates (line ~608) - is computed but never used. """ """ @@ -264,6 +258,7 @@ def __init__(self, opal_env, opal_config, output_req_queue, infra_update_queue): self.log = logging.getLogger(str(self)) # Configuration + # Worker -> router SystemEvent push cadence (bounds how stale pooled metrics can be). self.periodic_infra_update_time = self.opalConfig["worker"]["worker_params"]["periodic_infra_update_time"] self.kvcevent_coalesce_time = self.opalConfig["worker"]["worker_params"]["kvcevent_coalesce_time"] @@ -606,6 +601,9 @@ def _run(self): # Then start request checker which may interrupt the scheduler self._check_new_request_process = self.simpy_env.process(self._check_new_requests()) self.simpy_env.process(self._periodic_kvc_updates()) + # Push periodic system telemetry (queue depth, KV util, GPU util) to the router. + if self.periodic_infra_update_time > 0: + self.simpy_env.process(self._periodic_infra_updates()) def __str__(self): return f"{__class__.__name__}.{self.id}" @@ -624,20 +622,23 @@ def _periodic_kvc_updates(self): self._pending_kvc_events = [] def _periodic_infra_updates(self): - """Periodically send infrastructure updates to router.""" + """Periodically send infrastructure updates to router every periodic_infra_update_time seconds.""" router = self.opalEnv.registry.get_router() while not self.opalEnv.are_we_done(): yield self.simpy_env.timeout(self.periodic_infra_update_time) - total_requests = len(self.waiting_requests) + len(self.running_requests) + queue_depth = len(self.waiting_requests) + len(self.running_requests) queue_occupancy = len(self.waiting_requests) / max(1, self.scheduler_config.max_num_seqs) gpu_utilization = min(1.0, len(self.running_requests) / max(1, self.scheduler_config.max_num_seqs)) + # Fraction of GPU KV-cache blocks in use, in [0, 1]. + kvc_utilization = 1.0 - (self.free_gpu_blocks / max(1, self.total_gpu_blocks)) sys_event = SystemEvent( worker_id=self.id, load=gpu_utilization, ingress_queue_occupancy=queue_occupancy, - mem_used=0.66, gpu_utilization=gpu_utilization, + kvc_utilization=kvc_utilization, + queue_depth=queue_depth, ) self.simpy_env.process(router.queue_events([sys_event])) diff --git a/tests/test_configs.py b/tests/test_configs.py index e498bc2..6b6c062 100644 --- a/tests/test_configs.py +++ b/tests/test_configs.py @@ -2,8 +2,8 @@ import pytest from pathlib import Path -from opal.opal import OpalSimulator -from opal.opal_config import OpalConfig +from opal.core.opal import OpalSimulator +from opal.config.opal_config import OpalConfig CONFIGS_DIR = Path(__file__).resolve().parent.parent / "configs" diff --git a/wiki/Configuration-Simulation.md b/wiki/Configuration-Simulation.md index badf084..014b803 100644 --- a/wiki/Configuration-Simulation.md +++ b/wiki/Configuration-Simulation.md @@ -39,8 +39,9 @@ Nested under `router.router_params`: | `max_queue_threshold` | int | `4` | When any worker's queue reaches this size, trigger a scale-up event. | | `scale_latency` | float | `40` | Virtual seconds it takes to start a new worker after a scale-up is triggered. | | `max_workers` | int | `50` | Maximum number of workers to scale up to. | -| `periodic_infra_update_collection_time` | float | `30` | Interval (virtual seconds) at which the router collects infrastructure status from workers. | -| `max_event_batch_size` | int | `64` | Maximum number of requests the router dispatches per scheduling cycle. | +| `latency_percentile` | string | `"p95"` | Percentile used for snapshot TTFT/ITL SLOs. Supported: `p90`, `p95`, `p99`. | +| `latency_window` | int | `50` | Number of most recent completed requests over which TTFT/ITL percentiles are computed. Snapshots store `-1` until this many requests have finished. | +| `max_event_batch_size` | int | `64` | Maximum number of infrastructure events processed per drain in `process_events`. | --- @@ -88,7 +89,7 @@ Workloads are defined as a list of stages under `workload.stages`. Each stage ha | Parameter | Type | Default | Description | |-----------|------|---------|-------------| | `worker_local_queue_capacity` | int | `1` | Each worker's local queue capacity where the router places incoming requests. | -| `periodic_infra_update_time` | float | `30` | Interval (virtual seconds) at which the worker reports its status to the router. | +| `periodic_infra_update_time` | float | `5` | Interval (virtual seconds) at which the worker pushes status (`SystemEvent`) to the router. This is the sole bound on how stale worker metrics in a `MetricsSnapshot` can be. | | `kvcevent_coalesce_time` | float | `30` | Time window for coalescing KV cache events before processing. | ### worker.hw diff --git a/wiki/Router.md b/wiki/Router.md index ebd9abe..9f89aa7 100644 --- a/wiki/Router.md +++ b/wiki/Router.md @@ -24,8 +24,8 @@ Workload Generator The router runs several concurrent SimPy processes: - **`_accept_requests`** — pulls requests from the input queue, applies the routing policy, and dispatches to a worker. - **`_collect_completion`** — gathers finished requests from the shared results queue and records statistics. -- **`_per_second_stats`** — samples per-second throughput and GPU utilization. -- **`process_events`** — batches and processes infrastructure events (KVC updates, system events) used by prefix-aware policies. +- **`_per_second_stats`** — samples per-second throughput, worker count, and GPU utilization for plots. +- **`process_events`** — drains infrastructure events as they arrive (KVC updates, worker `SystemEvent` telemetry). KVC events update prefix-aware routing; `SystemEvent`s produce a `MetricsSnapshot`. - **`_per_second_scaling`** (optional) — auto-scales workers when queue depth exceeds a threshold. ## Routing Policies @@ -58,7 +58,8 @@ When `enable_scaling` is `true`, the router checks worker queue depths every sim "max_queue_threshold": 4, "scale_latency": 40, "max_workers": 50, - "periodic_infra_update_collection_time": 30, + "latency_percentile": "p95", + "latency_window": 50, "max_event_batch_size": 64 } } @@ -71,9 +72,12 @@ When `enable_scaling` is `true`, the router checks worker queue depths every sim | `max_queue_threshold` | Queue depth that triggers a scale-up. | | `scale_latency` | Simulated seconds to provision a new worker. | | `max_workers` | Maximum number of workers allowed. | -| `periodic_infra_update_collection_time` | Interval (sim seconds) between event processing cycles when the event queue is empty. | +| `latency_percentile` | Percentile for snapshot TTFT/ITL SLOs (`p90`, `p95`, `p99`). | +| `latency_window` | Number of most recent completed requests over which those percentiles are computed. | | `max_event_batch_size` | Max events processed per batch in `process_events`. | ## Event Processing -Workers emit `KVCEvent` and `SystemEvent` messages to the router's event queue. These are batched (up to `max_event_batch_size`) and forwarded to the KVBM, which maintains the prefix-cache state used by the `MaxPrefix` policy. When the event queue is empty, the router sleeps for `periodic_infra_update_collection_time` before checking again. +Workers emit `KVCEvent` and `SystemEvent` messages to the router's event queue. `process_events` blocks until an event arrives, then drains up to `max_event_batch_size` events and forwards them to the KVBM. + +`SystemEvent` telemetry is the last report from each worker (pushed every `worker.periodic_infra_update_time` seconds). When a drain includes at least one `SystemEvent`, the router records a `MetricsSnapshot`: reported queue depth and KV-cache utilization, plus TTFT/ITL at `latency_percentile` over the last `latency_window` completions. Snapshots are not taken on a wall-clock tick; KVC-only batches do not create snapshots.