Collapse _pipeline_finished_event into BaseTask._finished_event

PipelineTask had its own ``_pipeline_finished_event`` that signalled
"pipeline run has truly finished" — the same role ``BaseTask._finished_event``
plays for bus-only tasks. They were two events with the same intent.

Set ``_finished_event`` directly when the pipeline-end frame propagates
through the sink, drop the now-redundant field, and drop the
``clear()`` after wait so the event stays set for the lifetime of the
task. As a side-effect, ``await pipeline_task.wait()`` from outside now
resolves at the moment the pipeline finishes, matching the semantics
of bus-only tasks.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-14 22:53:10 -07:00
parent de1bd7cb7e
commit 4d9e258e55

View File

@@ -396,9 +396,6 @@ class PipelineTask(BaseTask):
# StopFrame) has been received at the end of the pipeline.
self._pipeline_end_event = asyncio.Event()
# This event is set when the pipeline truly finishes.
self._pipeline_finished_event = asyncio.Event()
# When bridged, wrap the user pipeline with bus edge processors
# so frames tee onto the bus at the source/sink and incoming bus
# frames are injected back into the local pipeline. The edges
@@ -803,12 +800,12 @@ class PipelineTask(BaseTask):
self._pipeline_end_event.clear()
# We are really done.
self._pipeline_finished_event.set()
# We are really done. Setting ``_finished_event`` makes
# ``BaseTask.wait()`` resolve for callers awaiting this task.
self._finished_event.set()
async def _wait_for_pipeline_finished(self):
await self._pipeline_finished_event.wait()
self._pipeline_finished_event.clear()
await self._finished_event.wait()
# Make sure we wait for the main task to complete.
if self._process_push_task:
await self._process_push_task