diff --git a/CHANGELOG.md b/CHANGELOG.md index a8e536d09..1267263df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Updated `daily-python` to 0.19.7. +### Deprecated + +- `FrameProcessor.wait_for_task()` is deprecated. Use `await task` or `await + asyncio.wait_for(task, timeout)` instead. + ### Removed - Watchdog timers have been removed. They were introduced in 0.0.72 to help diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index b2f76ec60..84f73ebad 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -28,7 +28,7 @@ SPEAKING_THRESHOLD = 20 def create_default_resampler(**kwargs) -> BaseAudioResampler: """Create a default audio resampler instance. - . deprecated:: 0.0.74 + .. deprecated:: 0.0.74 This function is deprecated and will be removed in a future version. Use `create_stream_resampler` for real-time processing scenarios or `create_file_resampler` for batch processing of complete audio files. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 27e1fcd9a..487cf248f 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -363,9 +363,9 @@ class PipelineTask(BasePipelineTask): # Create all main tasks and wait of the main push task. This is the # task that pushes frames to the very beginning of our pipeline (our - # controlled PipelineTaskSource processor). + # controlled source processor). push_task = await self._create_tasks() - await self._task_manager.wait_for_task(push_task) + await push_task # We have already cleaned up the pipeline inside the task. cleanup_pipeline = False diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 7fed35f1c..3a5e42f3a 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -985,10 +985,6 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): def _context_updated_task_finished(self, task: asyncio.Task): self._context_updated_tasks.discard(task) - # The task is finished so this should exit immediately. We need to do - # this because otherwise the task manager would report a dangling task - # if we don't remove it. - asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop()) class LLMUserResponseAggregator(LLMUserContextAggregator): diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index fe36804fe..ab82f7aa1 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -442,11 +442,27 @@ class FrameProcessor(BaseObject): async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None): """Wait for a task to complete. + .. deprecated:: 0.0.81 + This function is deprecated, use `await task` or + `await asyncio.wait_for(task, timeout) instead. + Args: task: The task to wait for. timeout: Optional timeout for waiting. """ - await self.task_manager.wait_for_task(task, timeout) + import warnings + + warnings.warn( + "`FrameProcessor.wait_for_task()` is deprecated. " + "Use `await task` or `await asyncio.wait_for(task, timeout)` instead.", + DeprecationWarning, + stacklevel=2, + ) + + if timeout: + await asyncio.wait_for(task, timeout) + else: + await task async def setup(self, setup: FrameProcessorSetup): """Set up the processor with required components. diff --git a/src/pipecat/processors/metrics/sentry.py b/src/pipecat/processors/metrics/sentry.py index 75bd63c48..3a66bb1da 100644 --- a/src/pipecat/processors/metrics/sentry.py +++ b/src/pipecat/processors/metrics/sentry.py @@ -63,7 +63,7 @@ class SentryMetrics(FrameProcessorMetrics): await super().cleanup() if self._sentry_task: await self._sentry_queue.put(None) - await self.task_manager.wait_for_task(self._sentry_task) + await self._sentry_task self._sentry_task = None logger.trace(f"{self} Flushing Sentry metrics") sentry_sdk.flush(timeout=5.0) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index c3ad69e0e..8f12a598b 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -487,7 +487,7 @@ class LLMService(AIService): self._function_call_tasks[task] = runner_item # Since we run tasks sequentially we don't need to call # task.add_done_callback(self._function_call_task_finished). - await self.wait_for_task(task) + await task del self._function_call_tasks[task] async def _run_function_call(self, runner_item: FunctionCallRunnerItem): @@ -616,7 +616,3 @@ class LLMService(AIService): def _function_call_task_finished(self, task: asyncio.Task): if task in self._function_call_tasks: del self._function_call_tasks[task] - # The task is finished so this should exit immediately. We need to - # do this because otherwise the task manager would report a dangling - # task if we don't remove it. - asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop()) diff --git a/src/pipecat/services/soniox/stt.py b/src/pipecat/services/soniox/stt.py index 8121b1d74..1cf2d5194 100644 --- a/src/pipecat/services/soniox/stt.py +++ b/src/pipecat/services/soniox/stt.py @@ -208,7 +208,7 @@ class SonioxSTTService(STTService): if self._receive_task: # Task cannot cancel itself. If task called _cleanup() we expect it to cancel itself. if self._receive_task != asyncio.current_task(): - await self.wait_for_task(self._receive_task) + await self._receive_task self._receive_task = None async def stop(self, frame: EndFrame): diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 90eb1646d..2a210f093 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -794,7 +794,7 @@ class AudioContextWordTTSService(WebsocketWordTTSService): # Indicate no more audio contexts are available. this will end the # task cleanly after all contexts have been processed. await self._contexts_queue.put(None) - await self.wait_for_task(self._audio_context_task) + await self._audio_context_task self._audio_context_task = None async def cancel(self, frame: CancelFrame): diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 0a20e07b3..432908151 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -435,9 +435,9 @@ class BaseOutputTransport(FrameProcessor): # also need to wait for these tasks before cancelling the video task # because it might be still rendering. if self._audio_task: - await self._transport.wait_for_task(self._audio_task) + await self._audio_task if self._clock_task: - await self._transport.wait_for_task(self._clock_task) + await self._clock_task # Stop audio mixer. if self._mixer: diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index ffeee5338..8e73fb47e 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -154,7 +154,7 @@ class WebsocketServerInputTransport(BaseInputTransport): await self.cancel_task(self._monitor_task) self._monitor_task = None if self._server_task: - await self.wait_for_task(self._server_task) + await self._server_task self._server_task = None async def cancel(self, frame: CancelFrame): diff --git a/src/pipecat/utils/asyncio/task_manager.py b/src/pipecat/utils/asyncio/task_manager.py index 386826bcb..78c2ff0f1 100644 --- a/src/pipecat/utils/asyncio/task_manager.py +++ b/src/pipecat/utils/asyncio/task_manager.py @@ -69,21 +69,6 @@ class BaseTaskManager(ABC): """ pass - @abstractmethod - async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None): - """Wait for an asyncio.Task to complete with optional timeout handling. - - This function awaits the specified asyncio.Task and handles scenarios for - timeouts, cancellations, and other exceptions. It also ensures that the task - is removed from the set of registered tasks upon completion or failure. - - Args: - task: The asyncio Task to wait for. - timeout: The maximum number of seconds to wait for the task to complete. - If None, waits indefinitely. - """ - pass - @abstractmethod async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None): """Cancels the given asyncio Task and awaits its completion with an optional timeout. @@ -189,35 +174,6 @@ class TaskManager(BaseTaskManager): logger.trace(f"{name}: task created") return task - async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None): - """Wait for an asyncio.Task to complete with optional timeout handling. - - This function awaits the specified asyncio.Task and handles scenarios for - timeouts, cancellations, and other exceptions. It also ensures that the task - is removed from the set of registered tasks upon completion or failure. - - Args: - task: The asyncio Task to wait for. - timeout: The maximum number of seconds to wait for the task to complete. - If None, waits indefinitely. - """ - name = task.get_name() - try: - if timeout: - await asyncio.wait_for(task, timeout=timeout) - else: - await task - except asyncio.TimeoutError: - logger.warning(f"{name}: timed out waiting for task to finish") - except asyncio.CancelledError: - logger.trace(f"{name}: unexpected task cancellation (maybe Ctrl-C?)") - raise - except Exception as e: - logger.exception(f"{name}: unexpected exception while stopping task: {e}") - except BaseException as e: - logger.critical(f"{name}: fatal base exception while stopping task: {e}") - raise - async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None): """Cancels the given asyncio Task and awaits its completion with an optional timeout.