wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task: avoid canceling tasks more than once

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-01-24 19:29:59 -08:00
parent e48c0e52ef
commit 0a9daa2f56
1 changed files with 4 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/pipecat/pipeline/task.py
View File

@@ -155,7 +155,8 @@ class PipelineTask(BaseTask):
# out-of-band from the main streaming task which is what we want since # out-of-band from the main streaming task which is what we want since
# we want to cancel right away. # we want to cancel right away.
await self._source.push_frame(CancelFrame()) 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() await self._cleanup()
async def run(self): async def run(self):
@@ -171,7 +172,7 @@ class PipelineTask(BaseTask):
# well, because you get a CancelledError in every place you are # well, because you get a CancelledError in every place you are
# awaiting a task. # awaiting a task.
pass pass
await self._cancel_tasks(False) await self._cancel_tasks()
self._finished = True self._finished = True
async def queue_frame(self, frame: Frame): async def queue_frame(self, frame: Frame):
@@ -215,12 +216,9 @@ class PipelineTask(BaseTask):
loop, self._heartbeat_monitor_handler(), f"{self}::_heartbeat_monitor_handler" 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() 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_up_task)
await cancel_task(self._process_down_task) await cancel_task(self._process_down_task)