From e0a6c6871c75e4ff464b7f30773344f19dd6fe4e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 9 Dec 2024 09:35:18 -0800 Subject: [PATCH 1/2] parallel_pipeline: don't queue system frames --- CHANGELOG.md | 2 ++ src/pipecat/pipeline/parallel_pipeline.py | 16 +++++++++++++--- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 51eb668ac..9df14cc99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -70,6 +70,8 @@ async def on_audio_data(processor, audio, sample_rate, num_channels): ### Fixed +- Fixed a `ParallelPipeline` issue that would cause system frames to be queued. + - Fixed `FastAPIWebsocketTransport` so it can work with binary data (e.g. using the protobuf serializer). diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index ad3ebd8dd..1ccd14a98 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -12,7 +12,7 @@ from typing import List from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame +from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame, SystemFrame from loguru import logger @@ -27,7 +27,12 @@ class Source(FrameProcessor): match direction: case FrameDirection.UPSTREAM: - await self._up_queue.put(frame) + # 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: + await self._up_queue.put(frame) case FrameDirection.DOWNSTREAM: await self.push_frame(frame, direction) @@ -44,7 +49,12 @@ class Sink(FrameProcessor): case FrameDirection.UPSTREAM: await self.push_frame(frame, direction) case FrameDirection.DOWNSTREAM: - await self._down_queue.put(frame) + # 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: + await self._down_queue.put(frame) class ParallelPipeline(BasePipeline): From e690c9823006344fb02c8ea3f75140b67867c182 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 9 Dec 2024 09:36:48 -0800 Subject: [PATCH 2/2] transports(daily): no need for joining flag This was put back because of an issue in ParallelPipeline but that issue is now fixed so the joining check is not really necessary. --- src/pipecat/transports/services/daily.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 070fe3092..d2fe06b27 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -200,7 +200,6 @@ class DailyTransportClient(EventHandler): self._other_participant_has_joined = False self._joined = False - self._joining = False self._leave_counter = 0 self._executor = ThreadPoolExecutor(max_workers=5) @@ -315,12 +314,11 @@ class DailyTransportClient(EventHandler): async def join(self): # Transport already joined, ignore. - if self._joined or self._joining: + if self._joined: # Increment leave counter if we already joined. self._leave_counter += 1 return - self._joining = True logger.info(f"Joining {self._room_url}") # For performance reasons, never subscribe to video streams (unless a @@ -353,7 +351,6 @@ class DailyTransportClient(EventHandler): error_msg = f"Time out joining {self._room_url}" logger.error(error_msg) await self._callbacks.on_error(error_msg) - self._joining = False async def _start_transcription(self): if not self._token: