From f387776985867891ca4bd6b1b743323060e85690 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 20 Aug 2025 10:58:09 -0700 Subject: [PATCH] add custom asyncio.wait_for() This patch uses `wait_for2` package to implement `asyncio.wait_for()` for Python < 3.12. In Python 3.12, `asyncio.wait_for()` is implemented in terms of `asyncio.timeout()` which fixed a bunch of issues. However, this was never backported (because of the lack of `async.timeout()`) and there are still many remainig issues, specially in Python 3.10, in `async.wait_for()`. See https://github.com/python/cpython/pull/98518 --- CHANGELOG.md | 8 ++++++ pyproject.toml | 1 + scripts/evals/eval.py | 3 +- src/pipecat/pipeline/task.py | 7 ++--- .../processors/aggregators/dtmf_aggregator.py | 3 +- .../processors/aggregators/llm_response.py | 3 +- .../processors/idle_frame_processor.py | 3 +- src/pipecat/processors/user_idle_processor.py | 3 +- src/pipecat/services/anthropic/llm.py | 5 ++-- src/pipecat/services/assemblyai/stt.py | 8 ++---- src/pipecat/services/aws/llm.py | 5 ++-- src/pipecat/services/aws/stt.py | 3 +- src/pipecat/services/heygen/client.py | 3 +- src/pipecat/services/heygen/video.py | 3 +- src/pipecat/services/openai/base_llm.py | 3 +- src/pipecat/services/riva/tts.py | 5 ++-- src/pipecat/services/speechmatics/stt.py | 3 +- src/pipecat/services/tts_service.py | 7 +++-- src/pipecat/transports/base_input.py | 6 ++-- src/pipecat/transports/base_output.py | 5 ++-- .../transports/network/small_webrtc.py | 5 ++-- src/pipecat/transports/services/daily.py | 5 ++-- src/pipecat/utils/asyncio/task_manager.py | 8 ++++-- src/pipecat/utils/asyncio/timeout.py | 28 +++++++++++++++++++ .../utils/asyncio/watchdog_async_iterator.py | 6 ++-- .../utils/asyncio/watchdog_coroutine.py | 6 ++-- src/pipecat/utils/asyncio/watchdog_event.py | 3 +- .../utils/asyncio/watchdog_priority_queue.py | 3 +- src/pipecat/utils/asyncio/watchdog_queue.py | 3 +- tests/test_pipeline.py | 7 +++-- uv.lock | 11 ++++++++ 31 files changed, 119 insertions(+), 53 deletions(-) create mode 100644 src/pipecat/utils/asyncio/timeout.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f962d33b..6f979a6f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `pipecat.utils.timeout.wait_for()` which uses `wait_for2` package to + avoid `asyncio.wait_for()` issues in Python < 3.12. + - Allow passing custom pipeline sink and source processors to a `Pipeline`. Pipeline source and sink processors are used to know and control what's coming in and out of a `Pipeline` processor. @@ -46,6 +49,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Replaced `asyncio.wait_for()` for Pipecat's `wait_for()`. In Python 3.10, + `asyncio.wait_for()` has issues regarding task cancellation (i.e. cancellation + is never propagated). + See https://bugs.python.org/issue42130 + - Fixed an `AudioBufferProcessor` issues that would cause audio overlap when setting a max buffer size. diff --git a/pyproject.toml b/pyproject.toml index d61c3efe3..81948e886 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ dependencies = [ "openai>=1.74.0,<=1.99.1", # Pinning numba to resolve package dependencies "numba==0.61.2", + "wait_for2>=0.4.1; python_version<'3.12'", ] [project.urls] diff --git a/scripts/evals/eval.py b/scripts/evals/eval.py index b91f27c6e..b78b98bca 100644 --- a/scripts/evals/eval.py +++ b/scripts/evals/eval.py @@ -43,6 +43,7 @@ from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.llm_service import FunctionCallParams from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.utils.asyncio.timeout import wait_for SCRIPT_DIR = Path(__file__).resolve().parent @@ -122,7 +123,7 @@ class EvalRunner: logger.error(f"ERROR: Unable to run {example_file}: {e}") try: - result = await asyncio.wait_for(self._queue.get(), timeout=1.0) + result = await wait_for(self._queue.get(), timeout=1.0) except asyncio.TimeoutError: result = False diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 7ff924531..92cda3039 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -51,6 +51,7 @@ from pipecat.utils.asyncio.task_manager import ( TaskManager, TaskManagerParams, ) +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue from pipecat.utils.tracing.setup import is_tracing_available from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver @@ -654,7 +655,7 @@ class PipelineTask(BasePipelineTask): wait_time = HEARTBEAT_MONITOR_SECONDS while True: try: - frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time) + frame = await wait_for(self._heartbeat_queue.get(), timeout=wait_time) process_time = (self._clock.get_time() - frame.timestamp) / 1_000_000_000 logger.trace(f"{self}: heartbeat frame processed in {process_time} seconds") self._heartbeat_queue.task_done() @@ -677,9 +678,7 @@ class PipelineTask(BasePipelineTask): while running: try: - frame = await asyncio.wait_for( - self._idle_queue.get(), timeout=self._idle_timeout_secs - ) + frame = await wait_for(self._idle_queue.get(), timeout=self._idle_timeout_secs) if not isinstance(frame, InputAudioRawFrame): frame_buffer.append(frame) diff --git a/src/pipecat/processors/aggregators/dtmf_aggregator.py b/src/pipecat/processors/aggregators/dtmf_aggregator.py index 43c59b661..e55ae66bc 100644 --- a/src/pipecat/processors/aggregators/dtmf_aggregator.py +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -25,6 +25,7 @@ from pipecat.frames.frames import ( TranscriptionFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.time import time_now_iso8601 @@ -136,7 +137,7 @@ class DTMFAggregator(FrameProcessor): """Background task that handles timeout-based flushing.""" while True: try: - await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout) + await wait_for(self._digit_event.wait(), timeout=self._idle_timeout) self._digit_event.clear() except asyncio.TimeoutError: self.reset_watchdog() diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 38848ac1d..1d251d495 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -60,6 +60,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.time import time_now_iso8601 @@ -670,7 +671,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): if self._vad_params else self._params.turn_emulated_vad_timeout ) - await asyncio.wait_for(self._aggregation_event.wait(), timeout) + await wait_for(self._aggregation_event.wait(), timeout=timeout) await self._maybe_emulate_user_speaking() except asyncio.TimeoutError: if not self._user_speaking: diff --git a/src/pipecat/processors/idle_frame_processor.py b/src/pipecat/processors/idle_frame_processor.py index b8839124a..5357b2f64 100644 --- a/src/pipecat/processors/idle_frame_processor.py +++ b/src/pipecat/processors/idle_frame_processor.py @@ -11,6 +11,7 @@ from typing import Awaitable, Callable, List, Optional from pipecat.frames.frames import Frame, StartFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_event import WatchdogEvent @@ -85,7 +86,7 @@ class IdleFrameProcessor(FrameProcessor): """Handle idle timeout monitoring and callback execution.""" while True: try: - await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout) + await wait_for(self._idle_event.wait(), timeout=self._timeout) except asyncio.TimeoutError: await self._callback(self) finally: diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py index 5f6b25b95..a38e3f17f 100644 --- a/src/pipecat/processors/user_idle_processor.py +++ b/src/pipecat/processors/user_idle_processor.py @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_event import WatchdogEvent @@ -191,7 +192,7 @@ class UserIdleProcessor(FrameProcessor): """ while True: try: - await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout) + await wait_for(self._idle_event.wait(), timeout=self._timeout) except asyncio.TimeoutError: if not self._interrupted: self._retry_count += 1 diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index e351ecab2..93d39e265 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -53,6 +53,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator from pipecat.utils.tracing.service_decorators import traced_llm @@ -185,9 +186,7 @@ class AnthropicLLMService(LLMService): """ if self._retry_on_timeout: try: - response = await asyncio.wait_for( - api_call(**params), timeout=self._retry_timeout_secs - ) + response = await wait_for(api_call(**params), timeout=self._retry_timeout_secs) return response except (APITimeoutError, asyncio.TimeoutError): # Retry, this time without a timeout so we get a response diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index d601b3ad5..c999a3eed 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -31,6 +31,7 @@ from pipecat.frames.frames import ( from pipecat.processors.frame_processor import FrameDirection from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_stt @@ -219,10 +220,7 @@ class AssemblyAISTTService(STTService): await self._websocket.send(json.dumps({"type": "Terminate"})) try: - await asyncio.wait_for( - self._termination_event.wait(), - timeout=5.0, - ) + await wait_for(self._termination_event.wait(), timeout=5.0) except asyncio.TimeoutError: logger.warning("Timed out waiting for termination message from server") @@ -247,7 +245,7 @@ class AssemblyAISTTService(STTService): try: while self._connected: try: - message = await asyncio.wait_for(self._websocket.recv(), timeout=1.0) + message = await wait_for(self._websocket.recv(), timeout=1.0) data = json.loads(message) await self._handle_message(data) except asyncio.TimeoutError: diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 7a97c6cdb..cdf144d26 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -52,6 +52,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import LLMService +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.tracing.service_decorators import traced_llm try: @@ -801,8 +802,8 @@ class AWSBedrockLLMService(LLMService): """ if self._retry_on_timeout: try: - response = await asyncio.wait_for( - client.converse_stream(**request_params), timeout=self._retry_timeout_secs + response = await wait_for( + await client.converse_stream(**request_params), timeout=self._retry_timeout_secs ) return response except (ReadTimeoutError, asyncio.TimeoutError) as e: diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index f67f0979f..752866efa 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -31,6 +31,7 @@ from pipecat.frames.frames import ( from pipecat.services.aws.utils import build_event_message, decode_event, get_presigned_url from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_stt @@ -480,7 +481,7 @@ class AWSTranscribeSTTService(STTService): break try: - response = await asyncio.wait_for(self._ws_client.recv(), timeout=1.0) + response = await wait_for(self._ws_client.recv(), timeout=1.0) headers, payload = decode_event(response) diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py index 2a464f949..a91f5e675 100644 --- a/src/pipecat/services/heygen/client.py +++ b/src/pipecat/services/heygen/client.py @@ -31,6 +31,7 @@ from pipecat.processors.frame_processor import FrameProcessorSetup from pipecat.services.heygen.api import HeyGenApi, HeyGenSession, NewSessionRequest from pipecat.transports.base_transport import TransportParams from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue try: @@ -231,7 +232,7 @@ class HeyGenClient: """Handle incoming WebSocket messages.""" while self._connected: try: - message = await asyncio.wait_for(self._websocket.recv(), timeout=1.0) + message = await wait_for(self._websocket.recv(), timeout=1.0) parsed_message = json.loads(message) await self._handle_ws_server_event(parsed_message) except asyncio.TimeoutError: diff --git a/src/pipecat/services/heygen/video.py b/src/pipecat/services/heygen/video.py index 96c684641..f46ec7d76 100644 --- a/src/pipecat/services/heygen/video.py +++ b/src/pipecat/services/heygen/video.py @@ -40,6 +40,7 @@ from pipecat.services.ai_service import AIService from pipecat.services.heygen.api import NewSessionRequest from pipecat.services.heygen.client import HEY_GEN_SAMPLE_RATE, HeyGenCallbacks, HeyGenClient from pipecat.transports.base_transport import TransportParams +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue # Using the same values that we do in the BaseOutputTransport @@ -320,7 +321,7 @@ class HeyGenVideoService(AIService): while True: try: - frame = await asyncio.wait_for(self._queue.get(), timeout=AVATAR_VAD_STOP_SECS) + frame = await wait_for(self._queue.get(), timeout=AVATAR_VAD_STOP_SECS) if self._is_interrupting: break if isinstance(frame, TTSAudioRawFrame): diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index ecde66eec..fba80f238 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -39,6 +39,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator from pipecat.utils.tracing.service_decorators import traced_llm @@ -196,7 +197,7 @@ class BaseOpenAILLMService(LLMService): if self._retry_on_timeout: try: - chunks = await asyncio.wait_for( + chunks = await wait_for( self._client.chat.completions.create(**params), timeout=self._retry_timeout_secs ) return chunks diff --git a/src/pipecat/services/riva/tts.py b/src/pipecat/services/riva/tts.py index 6e5937c6f..a9f5f77d8 100644 --- a/src/pipecat/services/riva/tts.py +++ b/src/pipecat/services/riva/tts.py @@ -14,6 +14,7 @@ import asyncio import os from typing import AsyncGenerator, Mapping, Optional +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.tracing.service_decorators import traced_tts # Suppress gRPC fork warnings @@ -168,7 +169,7 @@ class RivaTTSService(TTSService): await asyncio.to_thread(read_audio_responses, queue) # Wait for the thread to start. - resp = await asyncio.wait_for(queue.get(), RIVA_TTS_TIMEOUT_SECS) + resp = await wait_for(queue.get(), timeout=RIVA_TTS_TIMEOUT_SECS) while resp: await self.stop_ttfb_metrics() frame = TTSAudioRawFrame( @@ -177,7 +178,7 @@ class RivaTTSService(TTSService): num_channels=1, ) yield frame - resp = await asyncio.wait_for(queue.get(), RIVA_TTS_TIMEOUT_SECS) + resp = await wait_for(queue.get(), timeout=RIVA_TTS_TIMEOUT_SECS) except asyncio.TimeoutError: logger.error(f"{self} timeout waiting for audio response") diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 59145742e..523c2b89d 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -33,6 +33,7 @@ from pipecat.frames.frames import ( from pipecat.processors.frame_processor import FrameDirection from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.tracing.service_decorators import traced_stt try: @@ -576,7 +577,7 @@ class SpeechmaticsSTTService(STTService): # Disconnect the client try: if self._client: - await asyncio.wait_for(self._client.close(), timeout=1.0) + await wait_for(self._client.close(), timeout=1.0) except asyncio.TimeoutError: logger.warning("Timeout while closing Speechmatics client connection") except Exception as e: diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 556276494..9e0bb61f6 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -37,6 +37,7 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.text.base_text_filter import BaseTextFilter @@ -427,8 +428,8 @@ class TTSService(AIService): has_started = False while True: try: - frame = await asyncio.wait_for( - self._stop_frame_queue.get(), self._stop_frame_timeout_s + frame = await wait_for( + self._stop_frame_queue.get(), timeout=self._stop_frame_timeout_s ) if isinstance(frame, TTSStartedFrame): has_started = True @@ -858,7 +859,7 @@ class AudioContextWordTTSService(WebsocketWordTTSService): running = True while running: try: - frame = await asyncio.wait_for(queue.get(), timeout=AUDIO_CONTEXT_TIMEOUT) + frame = await wait_for(queue.get(), timeout=AUDIO_CONTEXT_TIMEOUT) self.reset_watchdog() if frame: await self.push_frame(frame) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index bbb9179bc..dfa0e1c2f 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -49,6 +49,8 @@ from pipecat.frames.frames import ( from pipecat.metrics.metrics import MetricsData from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams +from pipecat.utils.asyncio.timeout import wait_for +from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue AUDIO_INPUT_TIMEOUT_SECS = 0.5 @@ -395,7 +397,7 @@ class BaseInputTransport(FrameProcessor): def _create_audio_task(self): """Create the audio processing task if audio input is enabled.""" if not self._audio_task and self._params.audio_in_enabled: - self._audio_in_queue = asyncio.Queue() + self._audio_in_queue = WatchdogQueue(self.task_manager) self._audio_task = self.create_task(self._audio_task_handler()) async def _cancel_audio_task(self): @@ -474,7 +476,7 @@ class BaseInputTransport(FrameProcessor): vad_state: VADState = VADState.QUIET while True: try: - frame: InputAudioRawFrame = await asyncio.wait_for( + frame: InputAudioRawFrame = await wait_for( self._audio_in_queue.get(), timeout=AUDIO_INPUT_TIMEOUT_SECS ) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index afb5abb0b..5975a9be7 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -46,6 +46,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_priority_queue import WatchdogPriorityQueue from pipecat.utils.time import nanoseconds_to_seconds @@ -623,9 +624,7 @@ class BaseOutputTransport(FrameProcessor): async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[Frame, None]: while True: try: - frame = await asyncio.wait_for( - self._audio_queue.get(), timeout=vad_stop_secs - ) + frame = await wait_for(self._audio_queue.get(), timeout=vad_stop_secs) self._transport.reset_watchdog() yield frame except asyncio.TimeoutError: diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index ddb8339c9..5d0429982 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -40,6 +40,7 @@ from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator try: @@ -289,7 +290,7 @@ class SmallWebRTCClient: continue try: - frame = await asyncio.wait_for(self._video_input_track.recv(), timeout=2.0) + frame = await wait_for(self._video_input_track.recv(), timeout=2.0) except asyncio.TimeoutError: if self._webrtc_connection.is_connected(): logger.warning("Timeout: No video frame received within the specified time.") @@ -332,7 +333,7 @@ class SmallWebRTCClient: continue try: - frame = await asyncio.wait_for(self._audio_input_track.recv(), timeout=2.0) + frame = await wait_for(self._audio_input_track.recv(), timeout=2.0) except asyncio.TimeoutError: if self._webrtc_connection.is_connected(): logger.warning("Timeout: No audio frame received within the specified time.") diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 77f7f5e29..77ac20bc1 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -49,6 +49,7 @@ from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.timeout import wait_for from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue try: @@ -694,7 +695,7 @@ class DailyTransportClient(EventHandler): }, ) - return await asyncio.wait_for(future, timeout=10) + return await wait_for(future, timeout=10) async def leave(self): """Leave the Daily room and cleanup resources.""" @@ -735,7 +736,7 @@ class DailyTransportClient(EventHandler): """Execute the actual room leave operation.""" future = self._get_event_loop().create_future() self._client.leave(completion=completion_callback(future)) - return await asyncio.wait_for(future, timeout=10) + return await wait_for(future, timeout=10) def _cleanup(self): """Cleanup the Daily client instance.""" diff --git a/src/pipecat/utils/asyncio/task_manager.py b/src/pipecat/utils/asyncio/task_manager.py index 3bc3453ac..58e337034 100644 --- a/src/pipecat/utils/asyncio/task_manager.py +++ b/src/pipecat/utils/asyncio/task_manager.py @@ -20,6 +20,8 @@ from typing import Coroutine, Dict, Optional, Sequence from loguru import logger +from pipecat.utils.asyncio.timeout import wait_for + WATCHDOG_TIMEOUT = 5.0 @@ -285,7 +287,7 @@ class TaskManager(BaseTaskManager): name = task.get_name() try: if timeout: - await asyncio.wait_for(task, timeout=timeout) + await wait_for(task, timeout=timeout) else: await task except asyncio.TimeoutError: @@ -315,7 +317,7 @@ class TaskManager(BaseTaskManager): # Make sure to reset watchdog if a task is cancelled. self.reset_watchdog(task) if timeout: - await asyncio.wait_for(task, timeout=timeout) + await wait_for(task, timeout=timeout) else: await task except asyncio.TimeoutError: @@ -402,7 +404,7 @@ class TaskManager(BaseTaskManager): break start_time = time.time() - await asyncio.wait_for(timer.wait(), timeout=watchdog_timeout) + await wait_for(timer.wait(), timeout=watchdog_timeout) total_time = time.time() - start_time if enable_watchdog_logging: logger.debug(f"{name} time between watchdog timer resets: {total_time:.20f}") diff --git a/src/pipecat/utils/asyncio/timeout.py b/src/pipecat/utils/asyncio/timeout.py new file mode 100644 index 000000000..345c205bb --- /dev/null +++ b/src/pipecat/utils/asyncio/timeout.py @@ -0,0 +1,28 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Implementation of Python's `asyncio.wait_for()`. + +This module uses `wait_for2` package to implement `asyncio.wait_for()` for +Python < 3.12. + +In Python 3.12, `asyncio.wait_for()` is implemented in terms of +`asyncio.timeout()` which fixed a bunch of issues. However, this was never +backported (because of the lack of `async.timeout()`) and there are still many +remainig issues, specially in Python 3.10, in `async.wait_for()`. + +See https://github.com/python/cpython/pull/98518 +""" + +import asyncio +import sys + +import wait_for2 + +if sys.version_info >= (3, 12): + wait_for = asyncio.wait_for +else: + wait_for = wait_for2.wait_for diff --git a/src/pipecat/utils/asyncio/watchdog_async_iterator.py b/src/pipecat/utils/asyncio/watchdog_async_iterator.py index c9db0ba7e..4594568e0 100644 --- a/src/pipecat/utils/asyncio/watchdog_async_iterator.py +++ b/src/pipecat/utils/asyncio/watchdog_async_iterator.py @@ -15,6 +15,7 @@ import asyncio from typing import AsyncIterator, Optional from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.timeout import wait_for class WatchdogAsyncIterator: @@ -77,9 +78,8 @@ class WatchdogAsyncIterator: if not self._current_anext_task: self._current_anext_task = asyncio.create_task(self._iter.__anext__()) - item = await asyncio.wait_for( - asyncio.shield(self._current_anext_task), - timeout=self._timeout, + item = await wait_for( + asyncio.shield(self._current_anext_task), timeout=self._timeout ) self._manager.task_reset_watchdog() diff --git a/src/pipecat/utils/asyncio/watchdog_coroutine.py b/src/pipecat/utils/asyncio/watchdog_coroutine.py index 234776548..7827eb779 100644 --- a/src/pipecat/utils/asyncio/watchdog_coroutine.py +++ b/src/pipecat/utils/asyncio/watchdog_coroutine.py @@ -16,6 +16,7 @@ from typing import Optional from pipecat.pipeline import task from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.timeout import wait_for class WatchdogCoroutine: @@ -59,9 +60,8 @@ class WatchdogCoroutine: if not self._current_coro_task: self._current_coro_task = asyncio.create_task(self._coroutine) - result = await asyncio.wait_for( - asyncio.shield(self._current_coro_task), - timeout=self._timeout, + result = await wait_for( + asyncio.shield(self._current_coro_task), timeout=self._timeout ) self._manager.task_reset_watchdog() diff --git a/src/pipecat/utils/asyncio/watchdog_event.py b/src/pipecat/utils/asyncio/watchdog_event.py index c48356c50..cc5c6c19a 100644 --- a/src/pipecat/utils/asyncio/watchdog_event.py +++ b/src/pipecat/utils/asyncio/watchdog_event.py @@ -14,6 +14,7 @@ watchdog timeouts during legitimate waiting periods. import asyncio from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.timeout import wait_for class WatchdogEvent(asyncio.Event): @@ -55,7 +56,7 @@ class WatchdogEvent(asyncio.Event): """Wait for event while periodically resetting watchdog timer.""" while True: try: - await asyncio.wait_for(super().wait(), timeout=self._timeout) + await wait_for(super().wait(), timeout=self._timeout) self._manager.task_reset_watchdog() return True except asyncio.TimeoutError: diff --git a/src/pipecat/utils/asyncio/watchdog_priority_queue.py b/src/pipecat/utils/asyncio/watchdog_priority_queue.py index c9b8fe171..384af195c 100644 --- a/src/pipecat/utils/asyncio/watchdog_priority_queue.py +++ b/src/pipecat/utils/asyncio/watchdog_priority_queue.py @@ -17,6 +17,7 @@ from dataclasses import dataclass from loguru import logger from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.timeout import wait_for @dataclass @@ -124,7 +125,7 @@ class WatchdogPriorityQueue(asyncio.PriorityQueue): """Get item from queue while periodically resetting watchdog timer.""" while True: try: - item = await asyncio.wait_for(super().get(), timeout=self._timeout) + item = await wait_for(super().get(), timeout=self._timeout) self._manager.task_reset_watchdog() return item except asyncio.TimeoutError: diff --git a/src/pipecat/utils/asyncio/watchdog_queue.py b/src/pipecat/utils/asyncio/watchdog_queue.py index 45bcaff66..6968a4903 100644 --- a/src/pipecat/utils/asyncio/watchdog_queue.py +++ b/src/pipecat/utils/asyncio/watchdog_queue.py @@ -17,6 +17,7 @@ from dataclasses import dataclass from loguru import logger from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.timeout import wait_for @dataclass @@ -103,7 +104,7 @@ class WatchdogQueue(asyncio.Queue): """Get item from queue while periodically resetting watchdog timer.""" while True: try: - item = await asyncio.wait_for(super().get(), timeout=self._timeout) + item = await wait_for(super().get(), timeout=self._timeout) self._manager.task_reset_watchdog() return item except asyncio.TimeoutError: diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 4b2c34828..293cae9f0 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -24,6 +24,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.filters.identity_filter import IdentityFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.tests.utils import HeartbeatsObserver, run_test +from pipecat.utils.asyncio.timeout import wait_for class TestPipeline(unittest.IsolatedAsyncioTestCase): @@ -250,7 +251,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): await task.queue_frame(TextFrame(text="Hello Downstream!")) try: - await asyncio.wait_for( + await wait_for( asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), timeout=1.0, ) @@ -286,7 +287,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): await task.queue_frame(TextFrame(text="Hello!")) try: - await asyncio.wait_for( + await wait_for( asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), timeout=1.0, ) @@ -306,7 +307,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): pipeline = Pipeline([identity]) task = PipelineTask(pipeline, idle_timeout_secs=0.2, cancel_on_idle_timeout=False) try: - await asyncio.wait_for( + await wait_for( asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), timeout=0.3, ) diff --git a/uv.lock b/uv.lock index bbe6a1ca8..6edbbc7c3 100644 --- a/uv.lock +++ b/uv.lock @@ -4195,6 +4195,7 @@ dependencies = [ { name = "pyloudnorm" }, { name = "resampy" }, { name = "soxr" }, + { name = "wait-for2", marker = "python_full_version < '3.12'" }, ] [package.optional-dependencies] @@ -4466,6 +4467,7 @@ requires-dist = [ { name = "transformers", marker = "extra == 'ultravox'", specifier = ">=4.48.0" }, { name = "uvicorn", marker = "extra == 'runner'", specifier = ">=0.32.0,<1.0.0" }, { name = "vllm", marker = "extra == 'ultravox'", specifier = ">=0.9.0" }, + { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" }, { name = "websockets", marker = "extra == 'assemblyai'", specifier = ">=13.1,<15.0" }, { name = "websockets", marker = "extra == 'asyncai'", specifier = ">=13.1,<15.0" }, { name = "websockets", marker = "extra == 'aws'", specifier = ">=13.1,<15.0" }, @@ -7316,6 +7318,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f4/72/c14ff1acac64294f45782769b9c8144a1c3e8d4f2228d4648197511b015a/vllm-0.9.2-cp38-abi3-manylinux1_x86_64.whl", hash = "sha256:f3c5da29a286f4933b480a5b4749fab226564f35c96928eeef547f88d385cd34", size = 383350132, upload-time = "2025-07-08T04:48:54.133Z" }, ] +[[package]] +name = "wait-for2" +version = "0.4.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/8f/7c/ea09d6a11990a8aa3ceac206fb7ea82366ea2c200caa87966611e0e18597/wait_for2-0.4.1.tar.gz", hash = "sha256:7f415415d21845c441391d6b4abe68f5959d2c0fbe927c2f61be28a297bc2acb", size = 17519, upload-time = "2025-06-13T19:45:00.081Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/90/56/0f88040567af7ff376ec9eaabe18fd980a4f5089d3bf8c7a32598ef06b8d/wait_for2-0.4.1-py3-none-any.whl", hash = "sha256:c694503e8c7420929e8a86bcffd9b00d55acaec2c14223a2b1e92bdc2ebf2154", size = 10985, upload-time = "2025-06-13T19:44:58.82Z" }, +] + [[package]] name = "watchfiles" version = "1.1.0"