Merge pull request #3561 from modelscope/codex/realtime-l20-profile-20260830

perf(realtime): add opt-in decode profiling

Author: LauraGPT

docs/benchmark/realtime_ws_benchmark.md

@@ -19,7 +19,8 @@ CUDA_VISIBLE_DEVICES=0 python examples/industrial_data_pretraining/fun_asr_nano/
     --port 10095 --language 中文 \
     --partial-window-sec 8 --decode-interval 0.8 \
     --vad-device cpu --vad-ncpu 1 \
-    --decode-batch-wait-ms 10 --decode-max-batch-size 16
+    --decode-batch-wait-ms 10 --decode-max-batch-size 16 \
+    --log-decode-profile
 ```
 
 Speaker diarization is disabled by default. Add `--enable-spk` only when the
@@ -81,9 +82,16 @@ stress signal, not as user-facing realtime latency.
 | `errors` | Connection, timeout, protocol, or client-side validation errors |
 
 The script can observe only client-side timing and fields returned by the
-server. If you are debugging service internals, collect server logs separately
-for queue wait, VAD time, ASR decode time, speaker diarization time, GPU memory,
-and GPU utilization.
+server. For a performance investigation, add `--log-decode-profile` to record
+one structured line per engine call with the request and sample counts, audio
+duration range, queue-wait p50/max, and total engine latency. The underlying
+Fun-ASR-Nano vLLM path also logs audio-encoder and vLLM-generation time. Collect
+those server logs together with GPU memory/utilization and the client JSONL.
+
+When comparing releases, align `partial_messages` as well as audio, clients,
+and service flags. A server that blocks its WebSocket event loop can appear to
+finish sooner simply because it processes fewer provisional decodes; that is
+not an engine-throughput improvement and gives users fewer live updates.
 
 ## Concurrency Regression Reference
 
@@ -117,7 +125,7 @@ When publishing a realtime WebSocket benchmark or issue report, include:
 |----------|----------------|
 | Data | Audio duration, sample rate, language/domain, silence ratio or speaking pattern, and whether the same file was looped |
 | Load | `--clients`, `--loops`, `--chunk-ms`, paced or `--no-pace`, client ping interval/timeout, and total benchmark wall time |
-| Service | `serve_realtime_ws.py` command, WebSocket ping interval/timeout, `--partial-window-sec`, `--decode-interval`, `--vad-device`, `--vad-ncpu`, `--decode-batch-wait-ms`, `--decode-max-batch-size`, `--enable-spk`, language, and hotwords |
+| Service | `serve_realtime_ws.py` command, WebSocket ping interval/timeout, `--partial-window-sec`, `--decode-interval`, `--vad-device`, `--vad-ncpu`, `--decode-batch-wait-ms`, `--decode-max-batch-size`, `--log-decode-profile`, `--enable-spk`, language, and hotwords |
 | Hardware | GPU/NPU model, GPU count, memory, driver, CUDA/CANN/runtime versions, CPU model, and available RAM |
 | Software | `funasr`, PyTorch, torchaudio, vLLM, Python, OS, and container image if any |
 | Output | Summary line, JSONL artifact, server logs, and any failed client IDs |

funasr/bin/realtime_ws.py

@@ -41,6 +41,7 @@ class _BatchRequest:
     def __init__(self, inputs, kwargs):
         self.inputs = inputs
         self.kwargs = kwargs
+        self.enqueued_at = time.monotonic()
         self.event = threading.Event()
         self.result = None
         self.error = None
@@ -64,11 +65,14 @@ def __getattr__(self, name):
 class RealtimeBatchingEngine:
     """Serialize the shared engine while batching compatible session requests."""
 
-    def __init__(self, engine, batch_wait_ms=10.0, max_batch_size=16):
+    def __init__(
+        self, engine, batch_wait_ms=10.0, max_batch_size=16, log_profile=False
+    ):
         self.engine = engine
         self._engine = getattr(engine, "_engine", engine)
         self.batch_wait_s = max(0.0, float(batch_wait_ms)) / 1000.0
         self.max_batch_size = max(1, int(max_batch_size))
+        self.log_profile = bool(log_profile)
         self.requests = queue.Queue()
         self.pending_request = None
         self.worker = threading.Thread(
@@ -186,6 +190,7 @@ def _run(self):
 
     def _generate_group(self, requests):
         inputs = [item for request in requests for item in request.inputs]
+        started_at = time.monotonic()
         try:
             results = self.engine.generate(inputs, **requests[0].kwargs)
             if len(results) != len(inputs):
@@ -211,6 +216,36 @@ def _generate_group(self, requests):
                 request.event.set()
             return
 
+        if self.log_profile:
+            queue_waits_ms = sorted(
+                (started_at - request.enqueued_at) * 1000 for request in requests
+            )
+            midpoint = len(queue_waits_ms) // 2
+            if len(queue_waits_ms) % 2:
+                queue_p50_ms = queue_waits_ms[midpoint]
+            else:
+                queue_p50_ms = (
+                    queue_waits_ms[midpoint - 1] + queue_waits_ms[midpoint]
+                ) / 2
+            audio_seconds = [
+                item.shape[-1] / 16000.0
+                for item in inputs
+                if isinstance(item, (np.ndarray, torch.Tensor)) and item.ndim > 0
+            ]
+            logger.info(
+                "Realtime decode profile: requests=%d samples=%d "
+                "audio_sec_total=%.3f audio_sec_min=%.3f audio_sec_max=%.3f "
+                "queue_ms_p50=%.3f queue_ms_max=%.3f engine_ms=%.3f",
+                len(requests),
+                len(inputs),
+                sum(audio_seconds),
+                min(audio_seconds, default=0.0),
+                max(audio_seconds, default=0.0),
+                queue_p50_ms,
+                queue_waits_ms[-1],
+                (time.monotonic() - started_at) * 1000,
+            )
+
         offset = 0
         for request in requests:
             end = offset + len(request.inputs)
@@ -1158,6 +1193,7 @@ def load_models(args):
             engine,
             batch_wait_ms=getattr(args, "decode_batch_wait_ms", 10.0),
             max_batch_size=getattr(args, "decode_max_batch_size", 16),
+            log_profile=getattr(args, "log_decode_profile", False),
         )
 
         _asr_kwargs = {}
@@ -1451,6 +1487,14 @@ def build_arg_parser():
         default=16,
         help="Maximum number of audio segments submitted in one batched decode.",
     )
+    parser.add_argument(
+        "--log-decode-profile",
+        action="store_true",
+        help=(
+            "Log per-engine-batch request counts, audio durations, queue wait, "
+            "and engine latency for performance investigations."
+        ),
+    )
     parser.add_argument(
         "--endpoint-mode",
         choices=["server", "client"],

tests/test_realtime_ws_service.py

@@ -116,6 +116,7 @@ def test_cli_defaults_disable_speaker_and_bound_partial_window():
     assert args.log_session_stats_interval == 0.0
     assert args.decode_batch_wait_ms == 10.0
     assert args.decode_max_batch_size == 16
+    assert args.log_decode_profile is False
     assert args.vad_device == "cpu"
     assert args.vad_ncpu == 1
 
@@ -1690,6 +1691,37 @@ def generate(index, values):
     assert engine.calls[0][1] == {"language": "zh"}
 
 
+def test_realtime_batching_engine_logs_opt_in_decode_profile(caplog):
+    module = load_service_module()
+
+    class RecordingEngine:
+        def generate(self, inputs, **kwargs):
+            return [{"text": "ok"} for _ in inputs]
+
+    caplog.set_level("INFO")
+    batching_engine = module.RealtimeBatchingEngine(
+        RecordingEngine(), batch_wait_ms=0, max_batch_size=8, log_profile=True
+    )
+
+    result = batching_engine.generate(
+        [
+            np.zeros(16000, dtype=np.float32),
+            np.zeros(32000, dtype=np.float32),
+        ]
+    )
+
+    assert result == [{"text": "ok"}, {"text": "ok"}]
+    assert "Realtime decode profile:" in caplog.text
+    assert "requests=1" in caplog.text
+    assert "samples=2" in caplog.text
+    assert "audio_sec_total=3.000" in caplog.text
+    assert "audio_sec_min=1.000" in caplog.text
+    assert "audio_sec_max=2.000" in caplog.text
+    assert "queue_ms_p50=" in caplog.text
+    assert "queue_ms_max=" in caplog.text
+    assert "engine_ms=" in caplog.text
+
+
 def test_realtime_batching_engine_keeps_incompatible_options_separate():
     module = load_service_module()