parallel_pipeline: fix system frames again

src/pipecat/pipeline/parallel_pipeline.py

@@ -27,11 +27,8 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
 
         match direction:
             case FrameDirection.UPSTREAM:
-                # We don't want to queue system frames as they would be
-                # processed by a separate task.
-                if isinstance(frame, SystemFrame):
-                    await self.push_frame(frame, direction)
-                else:
+                # SystemFrames are pushed directly from ParallelPipeline.
+                if not isinstance(frame, SystemFrame):
                     await self._up_queue.put(frame)
             case FrameDirection.DOWNSTREAM:
                 await self.push_frame(frame, direction)
@@ -49,11 +46,8 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
             case FrameDirection.UPSTREAM:
                 await self.push_frame(frame, direction)
             case FrameDirection.DOWNSTREAM:
-                # We don't want to queue system frames as they would be
-                # processed by a separate task.
-                if isinstance(frame, SystemFrame):
-                    await self.push_frame(frame, direction)
-                else:
+                # SystemFrames are pushed directly from ParallelPipeline.
+                if not isinstance(frame, SystemFrame):
                     await self._down_queue.put(frame)
 
 
@@ -123,11 +117,13 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
             await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks])
         elif direction == FrameDirection.DOWNSTREAM:
             # If we get a downstream frame we process it in each source.
-            # TODO(aleix): We are creating task for each frame. For real-time
-            # video/audio this might be too slow. We should use an already
-            # created task instead.
             await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sources])
 
+        # If we have a SystemFrame we will push it from this task. Note that the
+        # connected sinks and sources ignore SystemFrames.
+        if isinstance(frame, SystemFrame):
+            await self.push_frame(frame, direction)
+
         # If we get an EndFrame we stop our queue processing tasks and wait on
         # all the pipelines to finish.
         if isinstance(frame, (CancelFrame, EndFrame)):