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

runner: warn about remaining dangling tasks

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-01-24 18:13:18 -08:00
parent 80a7f1b1e7
commit c2d8a45a07
1 changed files with 7 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/pipecat/pipeline/runner.py
View File

@@ -10,7 +10,7 @@ import signal
from loguru import logger from loguru import logger
from pipecat.pipeline.task import PipelineTask from pipecat.pipeline.task import PipelineTask
from pipecat.utils.utils import obj_count, obj_id from pipecat.utils.utils import current_tasks, obj_count, obj_id
class PipelineRunner: class PipelineRunner:
@@ -33,6 +33,7 @@ class PipelineRunner:
# everything gets cleaned up nicely. # everything gets cleaned up nicely.
if self._sig_task: if self._sig_task:
await self._sig_task await self._sig_task
self._print_dangling_tasks()
logger.debug(f"Runner {self} finished running {task}") logger.debug(f"Runner {self} finished running {task}")
async def stop_when_done(self): async def stop_when_done(self):
@@ -56,5 +57,10 @@ class PipelineRunner:
logger.warning(f"Interruption detected. Canceling runner {self}") logger.warning(f"Interruption detected. Canceling runner {self}")
await self.cancel() await self.cancel()
def _print_dangling_tasks(self):
tasks = [t.get_name() for t in current_tasks()]
if tasks:
logger.warning(f"Dangling tasks detected: {tasks}")
def __str__(self): def __str__(self):
return self.name return self.name