diff --git a/openpilot/selfdrive/c3_client.py b/openpilot/selfdrive/c3_client.py index 0d7cce48d..006a00fea 100755 --- a/openpilot/selfdrive/c3_client.py +++ b/openpilot/selfdrive/c3_client.py @@ -98,6 +98,14 @@ class _WebSocketClient: except asyncio.TimeoutError: raise _WebSocketError("连接或握手超时") + def _notify_closed(self): + """标记连接关闭并唤醒阻塞在 recv() 上的接收协程""" + self._closed = True + try: + self._recv_queue.put_nowait(("closed", "")) + except Exception: + pass + async def _recv_loop(self): """持续读取 WebSocket 帧""" try: @@ -124,7 +132,7 @@ class _WebSocketClient: except Exception as e: print(f"[C3] WebSocket 接收异常: {e}") finally: - self._closed = True + self._notify_closed() async def _read_frame(self): """读取一个 WebSocket 帧""" @@ -163,6 +171,8 @@ class _WebSocketClient: async def send(self, text): """发送文本消息""" + if self._closed: + raise ConnectionError("连接已关闭") payload = text.encode("utf-8") if isinstance(text, str) else text await self._send_frame(0x1, payload) @@ -199,6 +209,8 @@ class _WebSocketClient: try: while not self._closed: await asyncio.sleep(self.ping_interval) + if self._closed: + break self._pong_event.clear() await self._send_frame(0x9, b"") try: @@ -211,12 +223,16 @@ class _WebSocketClient: except Exception as e: print(f"[C3] Ping 循环异常: {e}") finally: - self._closed = True + self._notify_closed() async def recv(self): """接收一条消息""" - while not self._closed: + if self._closed: + raise ConnectionError("连接已关闭") + while True: msg_type, text = await self._recv_queue.get() + if msg_type == "closed": + break if msg_type == "text": return text raise ConnectionError("连接已关闭") @@ -232,7 +248,7 @@ class _WebSocketClient: break async def close(self): - self._closed = True + self._notify_closed() if self._writer: await self._send_close() try: @@ -765,8 +781,14 @@ async def run(): # 原为 wait_for(..., timeout=65):每 65 秒强制取消 gather,导致设备「自杀式」周期断线重连(~66s 一次) # 保活已由内部 heartbeat(5s) + WebSocket ping(20s/timeout15s) 兜底,无需外层硬超时。 - # 真正断连由 receiver() 的 async for 抛 ConnectionError 自动退出重连,逻辑不变。 - await asyncio.gather(heartbeat(), receiver()) + # 任一协程结束都会取消另一协程,避免 receiver 卡住时 gather 永远不返回。 + tasks = [asyncio.create_task(heartbeat()), asyncio.create_task(receiver())] + try: + await asyncio.gather(*tasks) + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) except (ConnectionError, OSError, _WebSocketError): print(f"[C3] 连接断开")