From 0a9daa2f56c9639c5b5aab3db21903364282c876 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 24 Jan 2025 19:29:59 -0800 Subject: [PATCH] task: avoid canceling tasks more than once --- src/pipecat/pipeline/task.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9149a41a2..8d659bb90 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -155,7 +155,8 @@ class PipelineTask(BaseTask): # 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()) - await self._cancel_tasks(True) + # Only cancel the push task. Everything else will be cancelled in run(). + await cancel_task(self._process_push_task) await self._cleanup() async def run(self): @@ -171,7 +172,7 @@ class PipelineTask(BaseTask): # well, because you get a CancelledError in every place you are # awaiting a task. pass - await self._cancel_tasks(False) + await self._cancel_tasks() self._finished = True async def queue_frame(self, frame: Frame): @@ -215,12 +216,9 @@ class PipelineTask(BaseTask): loop, self._heartbeat_monitor_handler(), f"{self}::_heartbeat_monitor_handler" ) - async def _cancel_tasks(self, cancel_push: bool): + async def _cancel_tasks(self): await self._maybe_cancel_heartbeat_tasks() - if cancel_push: - await cancel_task(self._process_push_task) - await cancel_task(self._process_up_task) await cancel_task(self._process_down_task)