diff --git a/docs/benchmark/realtime_ws_benchmark.md b/docs/benchmark/realtime_ws_benchmark.md index 98b3e1cbd..12afccd0a 100644 --- a/docs/benchmark/realtime_ws_benchmark.md +++ b/docs/benchmark/realtime_ws_benchmark.md @@ -50,10 +50,17 @@ python examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.p --clients 8 \ --loops 3 \ --chunk-ms 100 \ + --client-ping-interval 20 \ + --client-ping-timeout 0 \ --language 中文 \ --output-jsonl realtime_ws_8c.jsonl ``` +Values `<=0` disable the corresponding client ping setting. Record both client +settings when comparing disconnects. The `websockets` library's `max_queue` +setting bounds receive buffering for incoming messages; it doesn't change +ping/pong timeout semantics. + Use a representative audio file. A long, pauseless monologue creates a very different load shape from turn-taking meetings, because nearly every client is speaking and triggering partial decodes at the same time. @@ -109,8 +116,8 @@ When publishing a realtime WebSocket benchmark or issue report, include: | Category | What to record | |----------|----------------| | 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`, and total benchmark wall time | -| Service | `serve_realtime_ws.py` command, `--partial-window-sec`, `--decode-interval`, `--vad-device`, `--vad-ncpu`, `--decode-batch-wait-ms`, `--decode-max-batch-size`, `--enable-spk`, language, and hotwords | +| 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 | | 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 | diff --git a/docs/vllm_guide.md b/docs/vllm_guide.md index c24933761..759e37a45 100644 --- a/docs/vllm_guide.md +++ b/docs/vllm_guide.md @@ -638,17 +638,22 @@ CUDA_VISIBLE_DEVICES=0 python examples/industrial_data_pretraining/fun_asr_nano/ Speaker diarization is disabled by default; add `--enable-spk` only when the `spk` field is required. -For long-lived microphone sessions behind Docker, nginx, or a cloud load -balancer, keep WebSocket ping/pong enabled and tune the timeout to be longer -than short network stalls: +The server sends WebSocket pings every 20 seconds by default, while ping-timeout +closure is disabled. Under concurrent long-audio load, model and VAD work can +delay control-frame handling even though the connection is healthy. A fixed +timeout can therefore close valid sessions during compute or queue backpressure. ```bash CUDA_VISIBLE_DEVICES=0 python examples/industrial_data_pretraining/fun_asr_nano/serve_realtime_ws.py \ - --port 10095 --language 中文 \ - --ws-ping-interval 20 --ws-ping-timeout 60 + --port 10095 --language 中文 ``` -Set `--ws-ping-interval 0` only when an external gateway already owns +Set a positive `--ws-ping-timeout` only after measuring the worst-case decode +and queue delay for the production traffic shape; keep it above that delay and +coordinate it with the gateway idle-timeout policy. The `websockets` library's +`max_queue` setting bounds receive buffering for incoming messages; it doesn't +change ping/pong timeout semantics, so increasing it doesn't fix keepalive +timeouts. Set `--ws-ping-interval 0` only when an external gateway already owns keepalive/reconnect policy. For long-session debugging, especially with `--enable-spk`, enable periodic diff --git a/docs/vllm_guide_zh.md b/docs/vllm_guide_zh.md index 1f7bb1ba3..192870565 100644 --- a/docs/vllm_guide_zh.md +++ b/docs/vllm_guide_zh.md @@ -638,17 +638,21 @@ CUDA_VISIBLE_DEVICES=0 python examples/industrial_data_pretraining/fun_asr_nano/ 说话人分离默认关闭；只有确实需要 `spk` 字段时再加 `--enable-spk`。 -如果麦克风长连接经过 Docker、nginx 或云负载均衡，建议保持 WebSocket -ping/pong 开启，并把 timeout 调到能覆盖短暂网络抖动： +服务端默认每 20 秒发送一次 WebSocket ping，但不因 ping 超时主动断开连接。 +在长音频高并发下，模型推理和 VAD 处理会延迟控制帧，即使连接仍然健康； +固定 timeout 因此可能在计算或排队反压期间误杀正常会话。 ```bash CUDA_VISIBLE_DEVICES=0 python examples/industrial_data_pretraining/fun_asr_nano/serve_realtime_ws.py \ - --port 10095 --language 中文 \ - --ws-ping-interval 20 --ws-ping-timeout 60 + --port 10095 --language 中文 ``` -只有在外部网关已经统一负责 keepalive / reconnect 策略时，才考虑设置 -`--ws-ping-interval 0` 关闭服务端 ping。 +只有在按生产流量测得最坏推理和排队延迟后，才设置正数 +`--ws-ping-timeout`；该值应高于实测延迟，并与网关 idle timeout 策略配合。 +`websockets` 库的 `max_queue` 设置只限制入站消息的接收缓冲，不会改变 +ping/pong 的超时语义，因此增大它不能解决 keepalive timeout。只有外部网关 +已经统一负责 keepalive / reconnect 策略时，才设置 `--ws-ping-interval 0` +关闭服务端 ping。 长会话排障，尤其是启用 `--enable-spk` 时，可以打开周期性 session 状态日志： diff --git a/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py b/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py index 0963d8a12..0bd2c37e1 100644 --- a/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py +++ b/examples/industrial_data_pretraining/fun_asr_nano/realtime_ws_benchmark.py @@ -127,6 +127,8 @@ async def run_client(client_id, args, audio_bytes, audio_seconds): "client_id": client_id, "audio_seconds": total_audio_seconds, "chunk_ms": args.chunk_ms, + "client_ping_interval": args.client_ping_interval, + "client_ping_timeout": args.client_ping_timeout, "messages": 0, "result_messages": 0, "partial_messages": 0, @@ -146,7 +148,8 @@ async def run_client(client_id, args, audio_bytes, audio_seconds): try: async with websockets.connect( args.server, - ping_interval=None, + ping_interval=args.client_ping_interval, + ping_timeout=args.client_ping_timeout, open_timeout=args.connect_timeout, max_size=args.max_message_size, ) as ws: @@ -270,7 +273,7 @@ async def async_main(args): return 1 if summary["errors"] else 0 -def parse_args(): +def parse_args(argv=None): parser = argparse.ArgumentParser(description="Benchmark serve_realtime_ws.py with one or more clients") parser.add_argument("wav", help="16 kHz mono PCM16 WAV file") parser.add_argument("--server", default="ws://localhost:10095", help="WebSocket server URL") @@ -283,15 +286,33 @@ def parse_args(): parser.add_argument("--output-jsonl", default="", help="Write per-client metrics and summary JSONL") parser.add_argument("--connect-timeout", type=float, default=10.0, help="Connection timeout seconds") parser.add_argument("--recv-timeout", type=float, default=30.0, help="Timeout waiting for server messages") + parser.add_argument( + "--client-ping-interval", + type=float, + default=0.0, + help="Client WebSocket ping interval in seconds; <=0 disables client pings", + ) + parser.add_argument( + "--client-ping-timeout", + type=float, + default=20.0, + help="Client WebSocket ping timeout in seconds; <=0 disables ping timeout", + ) parser.add_argument("--max-message-size", type=int, default=16 * 1024 * 1024, help="WebSocket max message size") parser.set_defaults(pace=True) - args = parser.parse_args() + args = parser.parse_args(argv) if args.clients < 1: parser.error("--clients must be >= 1") if args.loops < 1: parser.error("--loops must be >= 1") if args.chunk_ms < 10: parser.error("--chunk-ms must be >= 10") + args.client_ping_interval = ( + None if args.client_ping_interval <= 0 else args.client_ping_interval + ) + args.client_ping_timeout = ( + None if args.client_ping_timeout <= 0 else args.client_ping_timeout + ) return args diff --git a/funasr/bin/realtime_ws.py b/funasr/bin/realtime_ws.py index 7a9926092..9a480dce4 100644 --- a/funasr/bin/realtime_ws.py +++ b/funasr/bin/realtime_ws.py @@ -1492,8 +1492,16 @@ def build_arg_parser(): parser.add_argument("--max-model-len", type=int, default=2048) parser.add_argument("--ws-ping-interval", type=float, default=20.0, help="WebSocket ping interval in seconds; <=0 disables keepalive pings.") - parser.add_argument("--ws-ping-timeout", type=float, default=20.0, - help="WebSocket ping timeout in seconds; <=0 disables ping timeout.") + parser.add_argument( + "--ws-ping-timeout", + type=float, + default=0.0, + help=( + "WebSocket ping timeout in seconds; disabled by default because " + "decode/queue backpressure can delay pong handling. Set a positive " + "value only above the measured worst-case delay." + ), + ) parser.add_argument("--ws-close-timeout", type=float, default=10.0, help="WebSocket close handshake timeout in seconds.") parser.add_argument("--ws-max-size", type=int, default=10 * 1024 * 1024, diff --git a/tests/test_realtime_ws_benchmark.py b/tests/test_realtime_ws_benchmark.py index db9f14ac2..f1224ea42 100644 --- a/tests/test_realtime_ws_benchmark.py +++ b/tests/test_realtime_ws_benchmark.py @@ -1,5 +1,6 @@ import asyncio import importlib.util +import json import sys from pathlib import Path @@ -62,3 +63,63 @@ async def fake_receive_message(ws, timeout): assert metrics["response_lag_ms"] == [200.0] assert metrics["final_update_ms"] == 2000.0 assert metrics["final_after_stop_ms"] == 500.0 + + +def test_client_ping_settings_are_forwarded_to_websocket_connect(monkeypatch): + module = load_benchmark_module() + args = module.parse_args( + [ + "audio.wav", + "--client-ping-interval", + "7", + "--client-ping-timeout", + "11", + "--no-pace", + ] + ) + connect_call = {} + + class FakeWebSocket: + def __init__(self): + self.messages = iter( + [ + {"event": "started"}, + {"is_final": True, "sentences": [{"text": "hello"}]}, + {"event": "stopped"}, + ] + ) + + async def send(self, _message): + return None + + async def recv(self): + return json.dumps(next(self.messages)) + + class FakeConnection: + async def __aenter__(self): + return FakeWebSocket() + + async def __aexit__(self, *_args): + return None + + def fake_connect(server, **kwargs): + connect_call.update({"server": server, **kwargs}) + return FakeConnection() + + monkeypatch.setattr(module.websockets, "connect", fake_connect) + + result = asyncio.run(module.run_client(0, args, b"\0\0" * 1600, 0.1)) + + assert result["errors"] == [] + assert result["client_ping_interval"] == 7.0 + assert result["client_ping_timeout"] == 11.0 + assert connect_call["ping_interval"] == 7.0 + assert connect_call["ping_timeout"] == 11.0 + + +def test_client_ping_timeout_zero_disables_timeout(): + module = load_benchmark_module() + + args = module.parse_args(["audio.wav", "--client-ping-timeout", "0"]) + + assert args.client_ping_timeout is None diff --git a/tests/test_realtime_ws_service.py b/tests/test_realtime_ws_service.py index 666447ab7..5b7153bff 100644 --- a/tests/test_realtime_ws_service.py +++ b/tests/test_realtime_ws_service.py @@ -171,6 +171,16 @@ def test_websocket_keepalive_kwargs_can_disable_ping(): assert kwargs["ping_timeout"] is None +def test_websocket_defaults_keep_pings_without_a_compute_timeout(): + module = load_service_module() + + args = module.build_arg_parser().parse_args([]) + + kwargs = module.build_websocket_serve_kwargs(args) + assert kwargs["ping_interval"] == 20.0 + assert kwargs["ping_timeout"] is None + + def test_cli_accepts_session_stats_interval(): module = load_service_module()