diff --git a/changelog/3789.fixed.md b/changelog/3789.fixed.md new file mode 100644 index 000000000..1bf2be1a3 --- /dev/null +++ b/changelog/3789.fixed.md @@ -0,0 +1 @@ +- Fixed `push_interruption_task_frame_and_wait()` hanging indefinitely when the `InterruptionFrame` does not reach the pipeline sink within the timeout. Added a `timeout` keyword argument to customize the wait duration. diff --git a/examples/foundational/07s-interruptible-google-audio-in.py b/examples/foundational/07s-interruptible-google-audio-in.py index 707db107e..1de374a3f 100644 --- a/examples/foundational/07s-interruptible-google-audio-in.py +++ b/examples/foundational/07s-interruptible-google-audio-in.py @@ -96,7 +96,7 @@ class UserAudioCollector(FrameProcessor): self._user_speaking = True elif isinstance(frame, UserStoppedSpeakingFrame): self._user_speaking = False - self._context.add_audio_frames_message(audio_frames=self._audio_frames) + await self._context.add_audio_frames_message(audio_frames=self._audio_frames) await self._user_context_aggregator.push_frame(LLMRunFrame()) elif isinstance(frame, InputAudioRawFrame): diff --git a/examples/foundational/25-google-audio-in.py b/examples/foundational/25-google-audio-in.py index 002aeaa1c..40903e1d5 100644 --- a/examples/foundational/25-google-audio-in.py +++ b/examples/foundational/25-google-audio-in.py @@ -98,7 +98,7 @@ class UserAudioCollector(FrameProcessor): self._user_speaking = True elif isinstance(frame, UserStoppedSpeakingFrame): self._user_speaking = False - self._context.add_audio_frames_message(audio_frames=self._audio_frames) + await self._context.add_audio_frames_message(audio_frames=self._audio_frames) await self._user_context_aggregator.push_frame(LLMContextFrame(context=self._context)) elif isinstance(frame, InputAudioRawFrame): if self._user_speaking: diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 4cf32c800..bcdb2d57b 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -52,8 +52,6 @@ from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMet from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.base_object import BaseObject -INTERRUPTION_COMPLETION_TIMEOUT = 2.0 - class FrameDirection(Enum): """Direction of frame flow in the processing pipeline. @@ -763,7 +761,7 @@ class FrameProcessor(BaseObject): await self._call_event_handler("on_after_push_frame", frame) - async def push_interruption_task_frame_and_wait(self): + async def push_interruption_task_frame_and_wait(self, *, timeout: float = 5.0): """Push an interruption task frame upstream and wait for the interruption. This function sends an `InterruptionTaskFrame` upstream to the @@ -772,9 +770,11 @@ class FrameProcessor(BaseObject): attached to both frames so the caller can wait until the interruption has fully traversed the pipeline. The event is set when the `InterruptionFrame` reaches the pipeline sink. If the frame does - not complete within `INTERRUPTION_COMPLETION_TIMEOUT` seconds, a - warning is logged periodically until it completes. + not complete within the given timeout, a warning is logged and the + event is forcibly set so the caller is unblocked. + Args: + timeout: Maximum seconds to wait for the interruption to complete. """ self._wait_for_interruption = True @@ -782,19 +782,20 @@ class FrameProcessor(BaseObject): await self.push_frame(InterruptionTaskFrame(event=event), FrameDirection.UPSTREAM) - # Wait for the `InterruptionFrame` to complete and log a warning - # periodically if it takes too long. + # Wait for the `InterruptionFrame` to complete and log a warning if it + # takes too long. If it does take too long make sure we unblock it, + # otherwise we will hang here forever. while not event.is_set(): try: - await asyncio.wait_for(event.wait(), timeout=INTERRUPTION_COMPLETION_TIMEOUT) + await asyncio.wait_for(event.wait(), timeout=timeout) except asyncio.TimeoutError: logger.warning( f"{self}: InterruptionFrame has not completed after" - f" {INTERRUPTION_COMPLETION_TIMEOUT}s. Make sure" - " InterruptionFrame.complete() is being called (e.g. if the" - " frame is being blocked or consumed before reaching the" - " pipeline sink)." + f" {timeout}s. Make sure InterruptionFrame.complete()" + " is being called (e.g. if the frame is being blocked" + " or consumed before reaching the pipeline sink)." ) + event.set() self._wait_for_interruption = False diff --git a/tests/test_frame_processor.py b/tests/test_frame_processor.py index 2ce4b7880..138c8e6d8 100644 --- a/tests/test_frame_processor.py +++ b/tests/test_frame_processor.py @@ -25,7 +25,6 @@ from pipecat.frames.frames import ( from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.filters.identity_filter import IdentityFilter from pipecat.processors.frame_processor import ( - INTERRUPTION_COMPLETION_TIMEOUT, FrameDirection, FrameProcessor, ) @@ -521,7 +520,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): # Complete after the timeout so the warning fires # but the test doesn't hang. async def delayed_complete(): - await asyncio.sleep(INTERRUPTION_COMPLETION_TIMEOUT + 1.0) + await asyncio.sleep(1.0) frame.complete() asyncio.create_task(delayed_complete()) @@ -532,7 +531,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, TextFrame): - await self.push_interruption_task_frame_and_wait() + await self.push_interruption_task_frame_and_wait(timeout=0.5) await self.push_frame(OutputTransportMessageUrgentFrame(message="done")) else: await self.push_frame(frame, direction)