FrameProcessor: fix system frame higher priorty and use a PriortyQueue

CHANGELOG.md

@@ -18,6 +18,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Updated `pyproject.toml` to once again pin `numba` to `>=0.61.2` in order to
   resolve package versioning issues.
 
+### Fixed
+
+- Fixed an issue that would cause system frames to not be processed with higher
+  priority than other frames. This could cause slower interruption times.
+
 ### Other
 
 - Improving the latency of the `HeyGenVideoService`.

src/pipecat/processors/frame_processor.py

@@ -14,7 +14,7 @@ management, and frame flow control mechanisms.
 import asyncio
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, Awaitable, Callable, Coroutine, List, Optional, Sequence
+from typing import Any, Awaitable, Callable, Coroutine, List, Optional, Sequence, Tuple
 
 from loguru import logger
 
@@ -38,6 +38,7 @@ from pipecat.observers.base_observer import BaseObserver, FramePushed
 from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.asyncio.watchdog_event import WatchdogEvent
+from pipecat.utils.asyncio.watchdog_priority_queue import WatchdogPriorityQueue
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 from pipecat.utils.base_object import BaseObject
 
@@ -54,6 +55,9 @@ class FrameDirection(Enum):
     UPSTREAM = 2
 
 
+FrameCallback = Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]]
+
+
 @dataclass
 class FrameProcessorSetup:
     """Configuration parameters for frame processor initialization.
@@ -71,7 +75,7 @@ class FrameProcessorSetup:
     watchdog_timers_enabled: bool = False
 
 
-class FrameProcessorQueue(WatchdogQueue):
+class FrameProcessorQueue(WatchdogPriorityQueue):
     """A priority queue for systems frames and other frames.
 
     This is a specialized queue for frame processors that separates and
@@ -88,6 +92,9 @@ class FrameProcessorQueue(WatchdogQueue):
 
     """
 
+    HIGH_PRIORITY = 1
+    LOW_PRIORITY = 2
+
     def __init__(self, manager: BaseTaskManager):
         """Initialize the FrameProcessorQueue.
 
@@ -96,11 +103,10 @@ class FrameProcessorQueue(WatchdogQueue):
 
         """
         super().__init__(manager)
-        self.__event = WatchdogEvent(manager)
+        self.__counter = 0
-        self.__main_queue = WatchdogQueue(manager)
+        self.__system_counter = 0
-        self.__system_queue = WatchdogQueue(manager)
 
-    async def put(self, item: Any):
+    async def put(self, item: Tuple[Frame, FrameDirection, FrameCallback]):
         """Put an item into the appropriate queue.
 
         System frames (`SystemFrame`) are placed into the system queue and all others
@@ -110,11 +116,13 @@ class FrameProcessorQueue(WatchdogQueue):
             item (Any): The item to enqueue.
 
         """
-        if isinstance(item, SystemFrame):
+        frame, _, _ = item
-            await self.__system_queue.put(item)
+        if isinstance(frame, SystemFrame):
+            await super().put((self.HIGH_PRIORITY, self.__system_counter, item))
+            self.__system_counter += 1
         else:
-            await self.__main_queue.put(item)
+            await super().put((self.LOW_PRIORITY, self.__counter, item))
-        self.__event.set()
+            self.__counter += 1
 
     async def get(self) -> Any:
         """Retrieve the next item from the queue.
@@ -126,38 +134,9 @@ class FrameProcessorQueue(WatchdogQueue):
             Any: The next item from the system or main queue.
 
         """
-        # Wait for an item in any of the queues if they are empty.
+        _, _, item = await super().get()
-        if self.__main_queue.empty() and self.__system_queue.empty():
-            await self.__event.wait()
-
-        # Prioritize system frames.
-        if self.__system_queue.qsize() > 0:
-            item = await self.__system_queue.get()
-            self.__system_queue.task_done()
-        else:
-            item = await self.__main_queue.get()
-            self.__main_queue.task_done()
-
-        # Clear the event only if all queues are empty.
-        if self.__main_queue.empty() and self.__system_queue.empty():
-            self.__event.clear()
-
         return item
 
-    def cancel(self):
-        """Cancel both internal queues.
-
-        This method is used to stop processing and release any pending tasks
-        in both the system and main queues. Typically used during shutdown
-        or cleanup to prevent further processing of frames.
-
-        """
-        self.__main_queue.cancel()
-        self.__system_queue.cancel()
-
-
-FrameCallback = Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]]
-
 
 class FrameProcessor(BaseObject):
     """Base class for all frame processors in the pipeline.
@@ -254,9 +233,7 @@ class FrameProcessor(BaseObject):
         # called. To resume processing frames we need to call
         # `resume_processing_frames()` which will wake up the event.
         self.__should_block_frames = False
-        self.__process_event = None
         self.__process_frame_task: Optional[asyncio.Task] = None
-        self.__process_queue = None
 
     @property
     def id(self) -> int:
@@ -745,9 +722,7 @@ class FrameProcessor(BaseObject):
         """Create the non-system frame processing task."""
         if not self.__process_frame_task:
             self.__should_block_frames = False
-            if not self.__process_event:
+            self.__process_event = WatchdogEvent(self.task_manager)
-                self.__process_event = WatchdogEvent(self.task_manager)
-            self.__process_event.clear()
             self.__process_queue = WatchdogQueue(self.task_manager)
             self.__process_frame_task = self.create_task(self.__process_frame_task_handler())
 
@@ -790,10 +765,12 @@ class FrameProcessor(BaseObject):
                     f"{self}: __process_queue is None when processing frame {frame.name}"
                 )
 
+            self.__input_queue.task_done()
+
     async def __process_frame_task_handler(self):
         """Handle non-system frames from the process queue."""
         while True:
-            if self.__should_block_frames and self.__process_event:
+            if self.__should_block_frames:
                 logger.trace(f"{self}: frame processing paused")
                 await self.__process_event.wait()
                 self.__process_event.clear()
@@ -803,3 +780,5 @@ class FrameProcessor(BaseObject):
             (frame, direction, callback) = await self.__process_queue.get()
 
             await self.__process_frame(frame, direction, callback)
+
+            self.__process_queue.task_done()