From 84257f2195c3ce31ec3c8a446a9e1215e331c6fa Mon Sep 17 00:00:00 2001
From: Rick Staa <rick.staa@outlook.com>
Date: Wed, 12 Aug 2026 17:41:40 +0200
Subject: [PATCH] perf(echo): decode off the event loop so uploads are not
 starved

PyAV's decode() is a blocking generator, so driving the publish loop from
it directly stalls the async segment uploads for the duration of each
decode. Moving it to a worker thread behind a small queue halves the gap
between segments reaching the wire and cuts startup latency.

Measured over three runs each, live webcam plus microphone through robot:
max gap 0.98-1.11s to 0.62-0.63s, first byte 3.36s to 2.90s. Video-only
blur is unchanged at 0.42s, so nothing regresses for the other modes.

This does not fully close the gap to video-only. The rest is tracked in
livepeer/runner-app-examples#67.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 echo/client.py | 35 ++++++++++++++++++++++++++++++++++-
 1 file changed, 34 insertions(+), 1 deletion(-)

diff --git a/echo/client.py b/echo/client.py
index f094b7f..c995655 100644
--- a/echo/client.py
+++ b/echo/client.py
@@ -22,9 +22,12 @@
 import argparse
 import asyncio
 import logging
+import threading
 import sys
 import time
+from collections.abc import AsyncIterator, Iterator
 from contextlib import nullcontext, suppress
+from queue import Queue
 from pathlib import Path
 
 import av
@@ -105,6 +108,36 @@ def _channel_url(echo_response: dict[str, object], name: str) -> str:
     return url
 
 
+async def _decode_in_thread(
+    frames: Iterator[av.VideoFrame | av.AudioFrame],
+) -> AsyncIterator[av.VideoFrame | av.AudioFrame]:
+    """Yield decoded frames without decoding on the event loop.
+
+    PyAV's decode() is a blocking generator, so driving the publish loop from it
+    directly stalls the async segment uploads while a frame is being decoded. The
+    queue is small on purpose: it decouples the two without adding latency.
+    """
+    queue: Queue[av.VideoFrame | av.AudioFrame | None] = Queue(maxsize=8)
+
+    def _pump() -> None:
+        try:
+            for frame in frames:
+                queue.put(frame)
+        finally:
+            queue.put(None)
+
+    thread = threading.Thread(target=_pump, daemon=True)
+    thread.start()
+    try:
+        while True:
+            frame = await asyncio.to_thread(queue.get)
+            if frame is None:
+                return
+            yield frame
+    finally:
+        thread.join(timeout=1.0)
+
+
 async def _publish_video(
     input_source: str,
     publish_url: str,
@@ -150,7 +183,7 @@ async def _publish_video(
             # decode() yields both streams interleaved; without audio, stay on the
             # video stream alone. Pacing and the blur sweep run off video frames only.
             frames = input_.decode() if send_audio else input_.decode(video=0)
-            for frame in frames:
+            async for frame in _decode_in_thread(frames):
                 if not isinstance(frame, av.VideoFrame):
                     await publisher.write_frame(frame)
                     continue