diff --git a/AGENTS.md b/AGENTS.md index ad8ce090d..be40a387a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -85,6 +85,7 @@ Use `--no-instantiate` before GPU work to inspect the resolved runner config wit - Inspect available runners with `uv run flashdreams-run --help`, then inspect a specific runner with `uv run flashdreams-run --no-instantiate `. - Prefer CPU checks first: config imports, checkpoint key-remap shape/bijection tests on CPU or meta tensors, docs builds, `pytest -m ci_cpu`, and static assertions about runner names and pipeline wiring. +- For interactive V2 host-side input latency, capture `--profile-path artifacts/.jsonl` and read `docs/source/developer_guides/latency_tuning.rst` for the two measurement boundaries. - Avoid `ci_gpu`, generation, `torchrun`, Docker GPU tests, large Hugging Face downloads, rollout parity, CUDA graph, WebRTC runtime, and quality-regression tests on CPU-only hosts unless the user requests them or the test explicitly skips cleanly. ## Testing Guidance diff --git a/docs/source/developer_guides/latency_tuning.rst b/docs/source/developer_guides/latency_tuning.rst index 2b32f91b0..11379be03 100644 --- a/docs/source/developer_guides/latency_tuning.rst +++ b/docs/source/developer_guides/latency_tuning.rst @@ -157,6 +157,44 @@ resolution, and native-acceleration knobs first. Profiling and validated reference --------------------------------- +V2 host-side input latency +~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Use ``--profile-path`` with a V2 application to write a JSONL profile: + +.. code-block:: bash + + uv run flashdreams-run-v2 interactive-drive-omnidreams-perf \ + --mode native-window \ + --profile-path artifacts/interactive-drive-input-latency.jsonl + +Replacement sessions append separate ``session_started`` and summary segments +to the same artifact. Each ``session_started`` record identifies the concrete +client-window type, output layout, UI and model rates, resolution, presentation +mode, backpressure mode, and both measurement endpoints. + +The input source publishes the monotonic origin for its session-relative +``UserInputEvent.timestamp`` values. ``run_session`` then records two metrics: + +- ``input_to_ui_step_s`` ends when the IUILoop claims the event. +- ``input_to_window_write_s`` ends when the first following client-window + ``write`` call returns. + +The native-window measurement ends when its window ``write`` returns, normally +after the presenter call. The WebRTC measurement also ends when ``write`` +returns. An active video track returns after host materialization and admission +to the bounded two-frame sender queue. A server waiting for a video track +returns after shape validation. Active-peer delivery, RTP transit, browser +decode, compositor scheduling, and physical scanout require matching client +telemetry. +The final ``profile_summary`` records report an exact count and maximum. +Median and p90 use every sample through 1,024 observations, then use a uniform +bounded reservoir. ``quantile_sample_count`` and ``quantiles_approximate`` make +that transition explicit in the artifact. + +JSONL writes add host overhead to the measured run. Keep profiling enabled for +every run in a direct comparison. Use Nsight Systems for GPU stage attribution. + Set ``output.profile_world_model: true`` to enable FlashDreams CUDA-event profiling for the world-model runtime. Set ``output.sync_gpu_timing: true`` only when you need raster compute diff --git a/flashdreams/flashdreams/api_v2/input_source.py b/flashdreams/flashdreams/api_v2/input_source.py index d64a4df88..f61c6ed3b 100644 --- a/flashdreams/flashdreams/api_v2/input_source.py +++ b/flashdreams/flashdreams/api_v2/input_source.py @@ -31,3 +31,17 @@ def get_user_input_events(self) -> UserInputEvents: Events in timestamp order, empty when nothing arrived. """ ... + + +@runtime_checkable +class TimestampedInputSource(InputSource, Protocol): + """Input source whose event timestamps share the runtime monotonic clock.""" + + @property + @abstractmethod + def input_timestamp_origin_ns(self) -> int | None: + """Return the runtime-clock origin for session-relative timestamps.""" + ... + + +__all__ = ["InputSource", "TimestampedInputSource"] diff --git a/flashdreams/flashdreams/runtime_v2/README.md b/flashdreams/flashdreams/runtime_v2/README.md index 4c239a3eb..f2f7b4aa3 100644 --- a/flashdreams/flashdreams/runtime_v2/README.md +++ b/flashdreams/flashdreams/runtime_v2/README.md @@ -224,6 +224,14 @@ The default UI loop, `BlitModelOutputToScreenLoop`, composites every model channel in list order as if they were image layers and reshapes the result into the session's layout. +`--profile-path artifacts/.jsonl` records host-side event-to-IUILoop and +event-to-window-write latency. Each input source supplies the monotonic origin +for its session-relative event timestamps. Replacement sessions append distinct +segments with measurement endpoints and runtime settings to the same JSONL +artifact. The +[latency tuning guide](../../../docs/source/developer_guides/latency_tuning.rst) +defines the JSONL records and host-side boundaries. + `SlangPyUILoop` is the alternative for SlangPy's retained widget subset. `ImGuiUILoop` exposes the complete ImGui API. Both return a `[1, C, H, W]` frame, so an `ISession` using either should declare a `tchw` output layout. diff --git a/flashdreams/flashdreams/runtime_v2/__init__.py b/flashdreams/flashdreams/runtime_v2/__init__.py index 8bcb7a5ce..69001db64 100644 --- a/flashdreams/flashdreams/runtime_v2/__init__.py +++ b/flashdreams/flashdreams/runtime_v2/__init__.py @@ -3,10 +3,11 @@ """Runtime that runs a FlashDreams v2 application against a client window.""" +from flashdreams.runtime_v2.runtime_profiler import RuntimeProfiler from flashdreams.runtime_v2.session_desc import ( BackpressureMode, PresentationMode, SessionDesc, ) -__all__ = ["BackpressureMode", "PresentationMode", "SessionDesc"] +__all__ = ["BackpressureMode", "PresentationMode", "RuntimeProfiler", "SessionDesc"] diff --git a/flashdreams/flashdreams/runtime_v2/application_runner.py b/flashdreams/flashdreams/runtime_v2/application_runner.py index 4e518d62d..ed0708bc8 100644 --- a/flashdreams/flashdreams/runtime_v2/application_runner.py +++ b/flashdreams/flashdreams/runtime_v2/application_runner.py @@ -12,6 +12,7 @@ from flashdreams.api_v2.application import IApplication from flashdreams.api_v2.client_window import IClientWindow from flashdreams.api_v2.output_sink import OutputSink +from flashdreams.runtime_v2.runtime_profiler import RuntimeProfiler from flashdreams.runtime_v2.session_desc import SessionDesc from flashdreams.runtime_v2.session_runner import run_session @@ -28,6 +29,7 @@ def __init__( client_window: IClientWindow, *, metrics_output_sink: OutputSink | None = None, + profiler: RuntimeProfiler | None = None, ) -> None: """ Args: @@ -35,10 +37,13 @@ def __init__( client_window: Window that supplies input and presents generated output. metrics_output_sink: Optional sink for model-step metrics. It is opened and closed once for each session. + profiler: Optional host-side input-latency profiler. It is opened + and closed once for each session. """ self._application = application self._client_window = client_window self._metrics_output_sink = metrics_output_sink + self._profiler = profiler def run( self, @@ -95,6 +100,7 @@ def run( session, self._client_window, metrics_output_sink=self._metrics_output_sink, + profiler=self._profiler, timeout_seconds=remaining_seconds, ) except BaseException: diff --git a/flashdreams/flashdreams/runtime_v2/cli.py b/flashdreams/flashdreams/runtime_v2/cli.py index e1e98366a..9dfcaabde 100644 --- a/flashdreams/flashdreams/runtime_v2/cli.py +++ b/flashdreams/flashdreams/runtime_v2/cli.py @@ -18,6 +18,7 @@ import sys from collections.abc import Sequence from dataclasses import replace +from pathlib import Path from typing import Any from flashdreams.api_v2.application import IApplication @@ -31,6 +32,7 @@ client_window_mode, ) from flashdreams.runtime_v2.metrics_output_sink import MetricsOutputSink +from flashdreams.runtime_v2.runtime_profiler import RuntimeProfiler from flashdreams.runtime_v2.session_desc import ( BackpressureMode, PresentationMode, @@ -68,6 +70,7 @@ def entrypoint(argv: Sequence[str] | None = None) -> None: if not wants_application_help: try: mode.check_arguments(parsed) + _validate_profile_path(parsed) except ValueError as error: parser.error(str(error)) @@ -84,10 +87,14 @@ def entrypoint(argv: Sequence[str] | None = None) -> None: metrics_output_sink = ( None if parsed.stats_path is None else MetricsOutputSink(parsed.stats_path) ) + profiler = ( + None if parsed.profile_path is None else RuntimeProfiler(parsed.profile_path) + ) ApplicationRunner( application, window, metrics_output_sink=metrics_output_sink, + profiler=profiler, ).run( session_desc, application_args, @@ -151,6 +158,12 @@ def _add_session_arguments(parser: argparse.ArgumentParser) -> None: Each defaults to asking for nothing, so a run that names none of them gets what the application generates. """ + parser.add_argument( + "--profile-path", + type=Path, + default=None, + help="Write host-side input-latency records as JSONL.", + ) parser.add_argument( "--pixel-width", type=int, default=None, help="Frame width to generate." ) @@ -191,6 +204,21 @@ def _add_session_arguments(parser: argparse.ArgumentParser) -> None: ) +def _validate_profile_path(parsed: argparse.Namespace) -> None: + """Keep the profile separate from artifacts written by the same run.""" + if parsed.profile_path is None: + return + profile_path = parsed.profile_path.expanduser().resolve() + other_paths = [parsed.stats_path] + if parsed.mode == "mp4": + other_paths.append(parsed.output_path) + if any( + path is not None and path.expanduser().resolve() == profile_path + for path in other_paths + ): + raise ValueError("--profile-path must use a distinct output path.") + + def _session_desc( application: IApplication, parsed_args: argparse.Namespace ) -> SessionDesc: diff --git a/flashdreams/flashdreams/runtime_v2/native_window_client_window.py b/flashdreams/flashdreams/runtime_v2/native_window_client_window.py index afaaa4612..960aef9f3 100644 --- a/flashdreams/flashdreams/runtime_v2/native_window_client_window.py +++ b/flashdreams/flashdreams/runtime_v2/native_window_client_window.py @@ -91,6 +91,13 @@ class NativeWindowClientWindow(IClientWindow): """Present UI output through a main-thread GLFW window.""" + @property + def input_timestamp_origin_ns(self) -> int | None: + """Return the native session's monotonic input timestamp origin.""" + if not self._input_timestamps_share_runtime_clock: + return None + return self._session_started_ns + def __init__( self, *, @@ -113,6 +120,7 @@ def __init__( self.title = title self._presenter_factory = presenter_factory self._clock_ns = clock_ns + self._input_timestamps_share_runtime_clock = clock_ns is time.monotonic_ns self._session_started_ns: int | None = None self._session_desc: SessionDesc | None = None self._input_events: queue.SimpleQueue[UserInputEvent] = queue.SimpleQueue() @@ -158,7 +166,8 @@ def open(self, session_desc: SessionDesc) -> None: """ if threading.current_thread() is not threading.main_thread(): raise RuntimeError( - "NativeWindowClientWindow.open() must run on the process main thread for event polling." + "NativeWindowClientWindow.open() must run on the process main " + "thread for event polling." ) if self._presenter is not None: raise RuntimeError("NativeWindowClientWindow is already open.") diff --git a/flashdreams/flashdreams/runtime_v2/runtime_profiler.py b/flashdreams/flashdreams/runtime_v2/runtime_profiler.py new file mode 100644 index 000000000..1eab638c1 --- /dev/null +++ b/flashdreams/flashdreams/runtime_v2/runtime_profiler.py @@ -0,0 +1,314 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Host-side input-latency profiling for the V2 runtime.""" + +from __future__ import annotations + +import json +import random +import statistics +import time +from dataclasses import dataclass +from pathlib import Path +from typing import IO, Any + +from flashdreams.runtime_v2.session_desc import SessionDesc +from flashdreams.runtime_v2.user_input_events import UserInputEvents + +_ARTIFACT_TYPE = "flashdreams.runtime_v2.input_latency_profile" +"""Artifact type for V2 host-side input-latency records.""" + +_SCHEMA_VERSION = 1 +"""Input-latency artifact schema version.""" + +_SUMMARY_SAMPLE_LIMIT = 1_024 +"""Maximum observations retained for session quantiles.""" + + +@dataclass(frozen=True, slots=True) +class _ClaimedInput: + received_at_ns: int + claimed_at_ns: int + claim_duration_s: float + timestamp_us: int + event_type: str + generation: int + ui_step: int + + +class _MetricAccumulator: + """Keep exact counts and maxima with bounded quantile samples.""" + + def __init__(self) -> None: + self._count = 0 + self._maximum: float | None = None + self._samples: list[float] = [] + self._random = random.Random(0) + + def add(self, value: float) -> None: + """Add one sample using deterministic reservoir sampling.""" + self._count += 1 + self._maximum = value if self._maximum is None else max(self._maximum, value) + if len(self._samples) < _SUMMARY_SAMPLE_LIMIT: + self._samples.append(value) + return + replacement = self._random.randrange(self._count) + if replacement < _SUMMARY_SAMPLE_LIMIT: + self._samples[replacement] = value + + def summary(self) -> dict[str, float | int | bool]: + """Return bounded-memory summary statistics.""" + if self._count == 0: + return {"count": 0} + ordered = sorted(self._samples) + maximum = self._maximum + assert maximum is not None + return { + "count": self._count, + "median_s": statistics.median(ordered), + "p90_s": _percentile(ordered, 0.9), + "max_s": maximum, + "quantile_sample_count": len(ordered), + "quantiles_approximate": self._count > len(ordered), + } + + +class RuntimeProfiler: + """Write input-to-IUILoop and input-to-window-write records as JSONL. + + One runtime UI thread owns an instance. Input timestamps and runtime + observations share ``time.monotonic_ns`` through the source-provided session + origin. Each claimed input remains pending through the first subsequent + client-window write. Reopening the profiler appends an independent session + segment to the same artifact. + """ + + def __init__(self, path: str | Path) -> None: + """Configure a profile artifact that opens with the session.""" + self._path = Path(path).expanduser() + self._output: IO[str] | None = None + self._input_timestamp_origin_ns: int | None = None + self._input_generation: int | None = None + self._pending_inputs: list[_ClaimedInput] = [] + self._samples = _metric_accumulators() + self._opened_once = False + + @property + def path(self) -> Path: + """Return the profile output path.""" + return self._path + + def session_started( + self, + *, + input_timestamp_origin_ns: int | None, + session_desc: SessionDesc, + client_window_type: str, + time_ns: int | None = None, + ) -> None: + """Open the artifact and bind input timestamps to the runtime clock.""" + if self._output is not None: + raise RuntimeError("RuntimeProfiler session already started.") + self._path.parent.mkdir(parents=True, exist_ok=True) + mode = "a" if self._opened_once else "w" + self._output = self._path.open(mode, encoding="utf-8") + self._opened_once = True + self._input_timestamp_origin_ns = input_timestamp_origin_ns + self._input_generation = None + self._pending_inputs.clear() + self._samples = _metric_accumulators() + self._write( + "session_started", + _now_ns(time_ns), + input_timestamp_origin_ns=input_timestamp_origin_ns, + measurement_endpoints={ + "input_to_ui_step_s": "ui_loop_begin_run", + "input_to_window_write_s": "client_window_write_return", + }, + runtime_settings={ + "client_window_type": client_window_type, + "output_layout": session_desc.output_layout.value, + "frames_per_second_for_ui": session_desc.frames_per_second_for_ui, + "frames_per_second_for_step": session_desc.frames_per_second_for_step, + "video_width": session_desc.video_width, + "video_height": session_desc.video_height, + "presentation_mode": session_desc.presentation_mode.value, + "backpressure_mode": session_desc.backpressure_mode.value, + }, + ) + + def ui_step_started( + self, + events: UserInputEvents, + *, + generation: int, + step: int, + time_ns: int | None = None, + ) -> None: + """Record the UI step that first claims each input event.""" + started_at_ns = _now_ns(time_ns) + self._ensure_started() + if generation != self._input_generation: + self._write_claims(self._pending_inputs) + self._pending_inputs.clear() + self._input_generation = generation + + origin_ns = self._input_timestamp_origin_ns + if origin_ns is None: + return + for event in events.get_events(): + received_at_ns = origin_ns + int(event.get_timestamp()) * 1_000 + claim_duration_s = _elapsed_s(received_at_ns, started_at_ns) + self._pending_inputs.append( + _ClaimedInput( + received_at_ns=received_at_ns, + claimed_at_ns=started_at_ns, + claim_duration_s=claim_duration_s, + timestamp_us=int(event.get_timestamp()), + event_type=event.get_type_name(), + generation=generation, + ui_step=step, + ) + ) + self._samples["input_to_ui_step_s"].add(claim_duration_s) + + def window_write_completed( + self, + *, + generation: int, + ui_step: int, + time_ns: int | None = None, + ) -> None: + """Finish pending input correlations at the next window write return.""" + completed_at_ns = _now_ns(time_ns) + self._ensure_started() + pending = tuple( + observation + for observation in self._pending_inputs + if observation.generation == generation + ) + self._pending_inputs.clear() + self._write_claims(pending) + for observation in pending: + duration_s = _elapsed_s(observation.received_at_ns, completed_at_ns) + self._samples["input_to_window_write_s"].add(duration_s) + self._write( + "input_to_window_write", + completed_at_ns, + generation=generation, + claimed_ui_step=observation.ui_step, + presented_ui_step=ui_step, + input_type=observation.event_type, + input_timestamp_us=observation.timestamp_us, + duration_s=duration_s, + ) + + def summary(self) -> dict[str, dict[str, float | int | bool]]: + """Return summary statistics for both host-side latency metrics.""" + return {name: values.summary() for name, values in self._samples.items()} + + def close(self) -> None: + """Write summaries and close the active session segment.""" + output = self._output + if output is None: + return + failure: BaseException | None = None + try: + self._write_claims(self._pending_inputs) + observed_at_ns = time.monotonic_ns() + for metric, values in self._samples.items(): + self._write( + "profile_summary", + observed_at_ns, + metric=metric, + **values.summary(), + ) + except BaseException as error: + failure = error + self._pending_inputs.clear() + try: + output.close() + except BaseException as error: + if failure is None: + failure = error + self._output = None + self._input_timestamp_origin_ns = None + self._input_generation = None + if failure is not None: + raise failure + + def _write_claims( + self, + observations: list[_ClaimedInput] | tuple[_ClaimedInput, ...], + ) -> None: + for observation in observations: + self._write( + "input_to_ui_step", + observation.claimed_at_ns, + generation=observation.generation, + step=observation.ui_step, + input_type=observation.event_type, + input_timestamp_us=observation.timestamp_us, + duration_s=observation.claim_duration_s, + ) + + def _write(self, phase: str, time_ns: int, **fields: Any) -> None: + output = self._output + if output is None: + raise RuntimeError("RuntimeProfiler session has not started.") + record = { + **fields, + "artifact_type": _ARTIFACT_TYPE, + "schema_version": _SCHEMA_VERSION, + "phase": phase, + "time_ns": time_ns, + } + output.write(json.dumps(record, sort_keys=True, separators=(",", ":"))) + output.write("\n") + + def _ensure_started(self) -> None: + if self._output is None: + raise RuntimeError("RuntimeProfiler session has not started.") + + +def _now_ns(value: int | None) -> int: + return time.monotonic_ns() if value is None else value + + +def _elapsed_s(start_ns: int, end_ns: int) -> float: + if end_ns < start_ns: + raise ValueError("A runtime profile observation moved backward in time.") + return (end_ns - start_ns) / 1_000_000_000 + + +def _metric_accumulators() -> dict[str, _MetricAccumulator]: + return { + "input_to_ui_step_s": _MetricAccumulator(), + "input_to_window_write_s": _MetricAccumulator(), + } + + +def _percentile(ordered: list[float], percentile: float) -> float: + if len(ordered) == 1: + return ordered[0] + index = percentile * (len(ordered) - 1) + lower = int(index) + upper = min(lower + 1, len(ordered) - 1) + fraction = index - lower + return ordered[lower] * (1.0 - fraction) + ordered[upper] * fraction + + +__all__ = ["RuntimeProfiler"] diff --git a/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py b/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py index 59535f190..a00b7a24c 100644 --- a/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py +++ b/flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py @@ -374,6 +374,11 @@ def url(self) -> str: """Return the browser URL for this server.""" return f"http://{self._host}:{self._port}/" + @property + def input_timestamp_origin_ns(self) -> int | None: + """Return the server event clock's monotonic origin.""" + return self._event_origin_ns + def metrics_snapshot(self) -> dict[str, float | int]: """Return non-blocking sender diagnostics.""" track = self._video_track diff --git a/flashdreams/flashdreams/runtime_v2/session_runner.py b/flashdreams/flashdreams/runtime_v2/session_runner.py index 7d9ed3ec8..261d9bb4b 100644 --- a/flashdreams/flashdreams/runtime_v2/session_runner.py +++ b/flashdreams/flashdreams/runtime_v2/session_runner.py @@ -11,10 +11,12 @@ from pathlib import Path from flashdreams.api_v2.client_window import IClientWindow +from flashdreams.api_v2.input_source import TimestampedInputSource from flashdreams.api_v2.loop import IModelLoop, IUILoop, ModelInferenceState from flashdreams.api_v2.output_sink import OutputSink from flashdreams.api_v2.session import ISession from flashdreams.runtime_v2.event_buffer import EventBuffer +from flashdreams.runtime_v2.runtime_profiler import RuntimeProfiler from flashdreams.runtime_v2.session_desc import PresentationMode, SessionDesc from flashdreams.runtime_v2.step_result import StepResult @@ -48,6 +50,7 @@ def run_session( window: IClientWindow, *, metrics_output_sink: OutputSink | None = None, + profiler: RuntimeProfiler | None = None, steps: int | None = None, timeout_seconds: float | None = None, ) -> SessionDesc | None: @@ -59,15 +62,16 @@ def run_session( is not running, an unfinished UI ticks regardless of presentation mode so it can request another session. - Both loops, the metrics sink, and the session are closed before this returns - or raises. The client window stays open only when a clean replacement was - requested; otherwise it is closed. + Both loops, the metrics sink, the profiler, and the session are closed before + this returns or raises. A clean replacement request preserves the client + window. Every other exit closes it. Args: session: Session to run. window: Source of input and destination for UI output. metrics_output_sink: Sink for model measurements, if requested. Receives the model loop's results rather than the UI loop's. + profiler: Optional host-side input-latency profiler. steps: Maximum model steps before ending the session; ``None`` leaves session completion to the UI or client window. timeout_seconds: Maximum session runtime; ``None`` means session does not have a time-limit. Timeout expiry signals both registered loops to stop. @@ -103,6 +107,8 @@ def run_session( trace_log: _ChunkTraceLog | None = None try: session_desc = session.session_desc + if profiler is not None: + _validate_profile_path(session_desc, profiler.path) tick_seconds = 1.0 / session_desc.frames_per_second_for_ui stop = session._shutdown_event presentation_manager = session._presentation_manager @@ -125,6 +131,7 @@ def run_ui_once(*, step_requested: bool = True) -> None: if ui_loop is None: return events, generation = event_buffer.read(_UI_READER_ID) + input_claimed_at_ns = time.monotonic_ns() if profiler is not None else None result: StepResult | None = None step_completed = False try: @@ -132,7 +139,6 @@ def run_ui_once(*, step_requested: bool = True) -> None: if loop_result.stop_requested: stop.set() return - request = ui_loop.flush_ui_loop_requests() if request is not None: if request.hide_cursor is not None: @@ -145,7 +151,16 @@ def run_ui_once(*, step_requested: bool = True) -> None: next_session_desc = request.new_session stop.set() return - if loop_result.step_index is None or not step_requested: + if loop_result.step_index is None: + return + if profiler is not None: + profiler.ui_step_started( + ui_loop.user_events, + generation=generation, + step=loop_result.step_index, + time_ns=input_claimed_at_ns, + ) + if not step_requested: return raw_result = ui_loop.step(loop_result.step_index, ui_loop.user_events) if raw_result is not None and not isinstance(raw_result, StepResult): @@ -156,6 +171,12 @@ def run_ui_once(*, step_requested: bool = True) -> None: ui_loop._finish_run(result, step_completed=step_completed) if result is not None: window.write(result) + if profiler is not None: + profiler.window_write_completed( + generation=generation, + ui_step=loop_result.step_index, + time_ns=time.monotonic_ns(), + ) def publish_model_results( generation: int, @@ -223,6 +244,16 @@ def tick_ui() -> None: event_buffer.register(_MODEL_READER_ID) window.open(session_desc) + if profiler is not None: + profiler.session_started( + input_timestamp_origin_ns=( + window.input_timestamp_origin_ns + if isinstance(window, TimestampedInputSource) + else None + ), + session_desc=session_desc, + client_window_type=type(window).__name__, + ) if metrics_output_sink is not None: metrics_output_sink.open(session_desc) collect_input() @@ -315,6 +346,11 @@ def should_end_ui() -> bool: metrics_output_sink.close() except BaseException as error: cleanup_failures.append(error) + if profiler is not None: + try: + profiler.close() + except BaseException as error: + cleanup_failures.append(error) if next_session_desc is None: try: window.close() @@ -364,6 +400,22 @@ def should_end_ui() -> bool: return next_session_desc +def _validate_profile_path(session_desc: SessionDesc, profile_path: Path) -> None: + """Keep the profile separate from an enabled chunk lifecycle trace.""" + metadata = session_desc.metadata + if metadata.get(_TRACE_METADATA_KEY) is not True: + return + trace_path = metadata.get(_TRACE_PATH_METADATA_KEY) + if ( + isinstance(trace_path, str | Path) + and Path(trace_path).expanduser().resolve() + == profile_path.expanduser().resolve() + ): + raise ValueError( + "Runtime profile and chunk lifecycle trace must use distinct paths." + ) + + def _open_chunk_trace(path_value: object) -> _ChunkTraceLog: """Open a line-buffered lifecycle trace for one session.""" if not isinstance(path_value, str | Path): diff --git a/flashdreams/flashdreams/runtime_v2/webrtc_client_window.py b/flashdreams/flashdreams/runtime_v2/webrtc_client_window.py index 218671520..ba5b3f5fc 100644 --- a/flashdreams/flashdreams/runtime_v2/webrtc_client_window.py +++ b/flashdreams/flashdreams/runtime_v2/webrtc_client_window.py @@ -35,6 +35,16 @@ class WebRTCClientWindow(IClientWindow): the application is explicitly stopped. """ + @property + def input_timestamp_origin_ns(self) -> int | None: + """Return the current session's monotonic input timestamp origin.""" + server_origin_ns = self.server.input_timestamp_origin_ns + if server_origin_ns is None: + return None + with self._input_lock: + session_event_offset_us = int(self._session_event_offset_us) + return server_origin_ns + session_event_offset_us * 1_000 + def __init__( self, *, @@ -66,7 +76,8 @@ def __init__( def handle_input(event: UserInputEvent) -> None: """Buffer one backend event for the ``InputSource`` protocol.""" - # TODO: do we really need to buffer all events? Some mouse moves may be superseded by later ones. + # TODO: do we need to buffer every event? Later mouse moves may + # supersede earlier ones. with self._input_lock: self._input_events.append(event) diff --git a/flashdreams/test_v2/test_application_runner.py b/flashdreams/test_v2/test_application_runner.py index 85d53c90a..d66baf55b 100644 --- a/flashdreams/test_v2/test_application_runner.py +++ b/flashdreams/test_v2/test_application_runner.py @@ -3,6 +3,7 @@ """CPU tests for the v2 application runner.""" +import json import logging from collections.abc import Sequence from dataclasses import replace @@ -16,6 +17,7 @@ from flashdreams.api_v2.loop import IModelLoop, IUILoop from flashdreams.api_v2.session import ISession from flashdreams.runtime_v2.application_runner import ApplicationRunner +from flashdreams.runtime_v2.runtime_profiler import RuntimeProfiler from flashdreams.runtime_v2.session_desc import PresentationMode, SessionDesc from flashdreams.runtime_v2.step_result import StepResult from flashdreams.runtime_v2.user_input_event import CloseUserInputEvent @@ -356,6 +358,21 @@ def test_application_runner_replaces_a_session_before_closing_the_window() -> No assert calls.count("application.close") == 1 +def test_application_runner_profiles_each_replacement_session(tmp_path) -> None: + calls: list[str] = [] + profile_path = tmp_path / "runtime.jsonl" + + ApplicationRunner( + _Application(calls, replace_first_session=True), + _SecondSessionClosingWindow(calls), + profiler=RuntimeProfiler(profile_path), + ).run(_session_desc()) + + records = [json.loads(line) for line in profile_path.read_text().splitlines()] + assert sum(record["phase"] == "session_started" for record in records) == 2 + assert sum(record["phase"] == "profile_summary" for record in records) == 4 + + def test_application_runner_closes_a_preserved_window_if_replacement_fails() -> None: calls: list[str] = [] diff --git a/flashdreams/test_v2/test_client_window_factory.py b/flashdreams/test_v2/test_client_window_factory.py index c4f9c67da..a59c3ff35 100644 --- a/flashdreams/test_v2/test_client_window_factory.py +++ b/flashdreams/test_v2/test_client_window_factory.py @@ -12,6 +12,7 @@ import pytest +from flashdreams.runtime_v2.cli import _parser, _validate_profile_path from flashdreams.runtime_v2.client_window_factory import ( add_client_window_arguments, client_window_mode, @@ -69,6 +70,26 @@ def test_a_native_window_mode_is_lazy_and_keeps_its_title() -> None: assert window.title == "World model" +@pytest.mark.parametrize("other_flag", ["--stats-path", "--output-path"]) +def test_profile_path_stays_distinct_from_run_outputs( + tmp_path: Path, + other_flag: str, +) -> None: + shared = tmp_path / "run-artifact" + arguments = [ + "demo", + "--output-path", + str(tmp_path / "output.mp4"), + "--profile-path", + str(shared), + other_flag, + str(shared), + ] + + with pytest.raises(ValueError, match="distinct output path"): + _validate_profile_path(_parser().parse_args(arguments)) + + class TestWebRTC: """Modes that serve a client, which need the serving stack installed.""" diff --git a/flashdreams/test_v2/test_native_window_client_window.py b/flashdreams/test_v2/test_native_window_client_window.py index 873ad73e2..fc588a451 100644 --- a/flashdreams/test_v2/test_native_window_client_window.py +++ b/flashdreams/test_v2/test_native_window_client_window.py @@ -336,6 +336,7 @@ def record_conversion(result: StepResult, desc: SessionDesc) -> torch.Tensor: ) window = NativeWindowClientWindow(presenter_factory=cast(Any, create_presenter)) window.open(_session_desc()) + assert window.input_timestamp_origin_ns is not None window.get_user_input_events() window.write(_result()) window.close() @@ -386,6 +387,7 @@ def test_native_window_reports_input_and_close_from_event_pump() -> None: clock_ns=lambda: next(clock_values), ) window.open(_session_desc()) + assert window.input_timestamp_origin_ns is None presenter.pending_events.put(("keyboard", _KeyboardEvent("up", pressed=True))) presenter.pending_events.put( ( diff --git a/flashdreams/test_v2/test_runtime_profiler.py b/flashdreams/test_v2/test_runtime_profiler.py new file mode 100644 index 000000000..4cd1b2af1 --- /dev/null +++ b/flashdreams/test_v2/test_runtime_profiler.py @@ -0,0 +1,244 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""CPU checks for V2 host-side input-latency profiles.""" + +import json + +import pytest +from numpy import uint64 + +from flashdreams.runtime_v2.runtime_profiler import RuntimeProfiler +from flashdreams.runtime_v2.session_desc import ( + BackpressureMode, + PresentationMode, + SessionDesc, +) +from flashdreams.runtime_v2.user_input_event import ( + KeyboardInputState, + KeyboardUserInputEvent, +) +from flashdreams.runtime_v2.user_input_events import UserInputEvents + +pytestmark = pytest.mark.ci_cpu + +_SESSION_DESC = SessionDesc( + backpressure_mode=BackpressureMode.DROP_OLDEST, + presentation_mode=PresentationMode.ON_DEMAND, + frames_per_second_for_ui=60, + frames_per_second_for_step=24, + video_width=640, + video_height=360, +) + + +def _input(timestamp_us: int = 1_000) -> UserInputEvents: + return UserInputEvents( + [ + KeyboardUserInputEvent( + timestamp=uint64(timestamp_us), + key="w", + state=KeyboardInputState.PRESSED, + ) + ] + ) + + +def test_profile_correlates_claimed_input_with_first_following_write(tmp_path) -> None: + path = tmp_path / "runtime.jsonl" + profiler = RuntimeProfiler(path) + profiler.session_started( + input_timestamp_origin_ns=1_000_000, + session_desc=_SESSION_DESC, + client_window_type="NativeWindowClientWindow", + time_ns=1_000_000, + ) + + profiler.ui_step_started( + _input(), + generation=2, + step=9, + time_ns=5_000_000, + ) + profiler.ui_step_started( + UserInputEvents([]), + generation=2, + step=10, + time_ns=6_000_000, + ) + profiler.window_write_completed( + generation=2, + ui_step=10, + time_ns=9_000_000, + ) + + assert profiler.summary() == { + "input_to_ui_step_s": { + "count": 1, + "median_s": pytest.approx(0.003), + "p90_s": pytest.approx(0.003), + "max_s": pytest.approx(0.003), + "quantile_sample_count": 1, + "quantiles_approximate": False, + }, + "input_to_window_write_s": { + "count": 1, + "median_s": pytest.approx(0.007), + "p90_s": pytest.approx(0.007), + "max_s": pytest.approx(0.007), + "quantile_sample_count": 1, + "quantiles_approximate": False, + }, + } + + profiler.close() + profiler.close() + records = [json.loads(line) for line in path.read_text().splitlines()] + output_record = next( + record for record in records if record["phase"] == "input_to_window_write" + ) + assert output_record == { + "artifact_type": "flashdreams.runtime_v2.input_latency_profile", + "schema_version": 1, + "phase": "input_to_window_write", + "time_ns": 9_000_000, + "generation": 2, + "claimed_ui_step": 9, + "presented_ui_step": 10, + "input_type": "keyboard", + "input_timestamp_us": 1_000, + "duration_s": 0.007, + } + assert sum(record["phase"] == "profile_summary" for record in records) == 2 + session_record = records[0] + assert session_record["measurement_endpoints"] == { + "input_to_ui_step_s": "ui_loop_begin_run", + "input_to_window_write_s": "client_window_write_return", + } + assert session_record["runtime_settings"] == { + "client_window_type": "NativeWindowClientWindow", + "output_layout": "tchw", + "frames_per_second_for_ui": 60, + "frames_per_second_for_step": 24, + "video_width": 640, + "video_height": 360, + "presentation_mode": "on_demand", + "backpressure_mode": "drop_oldest", + } + + +def test_profile_skips_sources_without_a_shared_clock_origin(tmp_path) -> None: + profiler = RuntimeProfiler(tmp_path / "runtime.jsonl") + profiler.session_started( + input_timestamp_origin_ns=None, + session_desc=_SESSION_DESC, + client_window_type="UnknownWindow", + time_ns=1_000_000, + ) + + profiler.ui_step_started(_input(), generation=0, step=0, time_ns=5_000_000) + profiler.window_write_completed(generation=0, ui_step=0, time_ns=9_000_000) + + assert profiler.summary() == { + "input_to_ui_step_s": {"count": 0}, + "input_to_window_write_s": {"count": 0}, + } + profiler.close() + + +def test_generation_change_discards_unpresented_input(tmp_path) -> None: + profiler = RuntimeProfiler(tmp_path / "runtime.jsonl") + profiler.session_started( + input_timestamp_origin_ns=0, + session_desc=_SESSION_DESC, + client_window_type="UnknownWindow", + time_ns=0, + ) + profiler.ui_step_started(_input(), generation=0, step=0, time_ns=2_000_000) + + profiler.ui_step_started( + UserInputEvents([]), + generation=1, + step=0, + time_ns=3_000_000, + ) + profiler.window_write_completed(generation=1, ui_step=0, time_ns=4_000_000) + + assert profiler.summary()["input_to_window_write_s"] == {"count": 0} + profiler.close() + + +def test_replacement_sessions_append_independent_profile_segments(tmp_path) -> None: + path = tmp_path / "runtime.jsonl" + profiler = RuntimeProfiler(path) + + for origin_ns in (1_000_000, 10_000_000): + profiler.session_started( + input_timestamp_origin_ns=origin_ns, + session_desc=_SESSION_DESC, + client_window_type="NativeWindowClientWindow", + time_ns=origin_ns, + ) + profiler.ui_step_started( + _input(), + generation=0, + step=0, + time_ns=origin_ns + 2_000_000, + ) + profiler.close() + + records = [json.loads(line) for line in path.read_text().splitlines()] + assert sum(record["phase"] == "session_started" for record in records) == 2 + summaries = [ + record + for record in records + if record["phase"] == "profile_summary" + and record["metric"] == "input_to_ui_step_s" + ] + assert [record["count"] for record in summaries] == [1, 1] + + +def test_profile_summary_bounds_quantile_storage(tmp_path) -> None: + profiler = RuntimeProfiler(tmp_path / "runtime.jsonl") + profiler.session_started( + input_timestamp_origin_ns=0, + session_desc=_SESSION_DESC, + client_window_type="WebRTCClientWindow", + time_ns=0, + ) + + for index in range(1_100): + profiler.ui_step_started( + _input(index), + generation=0, + step=index, + time_ns=index * 1_000 + 1, + ) + profiler.window_write_completed( + generation=0, + ui_step=index, + time_ns=index * 1_000 + 2, + ) + + for summary, maximum in zip( + profiler.summary().values(), + (1e-9, 2e-9), + strict=True, + ): + assert summary["count"] == 1_100 + assert summary["quantile_sample_count"] == 1_024 + assert summary["quantiles_approximate"] is True + assert summary["max_s"] == pytest.approx(maximum) + profiler.close() diff --git a/flashdreams/test_v2/test_session_runner.py b/flashdreams/test_v2/test_session_runner.py index 60ae42492..c5bd7ff66 100644 --- a/flashdreams/test_v2/test_session_runner.py +++ b/flashdreams/test_v2/test_session_runner.py @@ -3,6 +3,7 @@ """CPU test for the v2 session loop, independent of any application.""" +import json import logging import queue import threading @@ -20,7 +21,6 @@ IModelLoop, IUILoop, ModelInferenceState, - UILoopRequests, invoke_async, ) from flashdreams.api_v2.session import ISession @@ -34,6 +34,7 @@ PresentationManager, _PresentationClock, ) +from flashdreams.runtime_v2.runtime_profiler import RuntimeProfiler from flashdreams.runtime_v2.session_desc import ( BackpressureMode, PresentationMode, @@ -485,6 +486,7 @@ def __init__( self._hold_writes = hold_writes self._lock = threading.Lock() self.session_desc: SessionDesc | None = None + self._input_timestamp_origin_ns: int | None = None self.results: list[StepResult] = [] self.cursor_requests: list[tuple[str, bool]] = [] @@ -498,6 +500,10 @@ def request_lock_cursor_to_window(self, lock_cursor_to_window: bool) -> None: self._log.record("window.request_lock_cursor_to_window") self.cursor_requests.append(("lock", lock_cursor_to_window)) + @property + def input_timestamp_origin_ns(self) -> int | None: + return self._input_timestamp_origin_ns + def get_user_input_events(self) -> UserInputEvents: self._log.record("window.get_user_input_events") with self._lock: @@ -510,6 +516,7 @@ def open(self, session_desc: SessionDesc) -> None: if self._fail_to_open: raise RuntimeError("open failed") self.session_desc = session_desc + self._input_timestamp_origin_ns = time.monotonic_ns() def write(self, result: StepResult) -> None: if self._hold_writes is not None: @@ -575,6 +582,55 @@ def test_run_session_presents_every_step_in_order() -> None: assert steps == ["session.step(0)", "session.step(1)", "session.step(2)"] +def test_run_session_profiles_input_through_window_write(tmp_path) -> None: + log = CallLog() + session = FakeSession(_session_desc(), log) + window = RecordingClientWindow(log, [_key_event()]) + profile_path = tmp_path / "runtime.jsonl" + + run_session( + session, + window, + profiler=RuntimeProfiler(profile_path), + steps=1, + ) + + records = [json.loads(line) for line in profile_path.read_text().splitlines()] + phases = {record["phase"] for record in records} + assert { + "input_to_ui_step", + "input_to_window_write", + "profile_summary", + } <= phases + output_record = next( + record for record in records if record["phase"] == "input_to_window_write" + ) + assert output_record["input_type"] == "keyboard" + + +def test_run_session_keeps_profile_separate_from_chunk_trace(tmp_path) -> None: + log = CallLog() + shared_path = tmp_path / "runtime.jsonl" + session_desc = replace( + _session_desc(), + metadata={ + "trace_chunk_lifecycle": True, + "trace_chunk_lifecycle_path": str(shared_path), + }, + ) + + with pytest.raises(ValueError, match="must use distinct paths"): + run_session( + FakeSession(session_desc, log), + RecordingClientWindow(log, []), + profiler=RuntimeProfiler(shared_path), + steps=1, + ) + + assert log.calls == ["window.close", "session.close"] + assert not shared_path.exists() + + def test_run_session_opens_before_writing_and_closes_after() -> None: log = CallLog() session = FakeSession(_session_desc(), log) diff --git a/flashdreams/test_v2/test_webrtc_client_window.py b/flashdreams/test_v2/test_webrtc_client_window.py index 139fe628a..0bac803e3 100644 --- a/flashdreams/test_v2/test_webrtc_client_window.py +++ b/flashdreams/test_v2/test_webrtc_client_window.py @@ -647,6 +647,7 @@ def test_window_write_materializes_before_synchronous_sender_admission( source = torch.full((1, 3, 16, 16), 31, dtype=torch.uint8) try: window.open(_session_desc()) + assert window.input_timestamp_origin_ns is not None window.server._video_track = cast(Any, track) window.server._media_connected.set() with monkeypatch.context() as patch: @@ -829,6 +830,8 @@ def test_window_discards_events_buffered_during_a_session_handoff() -> None: window = WebRTCClientWindow() try: window.open(_session_desc()) + first_session_origin_ns = window.input_timestamp_origin_ns + assert first_session_origin_ns is not None time.sleep(0.01) window.server._buffer_browser_message( json.dumps({"type": "keyboard", "key": "w", "pressed": True}) @@ -838,6 +841,9 @@ def test_window_discards_events_buffered_during_a_session_handoff() -> None: window.server._buffer_browser_message(json.dumps({"type": "close"})) window.open(_session_desc()) + replacement_origin_ns = window.input_timestamp_origin_ns + assert replacement_origin_ns is not None + assert replacement_origin_ns > first_session_origin_ns window.server._buffer_browser_message( json.dumps({"type": "focus", "focused": True}) ) @@ -847,6 +853,10 @@ def test_window_discards_events_buffered_during_a_session_handoff() -> None: assert ( replacement_events[0].get_timestamp() < first_session_event.get_timestamp() ) + assert ( + first_session_origin_ns + int(first_session_event.get_timestamp()) * 1_000 + < replacement_origin_ns + int(replacement_events[0].get_timestamp()) * 1_000 + ) finally: window.close()