From fa60bc5a9064a6af6354f56f0cc3d47763f3e053 Mon Sep 17 00:00:00 2001 From: kshitij <82637225+kshitijk4poor@users.noreply.github.com> Date: Sun, 2 Aug 2026 11:46:30 +0530 Subject: [PATCH] fix(tts): apply 7 review fixes to prefetch pipeline - carry mark_audio_output_active into playback worker (df093bf33) - close temp WAV handle before playback (555d4e10a) - move sentinel + join into finally block (exception-path deadlock) - update _pcm_leftover before continue on reinit+rewrite success - remove dead _playback_done event - extract shared _create_output_stream helper (dedup) - extract shared _align_int16_chunks generator (dedup) --- tools/tts_tool.py | 310 +++++++++++++++++++--------------------------- 1 file changed, 125 insertions(+), 185 deletions(-) diff --git a/tools/tts_tool.py b/tools/tts_tool.py index 7f7d759b02bd8..67a0f48851f8f 100644 --- a/tools/tts_tool.py +++ b/tools/tts_tool.py @@ -3375,6 +3375,10 @@ def stream_tts_to_speaker( try: output_stream = None + streamer = None # type: ignore[assignment] + _worker_thread = None + _audio_queue = None # type: ignore[assignment] + _prefetch_threads = [] tts_config = _load_tts_config() # Prefer a chunked streamer for low time-to-first-audio; fall back to @@ -3427,31 +3431,26 @@ def stream_tts_to_speaker( # playing, so by the time the worker reaches it, audio is already # arriving — no inter-sentence gap. _audio_queue: queue.Queue[Optional[queue.Queue[Optional[bytes]]]] = queue.Queue() - _playback_done = threading.Event() _prefetch_threads: list[threading.Thread] = [] - # Limit concurrent prefetch threads so a long reply with many sentences - # doesn't spawn an unbounded number of HTTP connections / RAM buffers. - # The playback worker drains segments in FIFO order, so capping - # in-flight prefetches to 3 still keeps sentence N+1's audio warm - # while N plays, without letting a 100-sentence reply reserve 100 - # threads + 100 unbounded queues up front. _prefetch_sem = threading.Semaphore(3) - # Bound each per-sentence chunk queue so a slow playback worker - # can't let a fast prefetch thread buffer an entire sentence's - # audio in RAM. ~64 chunks ≈ 64 KB of int16 PCM at 2 bytes/frame - # — enough headroom for smooth playback without unbounded growth. _CHUNK_QUEUE_MAX = 64 + def _create_output_stream(): + """Create and start a fresh PortAudio OutputStream.""" + sd = _import_sounddevice() + new_stream = sd.OutputStream( + samplerate=streamer.sample_rate, + channels=streamer.channels, + dtype="int16", + ) + new_stream.start() + return new_stream + def _consume_to_queue( audio_iter: Iterator[bytes], chunk_queue: "queue.Queue[Optional[bytes]]", ) -> None: - """Consume a generator into a thread-safe queue. - - Fires the HTTP request immediately (the generator's first - iteration triggers it) and buffers every chunk so the playback - worker can drain without waiting for synthesis. - """ + """Consume a generator into a thread-safe queue.""" try: for chunk in audio_iter: if stop_event.is_set(): @@ -3472,13 +3471,7 @@ def stream_tts_to_speaker( _prefetch_sem.release() # free a prefetch slot def _reinit_output_stream(): - """Close the broken PortAudio stream and try to create a fresh one. - - Returns the new stream on success, ``None`` on failure. Updates - the enclosing ``output_stream`` so the finally block in - ``stream_tts_to_speaker`` closes the *current* stream, not the - stale broken one. - """ + """Close the broken PortAudio stream and try to create a fresh one.""" nonlocal output_stream if output_stream is not None: try: @@ -3487,13 +3480,7 @@ def stream_tts_to_speaker( except Exception: pass try: - sd = _import_sounddevice() - new_stream = sd.OutputStream( - samplerate=streamer.sample_rate, - channels=streamer.channels, - dtype="int16", - ) - new_stream.start() + new_stream = _create_output_stream() output_stream = new_stream logger.info( "TTS: PortAudio output stream reinitialized after error" @@ -3507,107 +3494,95 @@ def stream_tts_to_speaker( return None def _playback_worker() -> None: - """Single consumer: play audio segments from the queue in order. - - When PortAudio raises a transient error (e.g. PaErrorCode -9986 - on macOS device state changes), the worker attempts to - reinitialize the output stream (up to ``_max_reinit`` times). - If reinit fails it falls back to temp-file playback for the - remaining sentences instead of dropping them silently. - """ + """Single consumer: play audio segments from the queue in order.""" assert streamer is not None if output_stream is not None: import numpy as _np - _max_reinit = 3 - _reinit_count = 0 - _current_stream = output_stream - while True: - chunk_queue = _audio_queue.get() - if chunk_queue is None: - break - if stop_event.is_set(): - continue - # If the stream died and reinit is exhausted, fall back - # to temp-file playback for all remaining sentences. - if _current_stream is None: - _chunks = [] + + try: + from tools.voice_mode import mark_audio_output_active + except Exception: + def mark_audio_output_active(_active): + return None + + mark_audio_output_active(True) + try: + _max_reinit = 3 + _reinit_count = 0 + _current_stream = output_stream + while True: + chunk_queue = _audio_queue.get() + if chunk_queue is None: + break + if stop_event.is_set(): + continue + if _current_stream is None: + _chunks = [] + while True: + chunk = chunk_queue.get() + if chunk is None: + break + _chunks.append(chunk) + _play_via_tempfile( + iter(_chunks), stop_event, streamer.sample_rate + ) + continue + _pcm_leftover = b"" while True: chunk = chunk_queue.get() if chunk is None: break - _chunks.append(chunk) - _play_via_tempfile( - iter(_chunks), stop_event, streamer.sample_rate - ) - continue - _pcm_leftover = b"" - while True: - chunk = chunk_queue.get() - if chunk is None: - break - if stop_event.is_set(): - break - _buf = _pcm_leftover + chunk - _aligned_len = len(_buf) - (len(_buf) % 2) - if _aligned_len >= 2: - try: - _current_stream.write( - _np.frombuffer( - _buf[:_aligned_len], dtype="= 2: + try: + _current_stream.write( + _np.frombuffer( + _buf[:_aligned_len], dtype=" None: - """Synthesize *text_to_speak* and start prefetching immediately. - - A background thread begins consuming the generator (firing the - HTTP request) the moment this is called, buffering PCM chunks - into a per-segment queue. The queue is placed on the audio - queue for ordered playback by the worker thread. - """ + """Synthesize *text_to_speak* and start prefetching immediately.""" assert streamer is not None try: audio_iter = streamer.stream(text_to_speak) except Exception as exc: logger.warning("Streaming TTS synthesis failed: %s", exc) return - # Block until a prefetch slot is free — caps concurrent - # in-flight synthesis threads so a long reply doesn't - # reserve one thread + one queue per sentence up front. _prefetch_sem.acquire() chunk_queue: "queue.Queue[Optional[bytes]]" = queue.Queue(maxsize=_CHUNK_QUEUE_MAX) _audio_queue.put(chunk_queue) @@ -3655,7 +3619,6 @@ def stream_tts_to_speaker( _prefetch_threads.append(t) t.start() - # Start the single playback worker (only when we have a streamer). _worker_thread: Optional[threading.Thread] = None if streamer is not None: _worker_thread = threading.Thread(target=_playback_worker, daemon=True) @@ -3712,16 +3675,23 @@ def stream_tts_to_speaker( except OSError: pass - def _play_via_tempfile(audio_iter, stop_evt, sample_rate=24000): - """Write PCM chunks to a temp WAV file and play it. + def _align_int16_chunks(chunks, stop_evt): + """Yield int16-aligned byte chunks from an iterable.""" + leftover = b"" + for chunk in chunks: + if stop_evt.is_set(): + break + buf = leftover + chunk + aligned_len = len(buf) - (len(buf) % 2) + if aligned_len >= 2: + yield buf[:aligned_len] + leftover = buf[aligned_len:] if aligned_len < len(buf) else b"" + if leftover: + yield b"\x00" - Applies int16 (2-byte) frame alignment before writing: streaming - providers can yield chunks on arbitrary byte boundaries, and - ``wave.writeframes`` fed a half-sample at a chunk boundary would - produce a corrupted click. Leftover bytes are carried into the - next chunk so every frame written is whole — same carry logic as - the live playback path. - """ + def _play_via_tempfile(audio_iter, stop_evt, sample_rate=24000): + """Write PCM chunks to a temp WAV file and play it.""" + tmp = None tmp_path = None try: import wave @@ -3731,21 +3701,8 @@ def stream_tts_to_speaker( wf.setnchannels(1) wf.setsampwidth(2) # 16-bit wf.setframerate(sample_rate) - pcm_leftover = b"" - for chunk in audio_iter: - if stop_evt.is_set(): - break - buf = pcm_leftover + chunk - aligned_len = len(buf) - (len(buf) % 2) - if aligned_len >= 2: - wf.writeframes(buf[:aligned_len]) - pcm_leftover = ( - buf[aligned_len:] if aligned_len < len(buf) else b"" - ) - # Flush any trailing byte as silence (can't write half a - # sample, but a single leftover byte is inaudible). - if pcm_leftover: - wf.writeframes(b"\x00") + for aligned in _align_int16_chunks(audio_iter, stop_evt): + wf.writeframes(aligned) # wave.open() given a file object flushes but does NOT close it # (it only closes files it opened itself, by name), so the OS # handle to tmp stays open. On Windows an open write handle @@ -3796,35 +3753,18 @@ def stream_tts_to_speaker( except queue.Empty: break - # Signal the playback worker that no more audio is coming, then wait - # for it to finish playing everything in the queue. This ensures - # continuous voice mode doesn't start the next turn while audio is - # still playing. - # - # The timeout must be generous: a 1 000-character response is ~70 s - # of audio at 15 chars/s. The old 30-second timeout was too short - # for anything beyond a few sentences — the join would expire - # mid-playback, the finally block would close the output stream - # out from under the still-running worker, and the worker's next - # write would hit PortAudio -9986 on the closed stream. 300 s - # gives even the longest responses room to finish; the sentinel - # on the audio queue is the real exit signal, the timeout is just - # a safety net against a wedged worker. - if streamer is not None and _worker_thread is not None: - _audio_queue.put(None) - _worker_thread.join(timeout=300.0) - - # Join prefetch threads so in-flight HTTP requests complete before - # the pipeline exits (prevents orphaned connections on rapid turn - # changes). - for t in _prefetch_threads: - t.join(timeout=10.0) - # output_stream is closed in the finally block below except Exception as exc: logger.warning("Streaming TTS pipeline error: %s", exc) finally: + # Signal the playback worker that no more audio is coming. This lives + # in finally: so an exception in the text pump still sends the sentinel. + if streamer is not None and _worker_thread is not None: + _audio_queue.put(None) + _worker_thread.join(timeout=300.0) + for t in _prefetch_threads: + t.join(timeout=10.0) # Always close the audio output stream to avoid locking the device if output_stream is not None: try: