diff --git a/frigate/comms/ws.py b/frigate/comms/ws.py index 348640e2d6..6c62c3baad 100644 --- a/frigate/comms/ws.py +++ b/frigate/comms/ws.py @@ -3,6 +3,9 @@ import errno import json import logging +import queue +import socket +import struct import threading from collections.abc import Callable from typing import Any @@ -76,6 +79,12 @@ _WS_VIEWER_TOPICS = frozenset( # Camera-scoped command topics a camera-authorized (non-admin) user may send. _WS_CAMERA_COMMAND_TOPICS = frozenset({"ptz"}) +# Max outbound messages waiting on a client's writer thread. +WS_MAX_PENDING_MESSAGES = 256 + +# Seconds a single send() may wait for a client to free socket buffer space. +WS_SEND_TIMEOUT = 10 + def _check_ws_authorization( topic: str, @@ -464,6 +473,70 @@ def _materialize_for_ws( class WebSocket(WebSocket_): # type: ignore[misc] + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self._send_queue: queue.Queue[tuple[Any, bool] | None] = queue.Queue( + maxsize=WS_MAX_PENDING_MESSAGES + ) + self._writer: threading.Thread | None = None + self._aborted = False + + def opened(self) -> None: + # ws4py writes close and pong frames on the shared manager thread, so a + # stalled client must not block sendall forever. SO_SNDTIMEO leaves recv + # (and idle clients) untouched, unlike settimeout(). + self.sock.setsockopt( + socket.SOL_SOCKET, socket.SO_SNDTIMEO, struct.pack("ll", WS_SEND_TIMEOUT, 0) + ) + + # every client gets its own writer so a client that stops reading only + # blocks itself, never the thread that called publish() + self._writer = threading.Thread( + target=self._drain_send_queue, name="ws_writer", daemon=True + ) + self._writer.start() + + def send(self, payload: Any, binary: bool = False) -> None: + try: + self._send_queue.put_nowait((payload, binary)) + except queue.Full: + self._abort("Websocket client is not keeping up, disconnecting it") + + def _drain_send_queue(self) -> None: + while True: + item = self._send_queue.get() + if item is None or self.terminated or self.sock is None: + return + try: + super().send(*item) + except Exception: + self._abort() + return + + def _abort(self, reason: str | None = None) -> None: + # publish() keeps hitting a full queue until the manager thread removes + # the connection, so only act (and log) the first time + if self._aborted: + return + self._aborted = True + if reason: + logger.warning(reason) + + # shutdown rather than close so the ws4py manager thread sees EOF and + # runs its normal unregister/terminate; this also unblocks a stuck sendall + sock = self.sock + if sock is not None: + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + def closed(self, code: int, reason: str | None = None) -> None: + try: + self._send_queue.put_nowait(None) + except queue.Full: + pass + def unhandled_error(self, error: Any) -> None: """ Handles the unfriendly socket closures on the server side @@ -597,10 +670,7 @@ class WebSocketClient(Communicator): ) if message is None: continue - try: - ws.send(message) - except (ConnectionResetError, BrokenPipeError, ValueError): - pass + ws.send(message) def stop(self) -> None: if self.websocket_server is not None: diff --git a/frigate/output/birdseye.py b/frigate/output/birdseye.py index 4dfbb74172..44799db0f5 100644 --- a/frigate/output/birdseye.py +++ b/frigate/output/birdseye.py @@ -277,12 +277,7 @@ class BroadcastThread(threading.Thread): and ws.environ["PATH_INFO"] == f"/{self.camera}" and ws_has_camera_access(ws, self.camera, self.config) ): - try: - ws.send(buf, binary=True) - except ValueError: - pass - except (BrokenPipeError, ConnectionResetError, OSError) as e: - logger.debug(f"Websocket unexpectedly closed {e}") + ws.send(buf, binary=True) elif self.converter.process.poll() is not None: break diff --git a/frigate/output/camera.py b/frigate/output/camera.py index 88d16ed4b4..96ff2abbad 100644 --- a/frigate/output/camera.py +++ b/frigate/output/camera.py @@ -127,12 +127,7 @@ class BroadcastThread(threading.Thread): and ws.environ["PATH_INFO"] == f"/{self.camera}" and ws_has_camera_access(ws, self.camera, self.config) ): - try: - ws.send(buf, binary=True) - except ValueError: - pass - except (BrokenPipeError, ConnectionResetError) as e: - logger.debug(f"Websocket unexpectedly closed {e}") + ws.send(buf, binary=True) elif self.converter.process.poll() is not None: break diff --git a/frigate/test/test_ws_send_queue.py b/frigate/test/test_ws_send_queue.py new file mode 100644 index 0000000000..c758488c72 --- /dev/null +++ b/frigate/test/test_ws_send_queue.py @@ -0,0 +1,131 @@ +"""Outbound websocket sends must never block the thread calling publish().""" + +import socket +import threading +import unittest + +from frigate.comms.ws import WS_MAX_PENDING_MESSAGES, WebSocket + + +class _FakeSock: + """Socket stand-in; ``block`` makes sendall hang like a client that stopped reading.""" + + def __init__(self, block: bool = False) -> None: + self.block = block + self.released = threading.Event() + self.shutdown_called = threading.Event() + self.frames: list[bytes] = [] + + def sendall(self, data: bytes) -> None: + if self.shutdown_called.is_set(): + raise BrokenPipeError() + if self.block and not self.released.is_set(): + self.released.wait(timeout=10) + raise BrokenPipeError() + self.frames.append(data) + + def shutdown(self, how: int) -> None: + assert how == socket.SHUT_RDWR + self.shutdown_called.set() + self.released.set() + + def close(self) -> None: + pass + + def setsockopt(self, *args) -> None: + pass + + def fileno(self) -> int: + return 99 + + +def _wait_for(predicate, timeout: float = 2.0) -> bool: + deadline = threading.Event() + for _ in range(int(timeout / 0.01)): + if predicate(): + return True + deadline.wait(0.01) + return predicate() + + +class TestWebSocketSendQueue(unittest.TestCase): + def _open(self, sock: _FakeSock) -> WebSocket: + ws = WebSocket(sock) + ws.opened() + return ws + + def test_stalled_client_does_not_block_publisher(self): + sock = _FakeSock(block=True) + ws = self._open(sock) + + def publish_many(): + for i in range(WS_MAX_PENDING_MESSAGES + 5): + ws.send(f"message {i}") + + publisher = threading.Thread(target=publish_many, daemon=True) + publisher.start() + publisher.join(timeout=2.0) + + self.assertFalse(publisher.is_alive(), "publish() blocked on a stalled client") + self.assertTrue( + sock.shutdown_called.wait(timeout=2.0), + "a client that cannot keep up should be disconnected", + ) + ws._writer.join(2) + self.assertFalse(ws._writer.is_alive()) + + def test_overflow_warns_and_shuts_down_once(self): + sock = _FakeSock(block=True) + ws = self._open(sock) + shutdown_calls = [] + original_shutdown = sock.shutdown + sock.shutdown = lambda how: (shutdown_calls.append(how), original_shutdown(how)) + + with self.assertLogs("frigate.comms.ws", level="WARNING") as logs: + # keep publishing after overflow, as the dispatcher does until the + # manager thread removes the connection + for i in range(WS_MAX_PENDING_MESSAGES * 3): + ws.send(f"message {i}") + + self.assertEqual(len(logs.output), 1) + self.assertEqual(len(shutdown_calls), 1) + + def test_messages_delivered_in_order(self): + sock = _FakeSock() + ws = self._open(sock) + for i in range(3): + ws.send(f"message {i}") + + self.assertTrue(_wait_for(lambda: len(sock.frames) == 3)) + for i, frame in enumerate(sock.frames): + self.assertIn(f"message {i}".encode(), frame) + self.assertFalse(sock.shutdown_called.is_set()) + + def test_closed_stops_writer_thread(self): + sock = _FakeSock() + ws = self._open(sock) + writer = ws._writer + ws.closed(1000, "bye") + writer.join(timeout=2.0) + self.assertFalse(writer.is_alive()) + + def test_closed_with_full_queue_and_blocked_writer_exits_after_shutdown(self): + sock = _FakeSock(block=True) + ws = self._open(sock) + ws.send("in flight") + self.assertTrue(_wait_for(ws._send_queue.empty)) + for i in range(WS_MAX_PENDING_MESSAGES): + ws.send(f"message {i}") + self.assertTrue(ws._send_queue.full()) + + ws.closed(1006) + self.assertTrue(ws._writer.is_alive()) + self.assertFalse(sock.shutdown_called.is_set()) + + sock.shutdown(socket.SHUT_RDWR) + ws._writer.join(2) + self.assertFalse(ws._writer.is_alive()) + + +if __name__ == "__main__": + unittest.main()