From 4eed335bc7f2538ae0d43783d55c72a237a73101 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sat, 24 May 2025 21:01:23 -0700 Subject: [PATCH] PipelineTask: check if pipeline has already been cancelled --- CHANGELOG.md | 2 ++ src/pipecat/pipeline/task.py | 17 ++++++++++------- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f7db1e97d..2830b1240 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -89,6 +89,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Check if `PipelineTask` has already been cancelled. + - Don't raise an exception if event handler is not registered. - Upgraded `deepgram-sdk` to 4.1.0. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 692f325d9..0fe330655 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -236,6 +236,7 @@ class PipelineTask(BaseTask): ) observers.append(self._turn_trace_observer) self._finished = False + self._cancelled = False # This queue receives frames coming from the pipeline upstream. self._up_queue = asyncio.Queue() @@ -346,7 +347,6 @@ class PipelineTask(BaseTask): async def cancel(self): """Stops the running pipeline immediately.""" - logger.debug(f"Canceling pipeline task {self}") await self._cancel() async def run(self): @@ -406,12 +406,15 @@ class PipelineTask(BaseTask): await self.queue_frame(frame) async def _cancel(self): - # Make sure everything is cleaned up downstream. This is sent - # out-of-band from the main streaming task which is what we want since - # we want to cancel right away. - await self._source.push_frame(CancelFrame()) - # Only cancel the push task. Everything else will be cancelled in run(). - await self._task_manager.cancel_task(self._process_push_task) + if not self._cancelled: + logger.debug(f"Canceling pipeline task {self}") + self._cancelled = True + # Make sure everything is cleaned up downstream. This is sent + # out-of-band from the main streaming task which is what we want since + # we want to cancel right away. + await self._source.push_frame(CancelFrame()) + # Only cancel the push task. Everything else will be cancelled in run(). + await self._task_manager.cancel_task(self._process_push_task) async def _create_tasks(self): self._process_up_task = self._task_manager.create_task(