From 9cc498b1fa17e2571092084d5dc6d3762fbe0bfd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 5 May 2025 21:27:49 -0700 Subject: [PATCH] TaskManager: use a dictionary instead of a set to store tasks --- CHANGELOG.md | 2 ++ src/pipecat/utils/asyncio.py | 15 ++++++++------- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 95350157f..ac4e2db11 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `TaskManager` that was causing dangling tasks to be reported. + - Fixed an issue that could cause data to be sent to the transports when they were still not ready. diff --git a/src/pipecat/utils/asyncio.py b/src/pipecat/utils/asyncio.py index acc4acec8..cea447329 100644 --- a/src/pipecat/utils/asyncio.py +++ b/src/pipecat/utils/asyncio.py @@ -6,7 +6,7 @@ import asyncio from abc import ABC, abstractmethod -from typing import Coroutine, Optional, Set +from typing import Coroutine, Dict, Optional, Sequence, Set from loguru import logger @@ -69,14 +69,14 @@ class BaseTaskManager(ABC): pass @abstractmethod - def current_tasks(self) -> Set[asyncio.Task]: + def current_tasks(self) -> Sequence[asyncio.Task]: """Returns the list of currently created/registered tasks.""" pass class TaskManager(BaseTaskManager): def __init__(self) -> None: - self._tasks: Set[asyncio.Task] = set() + self._tasks: Dict[str, asyncio.Task] = {} self._loop: Optional[asyncio.AbstractEventLoop] = None def set_event_loop(self, loop: asyncio.AbstractEventLoop): @@ -179,16 +179,17 @@ class TaskManager(BaseTaskManager): finally: self._remove_task(task) - def current_tasks(self) -> Set[asyncio.Task]: + def current_tasks(self) -> Sequence[asyncio.Task]: """Returns the list of currently created/registered tasks.""" - return self._tasks + return list(self._tasks.values()) def _add_task(self, task: asyncio.Task): - self._tasks.add(task) + name = task.get_name() + self._tasks[name] = task def _remove_task(self, task: asyncio.Task): name = task.get_name() try: - self._tasks.remove(task) + del self._tasks[name] except KeyError as e: logger.trace(f"{name}: unable to remove task (already removed?): {e}")