Merge pull request #159 from pipecat-ai/create-pool-executor

transports: run threads in their own ThreadPoolExecutor

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to **pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Fixed an issue where tasks and threads could be paused because the executor
+  didn't have more tasks available. This was causing issues when cancelling and
+  recreating tasks during interruptions.
+
 ## [0.0.19] - 2024-05-20
 
 ### Changed

src/pipecat/transports/base_input.py

@@ -7,6 +7,8 @@
 import asyncio
 import queue
 
+from concurrent.futures import ThreadPoolExecutor
+
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.frames.frames import (
     AudioRawFrame,
@@ -34,7 +36,9 @@ class BaseInputTransport(FrameProcessor):
         self._running = False
         self._allow_interruptions = False
 
-        # Start media threads.
+        self._in_executor = ThreadPoolExecutor(max_workers=5)
+
+        # Create audio input queue if needed.
         if self._params.audio_in_enabled or self._params.vad_enabled:
             self._audio_in_queue = queue.Queue()
 
@@ -55,8 +59,10 @@ class BaseInputTransport(FrameProcessor):
 
         if self._params.audio_in_enabled or self._params.vad_enabled:
             loop = self.get_event_loop()
-            self._audio_in_thread = loop.run_in_executor(None, self._audio_in_thread_handler)
-            self._audio_out_thread = loop.run_in_executor(None, self._audio_out_thread_handler)
+            self._audio_in_thread = loop.run_in_executor(
+                self._in_executor, self._audio_in_thread_handler)
+            self._audio_out_thread = loop.run_in_executor(
+                self._in_executor, self._audio_out_thread_handler)
 
     async def stop(self):
         if not self._running:

src/pipecat/transports/base_output.py

@@ -11,6 +11,8 @@ import queue
 import time
 import threading
 
+from concurrent.futures import ThreadPoolExecutor
+
 from PIL import Image
 from typing import List
 
@@ -41,6 +43,8 @@ class BaseOutputTransport(FrameProcessor):
         self._running = False
         self._allow_interruptions = False
 
+        self._out_executor = ThreadPoolExecutor(max_workers=5)
+
         # These are the images that we should send to the camera at our desired
         # framerate.
         self._camera_images = None
@@ -67,9 +71,10 @@ class BaseOutputTransport(FrameProcessor):
         loop = self.get_event_loop()
 
         if self._params.camera_out_enabled:
-            self._camera_out_thread = loop.run_in_executor(None, self._camera_out_thread_handler)
+            self._camera_out_thread = loop.run_in_executor(
+                self._out_executor, self._camera_out_thread_handler)
 
-        self._sink_thread = loop.run_in_executor(None, self._sink_thread_handler)
+        self._sink_thread = loop.run_in_executor(self._out_executor, self._sink_thread_handler)
 
         # Create push frame task. This is the task that will push frames in
         # order. We also guarantee that all frames are pushed in the same task.

src/pipecat/transports/services/daily.py

@@ -5,6 +5,7 @@
 #
 
 import asyncio
+from concurrent.futures import ThreadPoolExecutor
 import inspect
 import queue
 import time
@@ -146,6 +147,8 @@ class DailyTransportClient(EventHandler):
         self._leaving = False
         self._sync_response = {k: queue.Queue() for k in ["join", "leave"]}
 
+        self._executor = ThreadPoolExecutor(max_workers=5)
+
         self._client: CallClient = CallClient(event_handler=self)
 
         self._camera: VirtualCameraDevice = Daily.create_camera_device(
@@ -195,7 +198,7 @@ class DailyTransportClient(EventHandler):
         self._joining = True
 
         loop = asyncio.get_running_loop()
-        await loop.run_in_executor(None, self._join)
+        await loop.run_in_executor(self._executor, self._join)
 
     def _join(self):
         logger.info(f"Joining {self._room_url}")
@@ -287,7 +290,7 @@ class DailyTransportClient(EventHandler):
         self._leaving = True
 
         loop = asyncio.get_running_loop()
-        await loop.run_in_executor(None, self._leave)
+        await loop.run_in_executor(self._executor, self._leave)
 
     def _leave(self):
         logger.info(f"Leaving {self._room_url}")
@@ -318,7 +321,7 @@ class DailyTransportClient(EventHandler):
 
     async def cleanup(self):
         loop = asyncio.get_running_loop()
-        await loop.run_in_executor(None, self._cleanup)
+        await loop.run_in_executor(self._executor, self._cleanup)
 
     def _cleanup(self):
         if self._client:
@@ -438,7 +441,8 @@ class DailyInputTransport(BaseInputTransport):
         await super().start(frame)
         # Create camera in thread (runs if _running is true).
         loop = asyncio.get_running_loop()
-        self._camera_in_thread = loop.run_in_executor(None, self._camera_in_thread_handler)
+        self._camera_in_thread = loop.run_in_executor(
+            self._in_executor, self._camera_in_thread_handler)
 
     async def stop(self):
         if not self._running: