From 825738440e12eaaba24eb712e8f010612b45c020 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 21 May 2024 18:50:29 -0700 Subject: [PATCH] transports: run threads in their own ThreadPoolExecutor --- CHANGELOG.md | 8 ++++++++ src/pipecat/transports/base_input.py | 12 +++++++++--- src/pipecat/transports/base_output.py | 9 +++++++-- src/pipecat/transports/services/daily.py | 12 ++++++++---- 4 files changed, 32 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 654eace66..1a3f9fbbb 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index c670bb757..4d4e4c987 100644 --- a/src/pipecat/transports/base_input.py +++ b/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: diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index a960f1534..377dcbab8 100644 --- a/src/pipecat/transports/base_output.py +++ b/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. diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 7fe2523f0..c0948495f 100644 --- a/src/pipecat/transports/services/daily.py +++ b/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: