Repository navigation
fix(rtc): offload FFI handle disposal to thread in aclose and add sync close - #872
NabiBukhsh-AI wants to merge 1 commit into
Conversation
…c close (livekit#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 livekit#832
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
There was a problem hiding this comment.
Devin Review found 2 potential issues.
1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)
| 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() |
There was a problem hiding this comment.
🔴 Synchronous close strands stream tasks
When close() runs before EOS, both streams unsubscribe while _run still waits for events. The task remains pending, retaining resources, and a later aclose() waits forever.
Learn more
A stream owns a background _run task that waits on its FFI subscription for frame or end-of-stream events. Synchronous close() removes that subscription immediately after dropping the native handle. If the task has not already handled EOS, it can no longer receive that event and stays pending. The same sequence applies to VideoStream.close. The pending task keeps its stream alive; AudioStream.aclose and VideoStream.aclose then wait on that task without completing.
Example: A user creates a stream while connected to a track, calls stream.close() before the track ends, then calls await stream.aclose(). The event subscription has gone away, so _run never sees EOS and aclose() never returns.
Recommended fix: Make both synchronous close methods terminate or wake _run before unsubscribing. Also arrange a terminal signal for consumers waiting on __anext__, and make subsequent aclose() calls able to join the terminated task.
Was this helpful? React with 👍 or 👎 to provide feedback.
| self._release_waiter() | ||
| self._ffi_handle.dispose() |
There was a problem hiding this comment.
🟡 Cross-thread audio close can strand playout
If close() runs outside the source’s active loop, _release_waiter completes a loop-owned future from the wrong thread. Pending wait_for_playout() calls can remain blocked, and debug mode raises RuntimeError.
Learn more
An AudioSource binds to an event loop during construction, and capture_frame creates a future on that loop. A caller can use the new synchronous close method from another thread while that loop runs. _release_waiter calls set_result directly rather than scheduling it on the owning loop. Asyncio debug mode rejects the cross-thread wakeup; without debug mode, the waiting loop need not wake promptly.
Example: A service runs an audio source on a worker event loop and calls source.close() from its management thread during playback. The playout waiter stays blocked, or close() raises before disposing the handle.
Recommended fix: Dispatch waiter and timer cleanup through self._loop.call_soon_threadsafe when outside the loop thread, while preserving synchronous disposal and ensuring close cannot race with a pending timer callback.
Was this helpful? React with 👍 or 👎 to provide feedback.
This branch has not been deployed
1 participant
Fixes #832
Problem
In
AudioSource,VideoSource,AudioStream, andVideoStream,aclose()is declaredasync, but it directly executedself._ffi_handle.dispose()synchronously on the event loop thread. Because handle disposal invokes the native C FFI calllivekit_ffi_drop_handle, any stall or wait in native teardown blocked the asyncio event loop and stalled other tasks in the process until a watchdog fired.Additionally:
AudioSource,VideoSource,AudioStream, andVideoStreamlacked a synchronousclose()method, unlikeDataStreamandDataTrack, making it difficult to cleanly close resources in synchronous contexts outside of an event loop.AudioSource, callingaclose()did not release pendingwait_for_playout()futures or cancel the scheduled_join_handletimer callback.VideoSource.capture_frame(), frames pushed after disposal were not guarded against calling the FFI with a disposed handle.Solution
aclose(): Runself._ffi_handle.dispose()viaasyncio.to_threadinAudioSource.aclose(),VideoSource.aclose(),AudioStream.aclose(), andVideoStream.aclose(). This ensures the synchronous native drop never blocks the asyncio event loop.close(): Added synchronousclose()methods toAudioSource,VideoSource,AudioStream, andVideoStreamfor parity withDataStream/DataTrack.AudioSource._release_waiter(), explicitly cancelself._join_handleif active. Call_release_waiter()from bothclose()andaclose()so tasks awaitingwait_for_playout()are promptly unblocked.if self._ffi_handle.disposed: returnguard toVideoSource.capture_frame()(mirroringAudioSource).Verification
livekit-rtc/tests/test_source_stream_close.py:AudioSource.aclose()releases pendingwait_for_playoutwaiters and cancels active timer handles.AudioSource.close()synchronous cleanup.VideoSource.close()andaclose(), and that post-close capture is safely ignored.uv run pytest livekit-rtc/tests/test_source_stream_close.pyuv run mypy livekit-protocol livekit-api livekit-rtc(106 files clean).uv run ruff checkanduv run ruff format.