Difference Between EndFrame and EndTaskFrame

[golbin]

Description

This issue seeks clarification regarding the usage of EndFrame and EndTaskFrame.

Issue description

• EndFrame: Used for each Processor in a Pipeline.
• EndTaskFrame: Used for the entire Pipeline task.

If this understanding is correct, it seems that the following lines in the bot examples should use EndTaskFrame instead of EndFrame.

Please let me know if I am missing something or if there’s any additional context to consider.

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await task.queue_frames([EndFrame()])

This clarification will help ensure consistent and accurate usage in the examples. Thank you!

[markbackman]

Hey @golbin, I find myself referring to the Frame definitions for this.

@dataclass
class EndFrame(ControlFrame):
    """Indicates that a pipeline has ended and frame processors and pipelines
    should be shut down. If the transport receives this frame, it will stop
    sending frames to its output channel(s) and close all its threads. Note,
    that this is a control frame, which means it will received in the order it
    was sent (unline system frames).

    """

    pass

@dataclass
class EndTaskFrame(SystemFrame):
    """This is used to notify the pipeline task that the pipeline should be
    closed nicely (flushing all the queued frames) by pushing an EndFrame
    downstream.

    """

    pass

When an EndTaskFrame is received by the processor, it generates an EndFrame. The EndFrame then propagates through the pipeline, causing cleanup.

So, essentially, EndTaskFrame is the signal to initiate shutdown, while EndFrame is the actual mechanism that performs the shutdown by propagating through the pipeline.

A few additional difference to point out:

1. Processing order:

• EndFrame being a ControlFrame means it's processed in order with other frames
• EndTaskFrame being a SystemFrame means it can be processed out of order with other frames

2. Pipeline direction:

• EndFrame flows downstream through the pipeline and causes processors and pipelines to shut down
• EndTaskFrame flows upstream to notify the pipeline task that it should initiate a graceful shutdown

Does that help?

[golbin]

Hi @markbackman,

Thank you so much for your kind and detailed response.

To clarify, if I want to finish processing within a processor, I can use EndTaskFrame like this:

async def process_frame(self, frame: Frame, direction: FrameDirection):
    await super().process_frame(frame, direction)

    if ~~~:
        await self.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)

On the other hand, if I want to finish outside the pipeline, I can use EndFrame like this:

main_task.queue_frame(EndFrame())

And if I want to finish immediately, I can also use EndTaskFrame.

Is my understanding correct?

[markbackman]

Yes, that makes sense. Within the processor, I think you can also await self.push_frame(EndFrame()). Both are effective.

There might be a nuance between these two that @aconchillo can elaborate on when he returns.

[golbin]

🫡