From 80a7f1b1e7c346e6e44311684d5ddd9ae493677c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 24 Jan 2025 18:08:44 -0800 Subject: [PATCH] runner: improve signal handler task cancellation --- src/pipecat/pipeline/runner.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 34f7a0a8c..62e9e5e6f 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -19,6 +19,7 @@ class PipelineRunner: self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}" self._tasks = {} + self._sig_task = None if handle_sigint: self._setup_sigint() @@ -28,6 +29,10 @@ class PipelineRunner: self._tasks[task.name] = task await task.run() del self._tasks[task.name] + # If we are cancelling through a signal, make sure we wait for it so + # everything gets cleaned up nicely. + if self._sig_task: + await self._sig_task logger.debug(f"Runner {self} finished running {task}") async def stop_when_done(self): @@ -40,14 +45,14 @@ class PipelineRunner: def _setup_sigint(self): loop = asyncio.get_running_loop() - loop.add_signal_handler( - signal.SIGINT, lambda *args: asyncio.create_task(self._sig_handler()) - ) - loop.add_signal_handler( - signal.SIGTERM, lambda *args: asyncio.create_task(self._sig_handler()) - ) + loop.add_signal_handler(signal.SIGINT, lambda *args: self._sig_handler()) + loop.add_signal_handler(signal.SIGTERM, lambda *args: self._sig_handler()) - async def _sig_handler(self): + def _sig_handler(self): + if not self._sig_task: + self._sig_task = asyncio.create_task(self._sig_cancel()) + + async def _sig_cancel(self): logger.warning(f"Interruption detected. Canceling runner {self}") await self.cancel()