From 5b577b34e4b6b4047580726689d2a9f2294f4278 Mon Sep 17 00:00:00 2001 From: David Maier Date: Fri, 3 Jul 2026 14:45:09 +0200 Subject: [PATCH] Add configurable timeout for first-frame wait and improve thread-safety --- tests/test_base_backend.py | 41 +++++++++++++++++++++++++++++++++--- whisper_live/backend/base.py | 32 +++++++++++++++------------- 2 files changed, 55 insertions(+), 18 deletions(-) diff --git a/tests/test_base_backend.py b/tests/test_base_backend.py index a926936..576d455 100644 --- a/tests/test_base_backend.py +++ b/tests/test_base_backend.py @@ -3,7 +3,7 @@ import queue import threading import time import unittest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import numpy as np @@ -38,6 +38,9 @@ class WaitTrackingEvent: def set(self): self._event.set() + def __getattr__(self, name): + return getattr(self._event, name) + class TestServeClientBaseInit(unittest.TestCase): def test_default_values(self): @@ -106,7 +109,6 @@ class TestAddFrames(unittest.TestCase): # timestamp_offset should be bumped to at least frames_offset self.assertGreaterEqual(self.client.timestamp_offset, self.client.frames_offset) - class TestAddFramesThreadSafety(unittest.TestCase): def test_concurrent_add_frames(self): ws = MagicMock() @@ -129,6 +131,18 @@ class TestAddFramesThreadSafety(unittest.TestCase): self.assertEqual(errors, []) self.assertIsNotNone(client.frames_np) + def test_exception_releases_lock_without_signaling_frames_ready(self): + ws = MagicMock() + client = ConcreteServeClient(client_uid="test", websocket=ws) + client.frames_np = np.array([0.1], dtype=np.float32) + + with patch("whisper_live.backend.base.np.concatenate", side_effect=RuntimeError("boom")): + with self.assertRaisesRegex(RuntimeError, "boom"): + client.add_frames(np.array([0.2], dtype=np.float32)) + + self.assertFalse(client.lock.locked()) + self.assertFalse(client.frames_ready.is_set()) + class TestGetAudioChunkForProcessing(unittest.TestCase): def setUp(self): @@ -281,6 +295,16 @@ class TestCleanup(unittest.TestCase): self.assertTrue(client.exit) +def _supports_thread_time(): + thread_time = getattr(time, "thread_time", None) + if thread_time is None: + return False + try: + thread_time() + except NotImplementedError: + return False + return True + class TestSpeechToTextWaitingBehavior(unittest.TestCase): """Tests the first-frame wait behavior in speech_to_text().""" @@ -344,6 +368,17 @@ class TestSpeechToTextWaitingBehavior(unittest.TestCase): self.assertTrue(self.client.exit) self.client.transcribe_audio.assert_not_called() + def test_exit_flag_unblocks_waiting_thread_without_signal(self): + self.client.transcribe_audio = MagicMock() + self._start_speech_thread() + self.assertTrue(self.client.frames_ready.wait_started.wait(timeout=1.0)) + + self.client.exit = True + + self.assertTrue(self._join_speech_thread()) + self.client.transcribe_audio.assert_not_called() + + @unittest.skipUnless(_supports_thread_time(), "time.thread_time() not supported") def test_waiting_for_first_frame_uses_negligible_thread_cpu(self): self._start_speech_thread(target=self._measure_waiting_cpu) self.assertTrue(self.thread_started.wait(timeout=1.0)) @@ -353,7 +388,7 @@ class TestSpeechToTextWaitingBehavior(unittest.TestCase): self.assertTrue(self._join_speech_thread()) self.assertIsNotNone(self.cpu_used) - self.assertLess(self.cpu_used, 0.05) + self.assertLess(self.cpu_used, 0.1) class TestTrimTranscript(unittest.TestCase): diff --git a/whisper_live/backend/base.py b/whisper_live/backend/base.py index 690b485..afa2eca 100644 --- a/whisper_live/backend/base.py +++ b/whisper_live/backend/base.py @@ -21,6 +21,8 @@ class ServeClientBase(object): """Duration threshold in seconds for clipping audio with no valid segments.""" CLIP_TAIL_DURATION_S = 5 """Duration in seconds of audio to keep after clipping.""" + FIRST_FRAME_WAIT_TIMEOUT_S = 0.1 + """Interval in seconds for re-checking exit while waiting for the first audio frame.""" client_uid: str """A unique identifier for the client.""" @@ -105,7 +107,8 @@ class ServeClientBase(object): break if self.frames_np is None: - self.frames_ready.wait() + while self.frames_np is None and not self.exit: + self.frames_ready.wait(timeout=self.FIRST_FRAME_WAIT_TIMEOUT_S) continue if self.clip_audio: @@ -184,20 +187,19 @@ class ServeClientBase(object): frame_np (numpy.ndarray): The audio frame data as a NumPy array. """ - self.lock.acquire() - if self.frames_np is not None and self.frames_np.shape[0] > self.MAX_BUFFER_DURATION_S*self.RATE: - self.frames_offset += float(self.BUFFER_TRIM_DURATION_S) - self.frames_np = self.frames_np[int(self.BUFFER_TRIM_DURATION_S*self.RATE):] - # check timestamp offset(should be >= self.frame_offset) - # this basically means that there is no speech as timestamp offset hasnt updated - # and is less than frame_offset - if self.timestamp_offset < self.frames_offset: - self.timestamp_offset = self.frames_offset - if self.frames_np is None: - self.frames_np = frame_np.copy() - else: - self.frames_np = np.concatenate((self.frames_np, frame_np), axis=0) - self.lock.release() + with self.lock: + if self.frames_np is not None and self.frames_np.shape[0] > self.MAX_BUFFER_DURATION_S*self.RATE: + self.frames_offset += float(self.BUFFER_TRIM_DURATION_S) + self.frames_np = self.frames_np[int(self.BUFFER_TRIM_DURATION_S*self.RATE):] + # check timestamp offset(should be >= self.frame_offset) + # this basically means that there is no speech as timestamp offset hasnt updated + # and is less than frame_offset + if self.timestamp_offset < self.frames_offset: + self.timestamp_offset = self.frames_offset + if self.frames_np is None: + self.frames_np = frame_np.copy() + else: + self.frames_np = np.concatenate((self.frames_np, frame_np), axis=0) self.frames_ready.set() def clip_audio_if_no_valid_segment(self):