Repository navigation
fix(rtc): keep an AudioMixer input alive after a read timeout - #862
xianshijing-lk merged 6 commits into
Conversation
The mixer read each stream with asyncio.wait_for(stream.__anext__(), ...). On timeout wait_for cancels that read, and cancelling __anext__ of an async generator closes it, so one late frame ended the stream and the rest of its audio was dropped. Keep the timed-out read and await it again in the next round; remove_stream() and aclose() cancel it.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
aclose() now waits for cancelled reads to run their cleanup, and a read cancelled by remove_stream() has its outcome consumed whenever it ends. A round scheduled before remove_stream() no longer starts a new read on the removed stream, which left an entry in _pending_reads (seen on the Windows CI run). A stream that raises asyncio.TimeoutError itself is handled like the mixer's own timeout again, so the samples it already produced are kept.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
A stream that catches CancelledError and keeps waiting made aclose() hang. aclose() now waits for cancelled reads for at most stream_timeout_ms; their outcome is still consumed when they end, and cancelled reads are referenced until then.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…lose" This reverts commit 2fe5a4e. aclose() goes back to awaiting the cancelled reads in full, which is what main does (wait_for awaits the cancelled read inside the mixer task, and aclose() awaits that task). A stream that ignores cancellation hangs shutdown on main as well; changing that is a separate decision from this fix.
| for read in pending: | ||
| _discard_read(read) | ||
| # let the streams finish their cleanup before the mixer counts as closed | ||
| await asyncio.gather(*pending, return_exceptions=True) |
There was a problem hiding this comment.
🔴 Mixer shutdown hangs on uncancellable input
When a pending stream read ignores cancellation, aclose() waits for it indefinitely. OutputPlayer.aclose() then cannot finish closing the player.
Learn more
A timed-out input read remains in _pending_reads until it completes. aclose() cancels these reads, but cancellation is cooperative: an async iterator can catch CancelledError and continue waiting. Awaiting all reads without a deadline then blocks shutdown indefinitely. OutputPlayer.aclose awaits mixer shutdown before clearing its mixer and closing the output device.
Example: A stream repeatedly catches CancelledError while waiting on an unset event. After its read times out, await mixer.aclose() never returns; closing an OutputPlayer with that stream also leaves its output device open.
Recommended fix: Bound the wait for cancelled reads in AudioMixer.aclose() using asyncio.wait(..., timeout=...) or an equivalent non-cancelling timeout. If cleanup outlives shutdown, retain and consume those tasks until they complete.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
This one is the same as on main, not introduced here: on main, wait_for cancels the read inside the mixer task and waits for it to finish, and aclose() awaits that task, so a stream that keeps catching CancelledError hangs aclose() there too (checked with a script against main's audio_mixer.py: aclose() had not returned after 10 s). The bounded wait you suggest was tried in 2fe5a4e and reverted in 09b6002: it cuts cleanup longer than stream_timeout_ms short and leaves reads outliving the mixer (your two comments on that commit). Keeping main's shutdown semantics here and leaving that trade-off to the maintainers; it is listed under Scope in the description.
There was a problem hiding this comment.
Documented the contract in e57d4fd instead: aclose() cancels and awaits pending reads, so a stream must let CancelledError propagate once its cleanup is done. That is the contract asyncio itself states ("it should generally be propagated when clean-up is complete"; TaskGroup and asyncio.timeout() "might misbehave if a coroutine swallows asyncio.CancelledError", https://docs.python.org/3/library/asyncio-task.html#task-cancellation), and TaskGroup likewise waits for cancelled children without a deadline. A bounded wait was tried in 2fe5a4e and reverted (it cut legitimate cleanup short and left reads outliving the mixer), and main hangs on such a stream as well.
| # a round scheduled before remove_stream() must not start a new read on that stream | ||
| while buf.shape[0] < self._chunk_size and not exhausted and stream in self._streams: | ||
| # The read stays in _pending_reads until it completes, so remove_stream() | ||
| # and aclose() can always cancel it. |
There was a problem hiding this comment.
how about end_input(): ?
when end_input() is called, _ending is set to true, but it doesn't cancel anything, termination of that drain depends entirely on the _mixer loop's exit condition, and in that case, end_input() no longer terminates on a permanently stalled generator ?
I wonder if we should consider bounding how long a stream may stall before it's dropped for real ?
There was a problem hiding this comment.
Good catch, thanks. That's a regression from this PR: on main the first timeout closed the stream, so the drain ended; here a stream that never yields again kept the drain alive forever.
5074667 bounds it: after end_input(), a stream that times out 10 rounds in a row (about 1 s at the default 100 ms) is dropped for real and its pending read is cancelled. A single late frame is still kept. Before end_input() streams aren't dropped, since a TTS stream can legitimately go quiet between utterances. Test: test_mixer_drops_a_stalled_stream_after_end_input (passes on main, fails on the previous commit, passes now).
Happy to make the bound a constructor argument instead, or apply it outside end_input() as well, if you'd prefer either.
On main a read timeout closed the stream, so a stalled stream was dropped and the drain after end_input() ended. With the read now kept across timeouts, a stream that never yields again kept the drain alive forever. After end_input(), a stream is now dropped after 10 consecutive timeouts; a single late frame is still kept.
| stalled = self._stalled_rounds.get(stream, 0) + 1 | ||
| self._stalled_rounds[stream] = stalled | ||
| # after end_input() a stream that keeps stalling is dropped, so the drain ends | ||
| exhausted = self._ending and stalled >= _MAX_STALLED_ROUNDS |
There was a problem hiding this comment.
🔴 Slow inputs lose remaining audio
After end_input(), a read delayed beyond ten timeouts makes _get_contribution remove its stream. The pending frame and every later frame disappear instead of draining.
Learn more
end_input() prevents new streams but otherwise drains existing streams until they finish. The mixer reuses a pending read across timeout rounds, but ten timeouts now mark its stream exhausted even when that read is still running. The removal cancels the read, so a delayed but valid frame and all later frames are lost.
Example: With the default 100 ms timeout, a generator yielding one frame, sleeping 1.2 seconds, and then yielding another loses its second frame after end_input(). The mixer stops roughly one second into the sleep, before that frame arrives.
Recommended fix: Keep waiting for pending reads during normal end_input() drainage, or introduce an explicit, configurable forced-drain deadline separate from the existing per-round timeout. Do not treat ten temporary timeouts as proof of stream exhaustion.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
This is the intended bound, added at the reviewer's request (thread above) and approved: after end_input() a stream must not stall for more than 10 consecutive timeouts, so the drain is guaranteed to end. It is looser than main, where the first timeout during the drain closed the stream and dropped the rest of its audio; a single late frame is still kept (test_mixer_keeps_async_generator_after_a_timeout).
xianshijing-lk
left a comment
xianshijing-lk
left a comment
There was a problem hiding this comment.
thanks for addressing the comment, lgtm
|
Thanks for the review and the end_input() catch! |
|
I am merging the PR now, thanks for the fixes |
|
Thanks for merging! In case it's useful, these are my other open LiveKit PRs, each small with a failing test on main:
No rush at all, happy to adjust any of them. |
2 participants
AudioMixer._get_contributionreads each input withasyncio.wait_for(stream.__anext__(), timeout=stream_timeout_ms / 1000). When a read times out,wait_forcancels it, and cancelling__anext__()of an async generator throwsCancelledErrorinto the generator and closes it. The next round getsStopAsyncIteration, so the mixer treats the stream as finished. A single frame that arrives later thanstream_timeout_msends that input and drops the rest of its audio, with only the "timeout, ignoring" warning in the log. Async generators are a normal input here;rtc.utils.sine_wave_generator, used in the existing mixer test, is one.The timed-out read is now kept in
_pending_readsand awaited again in the next round instead of being cancelled. It stays in that dict until it completes, soremove_stream()andaclose()can always cancel it; its outcome is consumed whenever it ends, so a failure isn't reported as "never retrieved", andaclose()waits for the cancelled reads to finish their cleanup, asmaindoes. A round scheduled beforeremove_stream()starts no new read on the removed stream. Timeouts, including anasyncio.TimeoutErrorraised by the stream itself, still log the same warning and the round still pads that stream with silence, as before. Afterend_input(), a stream that times out 10 rounds in a row is dropped, so the drain still ends on a stream that never yields again (onmainthe first timeout closed it).Tests in
tests/rtc/test_mixer.py:test_mixer_keeps_async_generator_after_a_timeout: fails onmain(assert 3 == 10), passes with this change.test_mixer_cancels_a_pending_read_when_the_stream_goes_away[remove_stream|aclose]: the parked read is cancelled (the generator'sfinallyruns) and nothing is left in_pending_reads.test_mixer_starts_no_read_on_a_removed_stream,test_mixer_keeps_buffered_audio_when_a_stream_raises_timeout_error,test_mixer_aclose_waits_for_a_cancelled_read_to_clean_up: added for the review in 8642fa8; each fails on the first commit and passes now.test_mixer_drops_a_stalled_stream_after_end_input: added in 5074667; passes onmain, fails on e57d4fd, passes now.Ran
tests/rtcagainst the publishedlivekitwheel with onlyaudio_mixer.pyswapped: the new tests are the only difference between main and this branch (test_agc/test_apm/test_resamplerfail andtest_encoded_videoerrors identically on both here, from missing local fixtures).ruff format --check,ruff check(0.15.4) andmypyon the module pass.Scope: a stream whose read raises something other than
StopAsyncIterationis still skipped silently bygather(..., return_exceptions=True), as before; I left that alone. A stream that catchesCancelledErrorand keeps waiting makesaclose()hang, onmainand here alike; bounding that wait (2fe5a4e, reverted in 09b6002) cuts other streams' cleanup short. Theaclose()docstring now states the contract asyncio itself sets: a stream must letCancelledErrorpropagate once its cleanup is done.