fix(realtime): avoid false keepalive timeouts under load

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 |

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

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 状态日志：
 

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
 
 

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,

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

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()