fix: guard audio decode against locale-sensitive PyAV failures#6288
fix: guard audio decode against locale-sensitive PyAV failures#6288nightcityblade wants to merge 1 commit into
Conversation
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| with _temporary_av_locale(): | ||
| # open container in low-latency streaming mode | ||
| container = av.open( | ||
| self._input_buf, | ||
| mode="r", | ||
| format=self._av_format, | ||
| buffer_size=256, | ||
| options={ | ||
| "probesize": "32", | ||
| "analyzeduration": "0", | ||
| "fflags": "nobuffer+flush_packets", | ||
| "flags": "low_delay", | ||
| "reorder_queue_size": "0", | ||
| "max_delay": "0", | ||
| "avioflags": "direct", | ||
| }, | ||
| ) | ||
| # explicitly disable internal buffering flags on the FFmpeg container | ||
| container.flags |= cast( | ||
| int, av.container.Flags.no_buffer.value | av.container.Flags.flush_packets.value | ||
| ) | ||
|
|
||
| if len(container.streams.audio) == 0: | ||
| raise ValueError("no audio stream found") | ||
| if len(container.streams.audio) == 0: | ||
| raise ValueError("no audio stream found") | ||
|
|
||
| audio_stream = container.streams.audio[0] | ||
| audio_stream = container.streams.audio[0] | ||
|
|
||
| # Set up resampler only if needed | ||
| if self._sample_rate is not None or self._layout is not None: | ||
| resampler = av.AudioResampler( | ||
| format="s16", layout=self._layout, rate=self._sample_rate | ||
| ) | ||
| # Set up resampler only if needed | ||
| if self._sample_rate is not None or self._layout is not None: | ||
| resampler = av.AudioResampler( | ||
| format="s16", layout=self._layout, rate=self._sample_rate | ||
| ) | ||
|
|
||
| for frame in container.decode(audio_stream): | ||
| if self._closed: | ||
| return | ||
| for frame in container.decode(audio_stream): | ||
| if self._closed: | ||
| return | ||
|
|
||
| if resampler: | ||
| frames = resampler.resample(frame) | ||
| else: | ||
| frames = [frame] | ||
| if resampler: | ||
| frames = resampler.resample(frame) | ||
| else: | ||
| frames = [frame] | ||
|
|
||
| for f in frames: | ||
| self._emit_av_frame(f) | ||
| for f in frames: | ||
| self._emit_av_frame(f) | ||
|
|
||
| # flush the resampler to get any remaining buffered samples | ||
| if resampler and not self._closed: | ||
| for f in resampler.resample(None): | ||
| self._emit_av_frame(f) | ||
| # flush the resampler to get any remaining buffered samples | ||
| if resampler and not self._closed: | ||
| for f in resampler.resample(None): | ||
| self._emit_av_frame(f) |
There was a problem hiding this comment.
🔴 Global lock held for entire audio stream lifetime serializes all concurrent decoders
A process-wide lock is held for the entire duration of audio decoding (_temporary_av_locale() at livekit-agents/livekit/agents/utils/codecs/decoder.py:438) including blocking I/O waits, so concurrent audio streams are fully serialized.
Impact: When multiple agents or TTS sessions decode audio simultaneously, each waits for the previous one to finish entirely, causing severe latency.
The _AV_LOCALE_LOCK is held across the blocking decode loop
The _temporary_av_locale() context manager acquires _AV_LOCALE_LOCK (a global threading.Lock at decoder.py:37) and yields while holding it (decoder.py:80-84). The _decode_loop wraps the entire decode loop inside this context manager (decoder.py:438-486), including:
av.open(self._input_buf, ...)— which callsStreamBuffer.read()to probe the formatcontainer.decode(audio_stream)— a generator that repeatedly callsStreamBuffer.read(), which blocks atdecoder.py:145(self._data_available.wait()) until more data arrives
Since StreamBuffer is fed incrementally by push() calls, the lock is held for the entire lifetime of the audio stream (potentially minutes). A second AudioStreamDecoder in another thread (e.g., a separate TTS pipeline via livekit-agents/livekit/agents/tts/tts.py:1106) will block on acquiring this lock until the first stream completes entirely.
The lock's purpose is to protect a brief locale change, but its scope covers long-running blocking I/O. A narrower scope — e.g., wrapping only the av.open() call or individual container.decode() iterations — would avoid serializing unrelated streams.
Prompt for agents
The _temporary_av_locale() context manager in _decode_loop (decoder.py:438) holds the global _AV_LOCALE_LOCK for the entire duration of audio decoding, including blocking StreamBuffer.read() calls. This serializes all concurrent AudioStreamDecoder instances in the process.
The lock is needed to protect the process-wide locale change (setlocale is not thread-safe and affects the entire process). However, it should be held for the minimum time necessary.
Possible approaches:
1. Narrow the scope: only wrap av.open() (the format-probing step) in the locale context manager, since that's the most likely locale-sensitive operation. The decode loop itself (container.decode) would run outside the lock.
2. If container.decode is also locale-sensitive, consider acquiring and releasing the lock around each individual frame decode iteration rather than the entire loop.
3. Consider using a different mechanism altogether, such as setting the locale once at process startup rather than toggling it per-decode.
Was this helpful? React with 👍 or 👎 to provide feedback.
| @contextmanager | ||
| def _temporary_av_locale() -> Iterator[None]: | ||
| """Force a stable C locale while PyAV/FFmpeg decodes audio. | ||
|
|
||
| PyAV can surface locale-sensitive decode paths through FFmpeg. Some | ||
| non-ASCII locales trigger UnicodeDecodeError while pulling audio frames, | ||
| so we temporarily switch to the C locale around decode/resample work and | ||
| then restore the caller's locale. | ||
| """ | ||
|
|
||
| with _AV_LOCALE_LOCK: | ||
| previous = locale.setlocale(locale.LC_ALL) | ||
| try: | ||
| locale.setlocale(locale.LC_ALL, "C") | ||
| yield | ||
| finally: | ||
| locale.setlocale(locale.LC_ALL, previous) |
There was a problem hiding this comment.
🚩 locale.setlocale is not thread-safe per POSIX / CPython docs
Beyond the lock-scope issue reported as a bug, there is a deeper concern: locale.setlocale() affects the entire process and the CPython docs explicitly warn it is not thread-safe. Even with the _AV_LOCALE_LOCK serializing callers of _temporary_av_locale, other threads in the process (including the event loop thread, other plugins, or user code) may observe the temporarily changed locale without any synchronization. This is inherent to the approach and worth acknowledging — the lock only protects against concurrent _temporary_av_locale callers, not against the broader process seeing 'C' locale mid-operation.
Was this helpful? React with 👍 or 👎 to provide feedback.
|
nightcityblade seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account. You have signed the CLA already but the status is still pending? Let us recheck it. |
| then restore the caller's locale. | ||
| """ | ||
|
|
||
| with _AV_LOCALE_LOCK: |
There was a problem hiding this comment.
the thread lock serializes every audio decode in this process.
|
I dug into this and agree the current lock scope is too broad. The tricky part is that the locale workaround is process-wide, so simply narrowing the lock can reintroduce the original decode failure. I haven't pushed a follow-up yet because I want to avoid trading one correctness bug for another; I'll come back with a safer revision rather than guessing here. |
4 participants
Fixes #2886
Summary
Validation
uv run pytest tests/test_audio_decoder.py -q -k "temporary_av_locale or stream_buffer"uv run ruff check livekit-agents/livekit/agents/utils/codecs/decoder.py tests/test_audio_decoder.py