Stream lookahead: concurrent remote/API step execution for video workflows (10-12x FPS, outputs identical to sequential)

development/benchmark_scripts/benchmark_remote_stream_pipeline_overhead.py

@@ -0,0 +1,255 @@
+"""Measure the scheduler-side overhead of stream-lookahead execution.
+
+All remote calls are mocked to return instantly, so the numbers isolate the
+execution engine's own costs — workflow compilation, frontier computation,
+and the per-frame difference between the classic single-pass run and the
+lookahead two-pass (deferred run + resume) — with zero network time.
+
+Example:
+    python development/benchmark_scripts/benchmark_lookahead_overhead.py \
+        --frames 200 --compiles 30 --output /tmp/lookahead_overhead.json
+"""
+
+import argparse
+import json
+import os
+import statistics
+import time
+from datetime import datetime
+from typing import Any, Callable, Dict, List
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+
+WORKFLOW_SPECIFICATION = {
+    "version": "1.0.0",
+    "inputs": [{"type": "WorkflowImage", "name": "image"}],
+    "steps": [
+        {
+            "type": "roboflow_core/roboflow_object_detection_model@v3",
+            "name": "model",
+            "images": "$inputs.image",
+            "model_id": "some-project/1",
+            "confidence_mode": "custom",
+            "custom_confidence": 0.35,
+        },
+        {
+            "type": "roboflow_core/byte_tracker@v3",
+            "name": "byte_tracker",
+            "image": "$inputs.image",
+            "detections": "$steps.model.predictions",
+        },
+        {
+            "type": "roboflow_core/bounding_box_visualization@v1",
+            "name": "bounding_box_visualization",
+            "image": "$inputs.image",
+            "predictions": "$steps.byte_tracker.tracked_detections",
+        },
+    ],
+    "outputs": [
+        {
+            "type": "JsonField",
+            "name": "tracked_detections",
+            "coordinates_system": "own",
+            "selector": "$steps.byte_tracker.tracked_detections",
+        },
+        {
+            "type": "JsonField",
+            "name": "visualization",
+            "coordinates_system": "own",
+            "selector": "$steps.bounding_box_visualization.image",
+        },
+    ],
+}
+
+CANNED_PREDICTION = {
+    "predictions": [
+        {
+            "x": 100.0,
+            "y": 100.0,
+            "width": 50.0,
+            "height": 50.0,
+            "confidence": 0.9,
+            "class": "car",
+            "class_id": 0,
+            "detection_id": "00000000-0000-0000-0000-000000000000",
+        }
+    ],
+    "image": {"width": 640, "height": 360},
+    "time": 0.0,
+}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Measure compilation, frontier and per-frame scheduler overhead."
+    )
+    parser.add_argument("--frames", type=int, default=200)
+    parser.add_argument("--warmup", type=int, default=20)
+    parser.add_argument("--compiles", type=int, default=30)
+    parser.add_argument("--depth", type=int, default=4)
+    parser.add_argument("--output", type=str, default=None)
+    return parser.parse_args()
+
+
+def time_repeated(task: Callable[[], Any], repeats: int) -> Dict[str, float]:
+    durations = []
+    for _ in range(repeats):
+        started_at = time.perf_counter()
+        task()
+        durations.append((time.perf_counter() - started_at) * 1000)
+    return {
+        "repeats": repeats,
+        "avg_ms": round(statistics.fmean(durations), 4),
+        "median_ms": round(statistics.median(durations), 4),
+        "min_ms": round(min(durations), 4),
+        "max_ms": round(max(durations), 4),
+    }
+
+
+def build_engine():
+    from inference.core.workflows.execution_engine.core import ExecutionEngine
+
+    return ExecutionEngine.init(
+        workflow_definition=WORKFLOW_SPECIFICATION,
+        init_parameters={
+            "workflows_core.api_key": "overhead-benchmark",
+            "workflows_core.model_manager": MagicMock(),
+            "workflows_core.step_execution_mode": _step_execution_mode(),
+        },
+    )
+
+
+def _step_execution_mode():
+    from inference.core.workflows.core_steps.common.entities import StepExecutionMode
+
+    return StepExecutionMode.REMOTE
+
+
+def make_video_frame(frame_id: int, image: np.ndarray):
+    from inference.core.interfaces.camera.entities import VideoFrame
+
+    return VideoFrame(
+        image=image,
+        frame_id=frame_id,
+        frame_timestamp=datetime.fromtimestamp(1_700_000_000 + frame_id / 30),
+        fps=30,
+        measured_fps=30,
+        source_id=0,
+        comes_from_video_file=True,
+    )
+
+
+def run_per_frame_measurement(
+    runner: Callable[[List[Any]], Any],
+    image: np.ndarray,
+    frames: int,
+    warmup: int,
+    drain: Callable[[], Any],
+) -> Dict[str, float]:
+    for frame_id in range(warmup):
+        runner([make_video_frame(frame_id, image)])
+    started_at = time.perf_counter()
+    for frame_id in range(warmup, warmup + frames):
+        runner([make_video_frame(frame_id, image)])
+    drain()
+    total_ms = (time.perf_counter() - started_at) * 1000
+    return {"frames": frames, "avg_ms_per_frame": round(total_ms / frames, 4)}
+
+
+def main() -> None:
+    args = parse_args()
+    os.environ.setdefault("WORKFLOWS_STREAM_LOOKAHEAD_DEPTH", str(args.depth))
+    # DEBUG logging renders more lines on the sequential path than the
+    # lookahead path (~0.3 ms per line) and would fabricate an overhead gap.
+    os.environ["LOG_LEVEL"] = os.environ.get("OVERHEAD_BENCHMARK_LOG_LEVEL", "WARNING")
+    from inference.core.interfaces.stream.model_handlers import (
+        workflows as workflows_module,
+    )
+    from inference.core.interfaces.stream.model_handlers.workflows import (
+        LookaheadPipelinedWorkflowRunner,
+        WorkflowRunner,
+        wrap_workflow_runner_for_stream_pipeline,
+    )
+    from inference.core.workflows.core_steps.models.roboflow.object_detection import (
+        v3 as object_detection_v3,
+    )
+    from inference.core.workflows.execution_engine.v1.executor.core import (
+        compute_stream_lookahead_frontier,
+    )
+
+    mock_client = MagicMock()
+    mock_client.infer.return_value = CANNED_PREDICTION
+    image = np.zeros((360, 640, 3), dtype=np.uint8)
+    results: Dict[str, Any] = {}
+
+    with patch.object(
+        object_detection_v3, "InferenceHTTPClient", MagicMock(return_value=mock_client)
+    ), patch.object(object_detection_v3, "InferenceConfiguration", MagicMock()):
+        results["compilation"] = time_repeated(build_engine, repeats=args.compiles)
+
+        engine = build_engine()
+        compiled_workflow = engine._engine._compiled_workflow
+        results["frontier_computation"] = time_repeated(
+            lambda: compute_stream_lookahead_frontier(workflow=compiled_workflow),
+            repeats=1000,
+        )
+
+        def build_workflow_runner():
+            return WorkflowRunner(
+                workflows_parameters=None,
+                execution_engine=engine,
+                image_input_name="image",
+                video_metadata_input_name="video_metadata",
+            )
+
+        workflows_module.WORKFLOWS_STREAM_LOOKAHEAD_DEPTH = 1
+        sequential_runner = build_workflow_runner()
+        results["sequential_per_frame"] = run_per_frame_measurement(
+            runner=sequential_runner,
+            image=image,
+            frames=args.frames,
+            warmup=args.warmup,
+            drain=lambda: None,
+        )
+
+        workflows_module.WORKFLOWS_STREAM_LOOKAHEAD_DEPTH = args.depth
+        lookahead_engine = build_engine()
+        lookahead_runner = wrap_workflow_runner_for_stream_pipeline(
+            workflow_runner=WorkflowRunner(
+                workflows_parameters=None,
+                execution_engine=lookahead_engine,
+                image_input_name="image",
+                video_metadata_input_name="video_metadata",
+            ),
+            execution_engine=lookahead_engine,
+        )
+        if not isinstance(lookahead_runner, LookaheadPipelinedWorkflowRunner):
+            raise RuntimeError(
+                f"Expected the lookahead runner to activate, got "
+                f"{type(lookahead_runner).__name__} — results would be invalid."
+            )
+        try:
+            results["lookahead_per_frame"] = run_per_frame_measurement(
+                runner=lookahead_runner,
+                image=image,
+                frames=args.frames,
+                warmup=args.warmup,
+                drain=lookahead_runner.flush,
+            )
+        finally:
+            lookahead_runner.close()
+
+    results["scheduler_overhead_ms_per_frame"] = round(
+        results["lookahead_per_frame"]["avg_ms_per_frame"]
+        - results["sequential_per_frame"]["avg_ms_per_frame"],
+        4,
+    )
+    print("OVERHEAD_RESULT " + json.dumps(results, indent=2))
+    if args.output:
+        with open(args.output, "w") as f:
+            json.dump(results, f, indent=2)
+
+
+if __name__ == "__main__":
+    main()

docs/workflows/execution_engine_changelog.md

@@ -2,6 +2,32 @@
 
 Below you can find the changelog for Execution Engine.
 
+## Execution Engine `v1.13.0` | inference `v1.3.5`
+
+**What changed**
+
+* **Stream-lookahead execution for video pipelines (opt-in, default off)** — With
+  `WORKFLOWS_STREAM_LOOKAHEAD_DEPTH=N` (N > 1), `InferencePipeline` video workflows may
+  keep up to N frames' long-latency step executions (remote model requests, external
+  API calls) in flight concurrently, while stateful steps (trackers, counters, sinks)
+  still observe frames strictly in stream order. Emission order and outputs are
+  identical to sequential execution.
+* **Per-block statefulness declaration** — `WorkflowBlockManifest` gained the
+  `is_stateful_for_video_processing()` classmethod (default `True`, conservative).
+  Blocks that are pure functions of their per-frame inputs may declare `False`;
+  blocks additionally declaring `is_async_stream_step()` on the block class certify
+  their `run()` as long-latency and re-entrant, letting the engine offload it to a
+  lookahead worker pool with per-output `Future` placeholders built from
+  `describe_outputs()`.
+* **New engine entry points** — `run_stream_lookahead(...)` (deferred pass: executes
+  the stateless-ancestor frontier of the DAG and returns the frame's live
+  `ExecutionDataManager`) and `resume_stream_lookahead(...)` (emission pass: runs the
+  remaining steps on that state). Both exist on `ExecutionEngine` and
+  `ExecutionEngineV1` only — like `flush_stream_pipeline`, they are intentionally not
+  part of the `BaseExecutionEngine` ABC. Default `run(...)` behavior is unchanged;
+  with the env variable unset (default `1`) nothing about compilation or execution
+  differs.
+
 ## Execution Engine `v1.12.0` | inference `v1.3.2`
 
 **What changed**

inference/core/env.py

@@ -741,6 +741,15 @@
 WORKFLOWS_REMOTE_EXECUTION_MAX_STEP_CONCURRENT_REQUESTS = int(
     os.getenv("WORKFLOWS_REMOTE_EXECUTION_MAX_STEP_CONCURRENT_REQUESTS", "8")
 )
+# Number of video frames whose async step executions (remote models, API
+# calls) may be in flight concurrently in stream-lookahead scheduling;
+# 1 disables lookahead. The old name is honored as a fallback.
+WORKFLOWS_STREAM_LOOKAHEAD_DEPTH = int(
+    os.getenv(
+        "WORKFLOWS_STREAM_LOOKAHEAD_DEPTH",
+        os.getenv("WORKFLOWS_REMOTE_EXECUTION_PIPELINE_DEPTH", "1"),
+    )
+)
 ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS = str2bool(
     os.getenv("ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS", True)
 )