transports: streamline max_workers for ThreadPoolExecutors

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.

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),

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),

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: