[Core] Record Event Driven dispatch queue high water

Signed-off-by: Sy03 <1370724210@qq.com>

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.<stage,replica>.outputs_queue_size` | MP client `outputs_queue` backlog for that replica |
 | `replicas.<stage,replica>.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:

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]
 

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:

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: