fix issues with Ctrl-C tasks cancellation

CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Added initial interruptions support.
 
+### Fixed
+
+- Fixed issues with Ctrl-C program termination.
+
 ## [0.0.16] - 2024-05-16
 
 ### Fixed

src/pipecat/pipeline/task.py

@@ -38,7 +38,7 @@ class PipelineTask:
         self._pipeline = pipeline
         self._allow_interruptions = allow_interruptions
 
-        self._task_queue = asyncio.Queue()
+        self._down_queue = asyncio.Queue()
         self._up_queue = asyncio.Queue()
 
         self._source = Source(self._up_queue)
@@ -50,15 +50,22 @@ class PipelineTask:
 
     async def cancel(self):
         logger.debug(f"Canceling pipeline task {self}")
-        await self.queue_frame(CancelFrame())
+        # Make sure everything is cleaned up downstream. This is sent
+        # out-of-band from the main streaming task which is what we want since
+        # we want to cancel right away.
+        await self._source.process_frame(CancelFrame(), FrameDirection.DOWNSTREAM)
+        self._process_down_task.cancel()
+        self._process_up_task.cancel()
 
     async def run(self):
-        await asyncio.gather(self._process_task_queue(), self._process_up_queue())
+        self._process_up_task = asyncio.create_task(self._process_up_queue())
+        self._process_down_task = asyncio.create_task(self._process_down_queue())
+        await asyncio.gather(self._process_up_task, self._process_down_task)
         await self._source.cleanup()
         await self._pipeline.cleanup()
 
     async def queue_frame(self, frame: Frame):
-        await self._task_queue.put(frame)
+        await self._down_queue.put(frame)
 
     async def queue_frames(self, frames: Iterable[Frame] | AsyncIterable[Frame]):
         if isinstance(frames, AsyncIterable):
@@ -70,30 +77,31 @@ class PipelineTask:
         else:
             raise Exception("Frames must be an iterable or async iterable")
 
-    async def _process_task_queue(self):
+    async def _process_down_queue(self):
         await self._source.process_frame(
             StartFrame(allow_interruptions=self._allow_interruptions), FrameDirection.DOWNSTREAM)
         running = True
         while running:
-            frame = await self._task_queue.get()
+            try:
-            await self._source.process_frame(frame, FrameDirection.DOWNSTREAM)
+                frame = await self._down_queue.get()
-            self._task_queue.task_done()
+                await self._source.process_frame(frame, FrameDirection.DOWNSTREAM)
-            running = not (isinstance(frame, StopTaskFrame) or
+                self._down_queue.task_done()
-                           isinstance(frame, CancelFrame) or
+                running = not (isinstance(frame, StopTaskFrame) or isinstance(frame, EndFrame))
-                           isinstance(frame, EndFrame))
+            except asyncio.CancelledError:
-        # We just enqueue None to terminate the task.
+                break
-        await self._up_queue.put(None)
+        # We just enqueue None to terminate the task gracefully.
+        self._process_up_task.cancel()
 
     async def _process_up_queue(self):
-        running = True
+        while True:
-        while running:
+            try:
-            frame = await self._up_queue.get()
+                frame = await self._up_queue.get()
-            if frame:
                 if isinstance(frame, ErrorFrame):
                     logger.error(f"Error running app: {frame.error}")
                     await self.queue_frame(CancelFrame())
-            self._up_queue.task_done()
+                self._up_queue.task_done()
-            running = frame is not None
+            except asyncio.CancelledError:
+                break
 
     def __str__(self):
         return self.name

src/pipecat/transports/base_input.py

@@ -65,8 +65,7 @@ class BaseInputTransport(FrameProcessor):
             await self._audio_in_thread
             await self._audio_out_thread
 
-        await self._internal_push_frame(None, None)
+        self._push_frame_task.cancel()
-        await self._push_frame_task
 
     def vad_analyze(self, audio_frames: bytes) -> VADState:
         pass
@@ -112,12 +111,12 @@ class BaseInputTransport(FrameProcessor):
         await self._push_queue.put((frame, direction))
 
     async def _push_frame_task_handler(self):
-        running = True
+        while True:
-        while running:
+            try:
-            (frame, direction) = await self._push_queue.get()
+                (frame, direction) = await self._push_queue.get()
-            if frame:
                 await self.push_frame(frame, direction)
-            running = frame is not None
+            except asyncio.CancelledError:
+                break
 
     #
     # Handle interruptions