Add server-side frame pacing scheduler to smooth real-time stream

app.py

@@ -81,6 +81,20 @@ MAX_BLOCKS_PER_SESSION = 512     # hard cap so a session can't run forever
 SESSION_IDLE_TIMEOUT = 600       # seconds; janitor reaps abandoned sessions
 GPU_DURATION = 120               # seconds per @spaces.GPU allocation
 
+# ── Real-time pacing configuration ───────────────────────────────────────────
+# The GPU decodes a whole block (NUM_FPB frames) at once, so all of a block's
+# frames become available in a burst. If we forward them to the browser the
+# instant they finish, the client sees N frames clustered together followed by a
+# gap while the next block generates — the fps counter averages out fine, but
+# the *felt* cadence is bursty. To deliver a steady real-time stream we pace the
+# frames of each block evenly across the time we expect one block to take
+# (mirroring the official ABot-World web_client's block-frame spreader), and we
+# smooth the per-block generation time with an EMA so a single slow/fast block
+# doesn't cause a visible speed-up/slow-down. See gpu_worker_thread().
+PACING_EMA_ALPHA = 0.25          # smoothing factor for per-block generation time
+MIN_PACING_SLEEP = 0.004         # don't bother sleeping for sub-4ms slices
+DEFAULT_BLOCK_SECONDS = 0.5      # initial per-block estimate before first measure
+
 # Actions map to the 8-key one-hot the model was trained on (W A S D I J K L).
 # The browser sends the currently-held key set; we translate to this dict.
 KEY_ORDER = ["W", "A", "S", "D", "I", "J", "K", "L"]
@@ -292,9 +306,17 @@ def gpu_worker_thread(session: "GameSession"):
             session.command_queue, session.seed_path, session.prompt, session.seed,
         )
         first = True
+        # Steady send clock: `next_send` is the monotonic time at which the next
+        # frame *should* be delivered. Each frame's slot is one smoothed
+        # inter-frame interval after the previous, so frames leave at a constant
+        # cadence regardless of the bursty block boundaries. `block_seconds` is
+        # an EMA of measured per-block generation time (frames/block ÷ that gives
+        # the target inter-frame interval).
+        block_seconds = DEFAULT_BLOCK_SECONDS
+        next_send = None
         while not session.stop_event.is_set():
             try:
-                frame, frame_count = next(gen)
+                frame, block_idx, frame_idx, frames_in_block, block_elapsed = next(gen)
             except StopIteration:
                 print("Rollout generator exhausted", flush=True)
                 break
@@ -310,20 +332,47 @@ def gpu_worker_thread(session: "GameSession"):
                 broadcast_status("Rolling out — use WASD / IJKL to steer.")
                 first = False
 
+            # Update the smoothed per-block time on the first frame of each block
+            # (block_elapsed is constant across a block's frames).
+            if frame_idx == 0 and block_elapsed > 0:
+                block_seconds = (
+                    PACING_EMA_ALPHA * block_elapsed
+                    + (1.0 - PACING_EMA_ALPHA) * block_seconds
+                )
+            fpb = max(1, frames_in_block)
+            interval = block_seconds / fpb        # target seconds between frames
+
+            # ── Steady-cadence gate ──────────────────────────────────────────
+            # Hold each frame until its scheduled slot so the parent emits at a
+            # constant interval instead of dumping a whole block at once.
+            now = time.time()
+            if next_send is None:
+                next_send = now
+            sleep_for = next_send - now
+            if sleep_for > MIN_PACING_SLEEP:
+                # Wake early if a stop is requested so we stay responsive.
+                if session.stop_event.wait(timeout=sleep_for):
+                    break
+                now = time.time()
+            # Advance the schedule; if we've fallen far behind (e.g. a long GPU
+            # stall), resync to now so we don't try to "catch up" in a burst.
+            next_send = max(now, next_send + interval)
+
             now = time.time()
             session.frame_times.append(now)
             fps = 0.0
             if len(session.frame_times) >= 2:
                 elapsed = session.frame_times[-1] - session.frame_times[0]
                 fps = (len(session.frame_times) - 1) / elapsed if elapsed > 0 else 0.0
-            # Keep only the freshest frame if the consumer fell behind.
-            while session.frame_queue.qsize() > 2:
+            # Keep only the freshest frame if the consumer fell behind: coalesce
+            # stale frames rather than letting them queue up and flush in a burst.
+            while session.frame_queue.qsize() > 1:
                 try:
                     session.frame_queue.get_nowait()
                 except queue.Empty:
                     break
             try:
-                session.frame_queue.put_nowait((frame, frame_count, round(fps, 1)))
+                session.frame_queue.put_nowait((frame, block_idx, round(fps, 1)))
             except queue.Full:
                 pass
     finally:
@@ -377,13 +426,21 @@ def create_gpu_rollout_loop(command_queue, seed_path, prompt_text, seed):
                     if stop_requested:
                         break
 
+                    # Time the full generate+decode of one block so the parent
+                    # thread can pace this block's frames over that duration.
+                    block_start = time.time()
                     pipeline.set_act(current_keys, height=STREAM_HEIGHT, width=STREAM_WIDTH,
                                      num_frames=NUM_FPB, device=device)
                     lat_block = pipeline.generate_next_block(noise)
                     noise = torch.randn_like(noise)
                     frames = _decode_block_to_frames(lat_block)
-                    for f in frames:
-                        yield (f, b)
+                    block_elapsed = time.time() - block_start
+                    n = len(frames)
+                    for i, f in enumerate(frames):
+                        # (frame, block_idx, frame_idx_in_block, frames_in_block,
+                        #  block_elapsed) — the pacing metadata lets the parent
+                        # spread this block's frames evenly rather than bursting.
+                        yield (f, b, i, n, block_elapsed)
             finally:
                 try:
                     pipeline.reset_stream(batch_size=1, dtype=torch.bfloat16,