diff --git a/docs/contributing/profiling.md b/docs/contributing/profiling.md index 62fff29c5..4957b0447 100644 --- a/docs/contributing/profiling.md +++ b/docs/contributing/profiling.md @@ -317,6 +317,8 @@ Each 1-second window records: |---|---| | `windows.duration_s` | Wall time covered by the window | | `windows.loop_idle` / `windows.loop_active` | Orchestrator poll-loop iterations with no work vs. active forwarding | +| `windows.dispatch_queue_size` | Ready-queue size sampled at each monitor window boundary | +| `windows.dispatch_queue_high_water` | Maximum Event Driven ready-queue size observed in each window | | `replicas..outputs_queue_size` | MP client `outputs_queue` backlog for that replica | | `replicas..inflight` | Requests currently bound/routed to the replica | @@ -331,6 +333,10 @@ orchestrator wakes only once per 0.5 s reconcile timeout, while a busy one still records one iteration per routed output. Window counts and the `loop_active_pct` summary are therefore not comparable across the two modes. +For Event Driven runs, use `dispatch_queue_high_water` rather than only the +window-boundary `dispatch_queue_size`: a serial dispatcher can drain a burst +before the window closes, hiding a transient ready-queue backlog in the latter. + ### Relationship to other diagnostics This monitor is intentionally separate from the existing profiling tools: diff --git a/tests/engine/test_orchestrator_monitor.py b/tests/engine/test_orchestrator_monitor.py index ee9126cf4..1961f0ddf 100644 --- a/tests/engine/test_orchestrator_monitor.py +++ b/tests/engine/test_orchestrator_monitor.py @@ -45,6 +45,8 @@ def test_note_loop_and_flush_json(tmp_path, monkeypatch): monitor.register_replica(0, 0) monitor.note_loop(idle=False) + monitor.set_dispatch_queue_size(9) + monitor.set_dispatch_queue_size(7) monitor.note_loop(idle=True) now[0] = 1.5 @@ -54,6 +56,8 @@ def test_note_loop_and_flush_json(tmp_path, monkeypatch): assert payload["configured_window_s"] == monitor_mod._WINDOW_S assert payload["windows"]["loop_idle"] == [1] assert payload["windows"]["loop_active"] == [1] + assert payload["windows"]["dispatch_queue_size"] == [7] + assert payload["windows"]["dispatch_queue_high_water"] == [9] assert payload["replicas"]["stage=0,replica=0"]["outputs_queue_size"] == [5] assert payload["replicas"]["stage=0,replica=0"]["inflight"] == [10] diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index fadbd4698..3c7089f80 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -1278,6 +1278,7 @@ class Orchestrator: await asyncio.sleep(0.001) continue await ready_q.put(("llm", stage_id, replica_id, raw_outputs)) + self._orch_monitor.set_dispatch_queue_size(ready_q.qsize()) except asyncio.CancelledError: raise except BaseException as e: # noqa: BLE001 - routed to the dispatcher @@ -1296,6 +1297,7 @@ class Orchestrator: if output is None: continue await ready_q.put(("diffusion", stage_id, replica_id, output)) + self._orch_monitor.set_dispatch_queue_size(ready_q.qsize()) got = True await asyncio.sleep(0 if got else 0.001) except asyncio.CancelledError: diff --git a/vllm_omni/engine/orchestrator_monitor.py b/vllm_omni/engine/orchestrator_monitor.py index 8be903a88..132e55190 100644 --- a/vllm_omni/engine/orchestrator_monitor.py +++ b/vllm_omni/engine/orchestrator_monitor.py @@ -56,6 +56,8 @@ class OrchestratorMonitorBase(Protocol): def note_loop(self, *, idle: bool) -> None: ... + def set_dispatch_queue_size(self, size: int) -> None: ... + def flush(self) -> None: ... @@ -66,6 +68,9 @@ class _NullOrchestratorMonitor: def note_loop(self, *, idle: bool) -> None: return + def set_dispatch_queue_size(self, size: int) -> None: + return + def flush(self) -> None: return @@ -94,6 +99,10 @@ class OrchestratorMonitor: self._duration_s: list[float] = [] self._loop_idle_buf: list[int] = [] self._loop_active_buf: list[int] = [] + self._dispatch_queue_size = 0 + self._dispatch_queue_size_buf: list[int] = [] + self._dispatch_queue_high_water = 0 + self._dispatch_queue_high_water_buf: list[int] = [] self._replicas: dict[str, _ReplicaSeries] = {} def register_replica(self, stage_id: int, replica_id: int) -> None: @@ -107,6 +116,11 @@ class OrchestratorMonitor: else: self._loop_active += 1 + def set_dispatch_queue_size(self, size: int) -> None: + size = max(int(size), 0) + self._dispatch_queue_size = size + self._dispatch_queue_high_water = max(self._dispatch_queue_high_water, size) + def flush(self) -> None: if self._flushed: return @@ -118,6 +132,8 @@ class OrchestratorMonitor: "duration_s": self._duration_s, "loop_idle": self._loop_idle_buf, "loop_active": self._loop_active_buf, + "dispatch_queue_size": self._dispatch_queue_size_buf, + "dispatch_queue_high_water": self._dispatch_queue_high_water_buf, }, "replicas": dict(sorted(self._replicas.items())), } @@ -154,6 +170,8 @@ class OrchestratorMonitor: self._duration_s.append(max(now_mono - self._window_start_mono, 1e-9)) self._loop_idle_buf.append(self._loop_idle) self._loop_active_buf.append(self._loop_active) + self._dispatch_queue_size_buf.append(self._dispatch_queue_size) + self._dispatch_queue_high_water_buf.append(self._dispatch_queue_high_water) for key, series in self._replicas.items(): outputs_qsize, inflight = sampled.get(key, (0, 0)) series["outputs_queue_size"].append(outputs_qsize) @@ -161,6 +179,8 @@ class OrchestratorMonitor: self._loop_idle = 0 self._loop_active = 0 + self._dispatch_queue_size = 0 + self._dispatch_queue_high_water = 0 self._window_start_mono = now_mono def _log_summary(self, output_path: str) -> None: