From db4aa564f48acf2e75905d62ba6fea7f338d3237 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Sat, 17 Oct 2026 00:17:25 +0500 Subject: [PATCH] fix(rtc): offload FFI handle disposal to thread in aclose and add sync close (#832) - In AudioSource, AudioStream, VideoSource, and VideoStream, add a synchronous close() method. - In AudioSource, VideoSource, AudioStream, and VideoStream, run self._ffi_handle.dispose() via asyncio.to_thread in aclose() so the synchronous native drop call does not block the asyncio event loop. - In AudioSource, cancel any pending _join_handle timer and resolve _join_fut when close() or aclose() is called so waiters on wait_for_playout are immediately unblocked. - In VideoSource.capture_frame, safely ignore frames if the handle is already disposed. Fixes #832 --- livekit-rtc/livekit/rtc/audio_source.py | 17 +++- livekit-rtc/livekit/rtc/audio_stream.py | 13 ++- livekit-rtc/livekit/rtc/video_source.py | 18 +++- livekit-rtc/livekit/rtc/video_stream.py | 8 +- livekit-rtc/tests/test_source_stream_close.py | 95 +++++++++++++++++++ 5 files changed, 145 insertions(+), 6 deletions(-) create mode 100644 livekit-rtc/tests/test_source_stream_close.py diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py index 63cc1a5d..a63a2df0 100644 --- a/livekit-rtc/livekit/rtc/audio_source.py +++ b/livekit-rtc/livekit/rtc/audio_source.py @@ -163,6 +163,10 @@ async def wait_for_playout(self) -> None: await asyncio.shield(self._join_fut) def _release_waiter(self) -> None: + if self._join_handle is not None: + self._join_handle.cancel() + self._join_handle = None + if self._join_fut is None: return # could be None when clear_queue is called @@ -173,9 +177,18 @@ def _release_waiter(self) -> None: self._q_size = 0.0 self._join_fut = None - async def aclose(self) -> None: - """Close the audio source + def close(self) -> None: + """Close the audio source. This method cleans up resources associated with the audio source. """ + self._release_waiter() self._ffi_handle.dispose() + + async def aclose(self) -> None: + """Close the audio source without blocking the event loop. + + This method cleans up resources associated with the audio source. + """ + self._release_waiter() + await asyncio.to_thread(self._ffi_handle.dispose) diff --git a/livekit-rtc/livekit/rtc/audio_stream.py b/livekit-rtc/livekit/rtc/audio_stream.py index 97772f4f..b4411cad 100644 --- a/livekit-rtc/livekit/rtc/audio_stream.py +++ b/livekit-rtc/livekit/rtc/audio_stream.py @@ -315,15 +315,24 @@ async def _run(self) -> None: FfiClient.instance.queue.unsubscribe(self._ffi_queue) + def close(self) -> None: + """Synchronously close the audio stream and release its resources.""" + if self._track is not None: + self._track._unregister_audio_stream(self) + self._ffi_handle.dispose() + FfiClient.instance.queue.unsubscribe(self._ffi_queue) + if self._processor is not None and self._processor_auto_close: + self._processor._close() + async def aclose(self) -> None: """Asynchronously close the audio stream. This method cleans up resources associated with the audio stream and waits for - any pending operations to complete. + any pending operations to complete without blocking the event loop. """ if self._track is not None: self._track._unregister_audio_stream(self) - self._ffi_handle.dispose() + await asyncio.to_thread(self._ffi_handle.dispose) await self._task if self._processor is not None and self._processor_auto_close: self._processor._close() diff --git a/livekit-rtc/livekit/rtc/video_source.py b/livekit-rtc/livekit/rtc/video_source.py index c8dd2a02..ad34fb2d 100644 --- a/livekit-rtc/livekit/rtc/video_source.py +++ b/livekit-rtc/livekit/rtc/video_source.py @@ -14,6 +14,8 @@ from __future__ import annotations +import asyncio + from ._ffi_client import FfiHandle, FfiClient from ._proto import ffi_pb2 as proto_ffi from ._proto import video_frame_pb2 as proto_video @@ -61,6 +63,9 @@ def capture_frame( metadata: proto_video.FrameMetadata | None = None, ) -> None: """Capture a frame, optionally attaching frame metadata.""" + if self._ffi_handle.disposed: + return + req = proto_ffi.FfiRequest() req.capture_video_frame.source_handle = self._ffi_handle.handle req.capture_video_frame.buffer.CopyFrom(frame._proto_info()) @@ -70,5 +75,16 @@ def capture_frame( req.capture_video_frame.metadata.CopyFrom(metadata) FfiClient.instance.request(req) - async def aclose(self) -> None: + def close(self) -> None: + """Close the video source. + + This method cleans up resources associated with the video source. + """ self._ffi_handle.dispose() + + async def aclose(self) -> None: + """Close the video source without blocking the event loop. + + This method cleans up resources associated with the video source. + """ + await asyncio.to_thread(self._ffi_handle.dispose) diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py index eeced550..8c686538 100644 --- a/livekit-rtc/livekit/rtc/video_stream.py +++ b/livekit-rtc/livekit/rtc/video_stream.py @@ -164,8 +164,14 @@ async def _run(self) -> None: FfiClient.instance.queue.unsubscribe(self._ffi_queue) - async def aclose(self) -> None: + def close(self) -> None: + """Synchronously close the video stream and release its resources.""" self._ffi_handle.dispose() + FfiClient.instance.queue.unsubscribe(self._ffi_queue) + + async def aclose(self) -> None: + """Asynchronously close the video stream without blocking the event loop.""" + await asyncio.to_thread(self._ffi_handle.dispose) await self._task def _is_event(self, e: proto_ffi.FfiEvent) -> bool: diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py new file mode 100644 index 00000000..cd84ac1d --- /dev/null +++ b/livekit-rtc/tests/test_source_stream_close.py @@ -0,0 +1,95 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + + +import pytest + +from livekit import rtc + + +@pytest.mark.asyncio +async def test_audio_source_aclose_releases_waiter_and_cancels_timer() -> None: + source = rtc.AudioSource(48000, 1) + frame = rtc.AudioFrame.create(48000, 1, 48000) + await source.capture_frame(frame) + + assert source._join_fut is not None + assert source._join_handle is not None + + await source.aclose() + + assert source._ffi_handle.disposed + assert source._join_fut is None + assert source._join_handle is None + + # After aclose, wait_for_playout should return immediately without hanging + await source.wait_for_playout() + + # Repeated calls are idempotent + await source.aclose() + + +def test_audio_source_sync_close() -> None: + source = rtc.AudioSource(48000, 1) + assert not source._ffi_handle.disposed + + source.close() + assert source._ffi_handle.disposed + + # Repeated close is idempotent + source.close() + + +@pytest.mark.asyncio +async def test_audio_source_capture_after_close_is_noop() -> None: + source = rtc.AudioSource(48000, 1) + source.close() + + frame = rtc.AudioFrame.create(48000, 1, 480) + # Should safely return without error + await source.capture_frame(frame) + + +def test_video_source_sync_close() -> None: + source = rtc.VideoSource(640, 480) + assert not source._ffi_handle.disposed + + source.close() + assert source._ffi_handle.disposed + + # Repeated close is idempotent + source.close() + + +@pytest.mark.asyncio +async def test_video_source_aclose_and_capture_after_close() -> None: + source = rtc.VideoSource(640, 480) + assert not source._ffi_handle.disposed + + await source.aclose() + assert source._ffi_handle.disposed + + # Capture after close is safely ignored + frame = rtc.VideoFrame( + 640, + 480, + rtc.VideoBufferType.RGBA, + b"\x00" * (640 * 480 * 4), + ) + source.capture_frame(frame) + + # Repeated aclose is idempotent + await source.aclose()