diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 994bd1351..93442f90a 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -35,7 +35,9 @@ class BaseInputTransport(FrameProcessor): self._params = params - self._executor = ThreadPoolExecutor(max_workers=5) + # We read audio from a single queue one at a time and we then run VAD in + # a thread. Therefore, only one thread should be necessary. + self._executor = ThreadPoolExecutor(max_workers=1) # Task to process incoming audio (VAD) and push audio frames downstream # if passthrough is enabled. diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index c042c3fde..c65279be9 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -69,7 +69,9 @@ class LocalAudioOutputTransport(BaseOutputTransport): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) - self._executor = ThreadPoolExecutor(max_workers=5) + # We only write audio frames from a single task, so only one thread + # should be necessary. + self._executor = ThreadPoolExecutor(max_workers=1) self._out_stream = py_audio.open( format=py_audio.get_format_from_width(2), diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 23c70868c..842eac2df 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -77,7 +77,9 @@ class TkOutputTransport(BaseOutputTransport): def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) - self._executor = ThreadPoolExecutor(max_workers=5) + # We only write audio frames from a single task, so only one thread + # should be necessary. + self._executor = ThreadPoolExecutor(max_workers=1) self._out_stream = py_audio.open( format=py_audio.get_format_from_width(2), diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index d3a56d5a7..4d474e9c9 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -202,7 +202,9 @@ class DailyTransportClient(EventHandler): self._joined = False self._leave_counter = 0 - self._executor = ThreadPoolExecutor(max_workers=5) + # We use the executor to cleanup the client. We just do it from one + # place, so only one thread is really needed. + self._executor = ThreadPoolExecutor(max_workers=1) self._client: CallClient = CallClient(event_handler=self) @@ -466,9 +468,11 @@ class DailyTransportClient(EventHandler): return await asyncio.wait_for(future, timeout=10) async def cleanup(self): - await self._loop.run_in_executor(self._executor, self._cleanup) self._callback_task.cancel() await self._callback_task + # Make sure we don't block the event loop in case `client.release()` + # takes extra time. + await self._loop.run_in_executor(self._executor, self._cleanup) def _cleanup(self): if self._client: