Fix ParallelPipeline flush ordering and buffered frame handling
Flush buffered frames before pushing the synchronization frame so downstream processors see the buffered frames first. Switch to a while-loop with pop(0) so frames added to the buffer during flush are also drained.
This commit is contained in:
@@ -143,6 +143,19 @@ class ParallelPipeline(BasePipeline):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
# Parallel pipeline synchronized frames.
|
||||
#
|
||||
# - StartFrame: If a fast branch completes first, processors in
|
||||
# other branches that haven't received StartFrame yet could
|
||||
# receive other frames before it, causing errors.
|
||||
#
|
||||
# - EndFrame: If EndFrame escapes from a fast branch, downstream
|
||||
# processors (e.g. output transport) begin shutting down while
|
||||
# other branches still have frames to flush, causing lost output.
|
||||
#
|
||||
# - CancelFrame: PipelineTask waits for CancelFrame to reach the
|
||||
# pipeline sink. If it escapes from a fast branch while slower
|
||||
# branches are still running, the task considers cancellation
|
||||
# complete prematurely.
|
||||
if isinstance(frame, (StartFrame, EndFrame, CancelFrame)):
|
||||
self._frame_counter[frame.id] = len(self._pipelines)
|
||||
self._synchronizing = True
|
||||
@@ -179,8 +192,13 @@ class ParallelPipeline(BasePipeline):
|
||||
# Only push the frame when all pipelines have processed it.
|
||||
if frame_counter == 0:
|
||||
self._synchronizing = False
|
||||
await self._parallel_push_frame(frame, direction)
|
||||
await self._flush_buffered_frames()
|
||||
# StartFrame should always go before any other frame.
|
||||
if isinstance(frame, StartFrame):
|
||||
await self._parallel_push_frame(frame, direction)
|
||||
await self._flush_buffered_frames()
|
||||
else:
|
||||
await self._flush_buffered_frames()
|
||||
await self._parallel_push_frame(frame, direction)
|
||||
await self.resume_processing_system_frames()
|
||||
await self.resume_processing_frames()
|
||||
else:
|
||||
@@ -188,7 +206,6 @@ class ParallelPipeline(BasePipeline):
|
||||
|
||||
async def _flush_buffered_frames(self):
|
||||
"""Flush frames that were buffered during lifecycle frame synchronization."""
|
||||
frames = self._buffered_frames
|
||||
self._buffered_frames = []
|
||||
for frame, direction in frames:
|
||||
while len(self._buffered_frames) > 0:
|
||||
frame, direction = self._buffered_frames.pop(0)
|
||||
await self.push_frame(frame, direction)
|
||||
|
||||
Reference in New Issue
Block a user