diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index 0f41fa2..7353da3 100644 --- a/inorbit_edge/robot.py +++ b/inorbit_edge/robot.py @@ -1093,6 +1093,12 @@ def publish(image, width, height, ts): camera_id, image, int(width), int(height), int(ts) ) + # Tear down any streamer previously registered under this id so its + # worker thread is not leaked. + existing_streamer = self.camera_streamers.get(camera_id) + if existing_streamer is not None: + existing_streamer.shutdown() + self.camera_streamers[camera_id] = CameraStreamer(camera, publish) with self.camera_streaming_mutex: if self.camera_streaming_on: @@ -1202,6 +1208,23 @@ def disconnect(self): self.logger.info("Ending robot session") self._stop_cameras_streaming() + # Permanently terminate the camera worker threads. Signal all first, + # then join, so the total wait is the slowest single teardown rather + # than the sum. The join is capped because a worker can be mid-open on + # an unreachable stream; daemon threads guarantee process exit even if + # a worker overruns the cap. + camera_teardown_timeout = 12 + streamers = list(self.camera_streamers.values()) + for streamer in streamers: + streamer.shutdown() + for streamer in streamers: + streamer.join(timeout=camera_teardown_timeout) + if streamer.is_alive(): + self.logger.warning( + "Camera streamer worker did not exit within " + f"{camera_teardown_timeout}s" + ) + # Send offline status (best effort, non-blocking) # InOrbit will detect offline via data absence if this fails self._send_robot_status(online=False) diff --git a/inorbit_edge/tests/test_video.py b/inorbit_edge/tests/test_video.py index d6fbc67..626f38a 100644 --- a/inorbit_edge/tests/test_video.py +++ b/inorbit_edge/tests/test_video.py @@ -1,11 +1,46 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- +import threading +import time + from inorbit_edge.robot import RobotSession -from inorbit_edge.video import OpenCVCamera +from inorbit_edge.video import CameraStreamer, OpenCVCamera from inorbit_edge.robot import INORBIT_MODULE_CAMERAS +def _wait_until(predicate, timeout=2.0): + """Poll ``predicate`` until it is truthy or ``timeout`` elapses.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return bool(predicate()) + + +class FakeCamera: + """Minimal Camera double that records open/close calls without touching cv2.""" + + rate = 50 + + def __init__(self): + self._lock = threading.Lock() + self.opens = 0 + self.closes = 0 + + def open(self): + with self._lock: + self.opens += 1 + + def close(self): + with self._lock: + self.closes += 1 + + def get_frame_jpg(self): + return None, 0, 0, 0 + + def test_robot_session_register_camera( mock_mqtt_client, mock_inorbit_api, mocker, mock_sleep ): @@ -33,6 +68,9 @@ def test_robot_session_register_camera( robot_session._handle_in_cmd( f"load_module|{INORBIT_MODULE_CAMERAS}|{runlevel}".encode() ) + # The worker opens the camera asynchronously; wait until it has actually + # started before tearing down, so teardown has a live session to close. + assert _wait_until(lambda: opencv_camera.capture_thread is not None, timeout=5) # Override _is_disconnected method to simulate successful MQTT client disconnection robot_session._is_disconnected = lambda: True robot_session.disconnect() @@ -41,6 +79,71 @@ def test_robot_session_register_camera( camera_stream_stop_spy.assert_called_once() opencv_camera_close_spy.assert_called_once() - # Verify threads are stopped - assert not camera_stream.thread.is_alive() - assert not opencv_camera.capture_thread.is_alive() + # disconnect() shuts down and joins the worker, so it must be gone. Join again + # with a timeout to keep the assertion non-flaky regardless of scheduling. + camera_stream.join(timeout=15) + assert not camera_stream.is_alive() + if opencv_camera.capture_thread is not None: + opencv_camera.capture_thread.join(timeout=10) + assert not opencv_camera.capture_thread.is_alive() + + +def test_camera_streamer_start_stop_start_keeps_single_worker(): + """A stop->start cycle resumes streaming on the same single worker thread.""" + camera = FakeCamera() + streamer = CameraStreamer(camera, lambda *args: None) + worker = streamer._worker + try: + streamer.start() + assert _wait_until(lambda: camera.opens >= 1) + + streamer.stop() + assert _wait_until(lambda: camera.closes >= 1) + + streamer.start() + assert _wait_until(lambda: camera.opens >= 2) + + # Same worker thread throughout -- no respawn, no ghost/duplicate. + assert streamer._worker is worker + assert streamer.is_alive() + finally: + streamer.shutdown() + streamer.join(timeout=5) + assert not streamer.is_alive() + + +def test_camera_streamer_shutdown_terminates_worker(): + camera = FakeCamera() + streamer = CameraStreamer(camera, lambda *args: None) + streamer.start() + assert _wait_until(lambda: camera.opens >= 1) + + streamer.shutdown() + streamer.join(timeout=5) + assert not streamer.is_alive() + # The camera is released on the way out. + assert _wait_until(lambda: camera.closes >= 1) + + +def test_register_camera_twice_replaces_and_shuts_down_old( + mock_mqtt_client, mock_inorbit_api, mocker, mock_sleep +): + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + camera_id = "cam0" + + # Streaming is off (no load_module), so workers stay idle -- no cv2 open. + robot_session.register_camera(camera_id, OpenCVCamera(None)) + first = robot_session.camera_streamers[camera_id] + + robot_session.register_camera(camera_id, OpenCVCamera(None)) + second = robot_session.camera_streamers[camera_id] + + assert second is not first + first.join(timeout=5) + assert not first.is_alive() # old streamer's worker was shut down + assert second.is_alive() + + second.shutdown() + second.join(timeout=5) diff --git a/inorbit_edge/video.py b/inorbit_edge/video.py index 0495206..e96e7bb 100644 --- a/inorbit_edge/video.py +++ b/inorbit_edge/video.py @@ -76,14 +76,17 @@ def open(self): self.capture = cv2.VideoCapture(self.video_url) if not self.running: self.running = True - self.capture_thread = threading.Thread(target=self._run) + self.capture_thread = threading.Thread(target=self._run, daemon=True) self.capture_thread.start() def close(self): """Closes the capturing device / stream""" self.running = False - self.logger.info("Waiting for the capture thread to finish") - self.capture_thread.join() + + if self.capture_thread is not None: + self.logger.info("Waiting for the capture thread to finish") + self.capture_thread.join() + self.capture_thread = None with self.capture_mutex: if self.capture is not None: self.capture.release() @@ -114,43 +117,91 @@ def _run(self): class CameraStreamer: - """Streams video from a camera to InOrbit""" + """Streams video from a camera to InOrbit. + + A single long-lived worker thread owns the camera lifecycle. ``start``, + ``stop`` and ``shutdown`` only toggle threading Events, so they never block + the caller -- in particular the MQTT callback thread that dispatches module + load/unload, where a blocking call would starve the keepalive and drop the + connection. Opening/closing the camera (which can block while a stream is + unreachable) happens entirely on the worker thread. + + Because there is exactly one worker, ``camera.open()`` and ``camera.close()`` + are always serialized -- the device is never opened and closed concurrently, + which removes the start/stop races of a spawn-per-start model. + """ + + # Backoff after a failed streaming session so a permanently-broken URL does + # not hot-loop open->fail->open. A pending stop/shutdown short-circuits it. + BACKOFF_SECONDS = 1.0 def __init__(self, camera, publish_frame_callback): self.logger = logging.getLogger(__class__.__name__) self.camera = camera - self.running = False - self.mutex = threading.Lock() - self.thread = None self.publish_frame = publish_frame_callback - self.must_stop = False + # Set => the worker should be streaming. Cleared => paused. + self._streaming = threading.Event() + # Set => the worker should exit permanently. + self._shutdown = threading.Event() + self._worker = threading.Thread(target=self._run, daemon=True) + self._worker.start() def start(self): - """Streams video to the platform""" - with self.mutex: - self.must_stop = False - if not self.running: - self.running = True - self.thread = threading.Thread(target=self._run) - self.thread.start() + """Request streaming. Non-blocking; safe to call repeatedly.""" + self._streaming.set() def stop(self): - """Stops streaming video to the platform""" - self.must_stop = True - self.logger.info("Waiting for the streaming thread to finish") - self.thread.join() + """Pause streaming. Non-blocking; safe to call repeatedly. + + The worker thread stays alive and idle, ready for a later ``start()``. + Use ``shutdown()`` to terminate the worker permanently. + """ + self._streaming.clear() + + def shutdown(self): + """Permanently stop the worker. Non-blocking. + + ``_shutdown`` is set before ``_streaming`` so a worker waking from the + idle wait observes the shutdown rather than starting a doomed session. + """ + self._shutdown.set() + self._streaming.set() + + def join(self, timeout=None): + """Wait for the worker thread to exit (after ``shutdown()``).""" + self._worker.join(timeout) + + def is_alive(self): + """Return True while the worker thread is running.""" + return self._worker.is_alive() def _run(self): - """This thread takes care of getting video from a camera at the desired rate, - converting it to the right format and publishing the video frames""" - self.camera.open() - while True: - jpg, width, height, ts = self.camera.get_frame_jpg() - if jpg is not None: - self.publish_frame(jpg, width, height, ts) - time.sleep(1.0 / self.camera.rate) - with self.mutex: - if self.must_stop: - break - self.camera.close() - self.running = False + """Worker loop: idle until streaming is requested, then grab frames from + the camera at the desired rate and publish them until paused or shut down. + + The per-session body is wrapped so a failed ``open()``/``get_frame_jpg()`` + cannot kill the long-lived worker -- it logs, backs off, and waits for the + next ``start()``. + """ + while not self._shutdown.is_set(): + # Idle until streaming or shutdown is requested. + self._streaming.wait() + if self._shutdown.is_set(): + break + try: + self.camera.open() + while self._streaming.is_set() and not self._shutdown.is_set(): + jpg, width, height, ts = self.camera.get_frame_jpg() + if jpg is not None: + self.publish_frame(jpg, width, height, ts) + time.sleep(1.0 / self.camera.rate) + except Exception: + self.logger.exception("Camera streaming session failed") + # Back off so a permanently-failing open does not hot-loop while + # streaming is still requested. + self._shutdown.wait(timeout=self.BACKOFF_SECONDS) + finally: + try: + self.camera.close() + except Exception: + self.logger.exception("Error closing camera")