diff --git a/flashdreams/flashdreams/serving/realtime/timing.py b/flashdreams/flashdreams/serving/realtime/timing.py new file mode 100644 index 00000000..9891cfcd --- /dev/null +++ b/flashdreams/flashdreams/serving/realtime/timing.py @@ -0,0 +1,546 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Transport-neutral realtime timing records and trace helpers.""" + +from __future__ import annotations + +from collections import defaultdict, deque +from collections.abc import Iterable, Iterator, Mapping +from dataclasses import dataclass +from threading import Lock +from typing import Protocol + +TraceComponentValue = str | int | float | bool | None + + +def trace_time_ns(seconds: float) -> int: + return int(seconds * 1_000_000_000) + + +def event_dependencies(*events: int | None) -> list[int]: + return [event for event in events if event is not None] + + +def _duration_ms(start_time: float | None, end_time: float | None) -> float | None: + if start_time is None or end_time is None: + return None + return (end_time - start_time) * 1000.0 + + +@dataclass +class FrameTimes: + frame_index: int + intended_present_time: float + image_ready_time: float | None = None + sample_display_pose_time: float | None = None + present_time: float | None = None + + def input_to_present_ms(self, *, input_sample_time: float) -> float | None: + return _duration_ms(input_sample_time, self.present_time) + + def present_jitter_ms(self) -> float | None: + return _duration_ms(self.intended_present_time, self.present_time) + + +@dataclass(frozen=True) +class ChunkPrediction: + """Predicted timestamps for a chunk's pipeline stages.""" + + first_present: float + + @classmethod + def create( + cls, *, request_time: float, frame_interval_s: float + ) -> "ChunkPrediction": + return cls(first_present=request_time + frame_interval_s) + + +@dataclass +class ChunkTimes: + """Mutable timing record that travels with one realtime chunk.""" + + chunk_index: int + input_sample_time: float + request_time: float + request_poses_ready_time: float + frames: list[FrameTimes] + prediction: ChunkPrediction | None = None + chunk_render_start_time: float | None = None + chunk_ready_time: float | None = None + + @classmethod + def create( + cls, + chunk_index: int, + input_sample_time: float, + request_time: float, + request_poses_ready_time: float, + intended_present_times: list[float], + prediction: ChunkPrediction | None = None, + ) -> "ChunkTimes": + frames = [ + FrameTimes(frame_index=index, intended_present_time=time_value) + for index, time_value in enumerate(intended_present_times) + ] + return cls( + chunk_index=chunk_index, + input_sample_time=input_sample_time, + request_time=request_time, + request_poses_ready_time=request_poses_ready_time, + frames=frames, + prediction=prediction, + ) + + def stage_durations_ms(self) -> dict[str, float]: + """Return available milestone-derived durations for this chunk.""" + return chunk_stage_durations_ms(self) + + +@dataclass(frozen=True) +class VideoModelTimings: + """Observable backend timing milestones for one video-model chunk.""" + + condition_start_time: float + condition_ready_time: float + model_start_time: float + model_ready_time: float + merge_start_time: float + merge_ready_time: float + decode_start_time: float | None = None + decode_ready_time: float | None = None + cache_update_start_time: float | None = None + cache_update_ready_time: float | None = None + + def stage_durations_ms(self) -> dict[str, float]: + durations: dict[str, float] = {} + _add_duration( + durations, + "condition", + self.condition_start_time, + self.condition_ready_time, + ) + _add_duration(durations, "model", self.model_start_time, self.model_ready_time) + _add_duration( + durations, + "cache_update", + self.cache_update_start_time, + self.cache_update_ready_time, + ) + _add_duration( + durations, + "decode", + self.decode_start_time, + self.decode_ready_time, + ) + _add_duration(durations, "merge", self.merge_start_time, self.merge_ready_time) + _add_duration( + durations, "total", self.condition_start_time, self.merge_ready_time + ) + return durations + + +class ChunkHistory: + def __init__(self, capacity: int) -> None: + self._deque: deque[ChunkTimes] = deque(maxlen=capacity) + + def append(self, chunk: ChunkTimes) -> None: + self._deque.append(chunk) + + def __iter__(self) -> Iterator[ChunkTimes]: + return iter(self._deque) + + def __len__(self) -> int: + return len(self._deque) + + def summary(self) -> "RecentTimingSummary": + return summarize_chunk_history(self._deque) + + +class TraceSink(Protocol): + def add_thread(self, name: str) -> int: ... + + def add_instant( + self, + name: str, + *, + thread: int, + time_ns: int, + depends_on: list[int] | None = None, + **components: TraceComponentValue, + ) -> int: ... + + def add_range( + self, + name: str, + *, + thread: int, + begin_ns: int, + end_ns: int, + depends_on: list[int] | None = None, + **components: TraceComponentValue, + ) -> int: ... + + +@dataclass(frozen=True) +class TraceContext: + sink: TraceSink + main_thread: int + worker_thread: int + lock: Lock + + @classmethod + def create(cls, sink: TraceSink) -> "TraceContext": + return cls( + sink=sink, + main_thread=sink.add_thread("main"), + worker_thread=sink.add_thread("pipeline-worker"), + lock=Lock(), + ) + + def add_instant( + self, + name: str, + *, + thread: int, + time_ns: int, + depends_on: list[int] | None = None, + **components: TraceComponentValue, + ) -> int: + with self.lock: + return self.sink.add_instant( + name, + thread=thread, + time_ns=time_ns, + depends_on=depends_on, + **components, + ) + + def add_range( + self, + name: str, + *, + thread: int, + begin_ns: int, + end_ns: int, + depends_on: list[int] | None = None, + **components: TraceComponentValue, + ) -> int: + with self.lock: + return self.sink.add_range( + name, + thread=thread, + begin_ns=begin_ns, + end_ns=end_ns, + depends_on=depends_on, + **components, + ) + + +@dataclass(frozen=True) +class VideoModelTraceEvents: + condition_event_id: int + model_event_id: int + merge_event_id: int + cache_update_event_id: int | None = None + decode_event_id: int | None = None + + @property + def final_event_id(self) -> int: + return self.merge_event_id + + +def emit_video_model_timing_ranges( + trace_context: TraceContext, + *, + timings: VideoModelTimings, + thread: int, + depends_on: list[int] | None = None, + chunk_index: int, +) -> VideoModelTraceEvents: + """Emit standard trace ranges for backend-visible video-model stages.""" + condition_event = trace_context.add_range( + "condition_raster", + thread=thread, + begin_ns=trace_time_ns(timings.condition_start_time), + end_ns=trace_time_ns(timings.condition_ready_time), + depends_on=depends_on, + chunk_index=chunk_index, + ) + model_event = trace_context.add_range( + "model_generate", + thread=thread, + begin_ns=trace_time_ns(timings.model_start_time), + end_ns=trace_time_ns(timings.model_ready_time), + depends_on=event_dependencies(condition_event), + chunk_index=chunk_index, + ) + cache_update_event = _add_optional_trace_range( + trace_context, + "cache_update", + thread=thread, + begin_time=timings.cache_update_start_time, + end_time=timings.cache_update_ready_time, + depends_on=event_dependencies(model_event), + chunk_index=chunk_index, + ) + decode_event = _add_optional_trace_range( + trace_context, + "decode", + thread=thread, + begin_time=timings.decode_start_time, + end_time=timings.decode_ready_time, + depends_on=event_dependencies( + cache_update_event if cache_update_event is not None else model_event + ), + chunk_index=chunk_index, + ) + last_event = decode_event + if last_event is None: + last_event = cache_update_event + if last_event is None: + last_event = model_event + merge_event = trace_context.add_range( + "frame_merge", + thread=thread, + begin_ns=trace_time_ns(timings.merge_start_time), + end_ns=trace_time_ns(timings.merge_ready_time), + depends_on=event_dependencies(last_event), + chunk_index=chunk_index, + ) + return VideoModelTraceEvents( + condition_event_id=condition_event, + model_event_id=model_event, + cache_update_event_id=cache_update_event, + decode_event_id=decode_event, + merge_event_id=merge_event, + ) + + +def chunk_stage_durations_ms(chunk: ChunkTimes) -> dict[str, float]: + durations: dict[str, float] = {} + _add_duration( + durations, "input_to_request", chunk.input_sample_time, chunk.request_time + ) + _add_duration( + durations, + "request_to_poses_ready", + chunk.request_time, + chunk.request_poses_ready_time, + ) + _add_duration( + durations, + "queue_wait", + chunk.request_poses_ready_time, + chunk.chunk_render_start_time, + ) + _add_duration( + durations, + "chunk_render", + chunk.chunk_render_start_time, + chunk.chunk_ready_time, + ) + if chunk.frames: + first_frame = chunk.frames[0] + _add_duration( + durations, + "chunk_ready_to_first_image", + chunk.chunk_ready_time, + first_frame.image_ready_time, + ) + _add_duration( + durations, + "first_image_to_present", + first_frame.image_ready_time, + first_frame.present_time, + ) + _add_duration( + durations, + "input_to_first_present", + chunk.input_sample_time, + first_frame.present_time, + ) + return durations + + +@dataclass(frozen=True) +class StageDurationSummary: + count: int + avg_ms: float + min_ms: float + max_ms: float + median_ms: float + p90_ms: float + + +@dataclass(frozen=True) +class RecentTimingSummary: + chunk_count: int + stages: dict[str, StageDurationSummary] + + +def summarize_stage_durations( + samples: Iterable[Mapping[str, float]], +) -> dict[str, StageDurationSummary]: + grouped: dict[str, list[float]] = defaultdict(list) + for sample in samples: + for stage_name, duration_ms in sample.items(): + grouped[stage_name].append(float(duration_ms)) + return { + stage_name: _summarize_values(values) + for stage_name, values in sorted(grouped.items()) + } + + +def summarize_chunk_history(chunks: Iterable[ChunkTimes]) -> RecentTimingSummary: + chunk_list = list(chunks) + return RecentTimingSummary( + chunk_count=len(chunk_list), + stages=summarize_stage_durations( + chunk.stage_durations_ms() for chunk in chunk_list + ), + ) + + +class RollingChunkTimingSummary: + def __init__(self, capacity: int) -> None: + self._chunks: deque[ChunkTimes] = deque(maxlen=capacity) + + def append(self, chunk: ChunkTimes) -> None: + self._chunks.append(chunk) + + def reset(self) -> None: + self._chunks.clear() + + def summary(self) -> RecentTimingSummary: + return summarize_chunk_history(self._chunks) + + +@dataclass(frozen=True) +class InputToPresentSummary: + window_s: float + samples: int + wall_present_fps: float + avg_raw_control_to_present_ms: float + avg_adj_control_to_present_ms: float + + def log_message(self) -> str: + return ( + "[profile] e2e " + f"wall_present_fps={self.wall_present_fps:.1f} " + f"avg_adj_control_to_present_ms={self.avg_adj_control_to_present_ms:.2f} " + f"avg_raw_control_to_present_ms={self.avg_raw_control_to_present_ms:.2f} " + f"samples={self.samples}" + ) + + +class InputToPresentProfileWindow: + """Rolling wall-clock input-to-present summary window.""" + + def __init__(self, *, interval_s: float = 2.0) -> None: + self.interval_s = interval_s + self.reset() + + def reset(self, *, interval_s: float | None = None) -> None: + if interval_s is not None: + self.interval_s = interval_s + self._sum_raw_ms = 0.0 + self._sum_adj_ms = 0.0 + self._count = 0 + self._window_start: float | None = None + + def record( + self, + *, + present_time: float, + input_sample_time: float, + frame_index: int, + frame_interval_s: float, + ) -> InputToPresentSummary | None: + raw_ms = (present_time - input_sample_time) * 1000.0 + scheduled_ms = frame_index * (frame_interval_s * 1000.0) + adj_ms = raw_ms - scheduled_ms + self._sum_raw_ms += raw_ms + self._sum_adj_ms += adj_ms + self._count += 1 + if self._window_start is None: + self._window_start = present_time + + window_s = present_time - self._window_start + if window_s < self.interval_s: + return None + + samples = self._count + summary = InputToPresentSummary( + window_s=window_s, + samples=samples, + wall_present_fps=float(samples) / window_s if window_s > 1e-9 else 0.0, + avg_raw_control_to_present_ms=self._sum_raw_ms / float(samples), + avg_adj_control_to_present_ms=self._sum_adj_ms / float(samples), + ) + self._sum_raw_ms = 0.0 + self._sum_adj_ms = 0.0 + self._count = 0 + self._window_start = present_time + return summary + + +def _add_duration( + durations: dict[str, float], + name: str, + start_time: float | None, + end_time: float | None, +) -> None: + duration = _duration_ms(start_time, end_time) + if duration is not None: + durations[name] = duration + + +def _add_optional_trace_range( + trace_context: TraceContext, + name: str, + *, + thread: int, + begin_time: float | None, + end_time: float | None, + depends_on: list[int] | None, + chunk_index: int, +) -> int | None: + if begin_time is None or end_time is None: + return None + return trace_context.add_range( + name, + thread=thread, + begin_ns=trace_time_ns(begin_time), + end_ns=trace_time_ns(end_time), + depends_on=depends_on, + chunk_index=chunk_index, + ) + + +def _summarize_values(values: list[float]) -> StageDurationSummary: + ordered = sorted(values) + count = len(ordered) + if count <= 0: + raise ValueError("Cannot summarize an empty stage duration list.") + return StageDurationSummary( + count=count, + avg_ms=sum(ordered) / float(count), + min_ms=ordered[0], + max_ms=ordered[-1], + median_ms=_percentile_sorted(ordered, 0.5), + p90_ms=_percentile_sorted(ordered, 0.9), + ) + + +def _percentile_sorted(values: list[float], percentile: float) -> float: + if not values: + raise ValueError("Cannot compute percentile of an empty list.") + if len(values) == 1: + return values[0] + clamped = min(1.0, max(0.0, percentile)) + index = clamped * (len(values) - 1) + lower_index = int(index) + upper_index = min(lower_index + 1, len(values) - 1) + fraction = index - lower_index + return values[lower_index] * (1.0 - fraction) + values[upper_index] * fraction diff --git a/flashdreams/tests/test_realtime_timing.py b/flashdreams/tests/test_realtime_timing.py new file mode 100644 index 00000000..623b31af --- /dev/null +++ b/flashdreams/tests/test_realtime_timing.py @@ -0,0 +1,326 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from collections.abc import Iterator +from dataclasses import dataclass + +import pytest + +from flashdreams.serving.realtime.timing import ( + ChunkHistory, + ChunkTimes, + InputToPresentProfileWindow, + TraceComponentValue, + TraceContext, + VideoModelTimings, + chunk_stage_durations_ms, + emit_video_model_timing_ranges, + summarize_chunk_history, +) + +pytestmark = pytest.mark.ci_cpu + + +@dataclass(frozen=True) +class _TraceEvent: + name: str + depends_on: list[int] + components: dict[str, TraceComponentValue] + + +class _RecordingTraceSink: + def __init__(self) -> None: + self.threads: list[str] = [] + self.events: list[_TraceEvent] = [] + + def add_thread(self, name: str) -> int: + self.threads.append(name) + return len(self.threads) - 1 + + def add_instant( + self, + name: str, + *, + thread: int, + time_ns: int, + depends_on: list[int] | None = None, + **components: TraceComponentValue, + ) -> int: + del thread, time_ns + return self._append_event(name, depends_on, components) + + def add_range( + self, + name: str, + *, + thread: int, + begin_ns: int, + end_ns: int, + depends_on: list[int] | None = None, + **components: TraceComponentValue, + ) -> int: + del thread + event_components = dict(components) + event_components["begin_ns"] = begin_ns + event_components["end_ns"] = end_ns + return self._append_event(name, depends_on, event_components) + + def _append_event( + self, + name: str, + depends_on: list[int] | None, + components: dict[str, TraceComponentValue], + ) -> int: + self.events.append( + _TraceEvent( + name=name, + depends_on=[] if depends_on is None else depends_on, + components=components, + ) + ) + return len(self.events) - 1 + + +class _NamedIdTraceSink(_RecordingTraceSink): + def __init__(self, event_ids: dict[str, int]) -> None: + super().__init__() + self._event_ids = event_ids + + def _append_event( + self, + name: str, + depends_on: list[int] | None, + components: dict[str, TraceComponentValue], + ) -> int: + super()._append_event(name, depends_on, components) + return self._event_ids[name] + + +def _chunk_times() -> ChunkTimes: + chunk = ChunkTimes.create( + chunk_index=7, + input_sample_time=1.0, + request_time=1.1, + request_poses_ready_time=1.2, + intended_present_times=[1.5, 1.6], + ) + chunk.chunk_render_start_time = 1.25 + chunk.chunk_ready_time = 1.45 + chunk.frames[0].image_ready_time = 1.47 + chunk.frames[0].sample_display_pose_time = 1.49 + chunk.frames[0].present_time = 1.52 + return chunk + + +def test_chunk_times_create_allocates_frame_times() -> None: + chunk = ChunkTimes.create( + chunk_index=0, + input_sample_time=1.0, + request_time=1.0, + request_poses_ready_time=1.001, + intended_present_times=[1.5, 1.6, 1.7], + ) + + assert [frame.frame_index for frame in chunk.frames] == [0, 1, 2] + assert [frame.intended_present_time for frame in chunk.frames] == [1.5, 1.6, 1.7] + + +def test_chunk_history_iter_returns_iterator() -> None: + first = _chunk_times() + second = _chunk_times() + history = ChunkHistory(capacity=2) + history.append(first) + history.append(second) + + iterator = iter(history) + + assert isinstance(iterator, Iterator) + assert next(iterator) is first + assert next(iterator) is second + + +def test_chunk_stage_durations_are_derived_from_milestones() -> None: + durations = chunk_stage_durations_ms(_chunk_times()) + + assert durations["input_to_request"] == pytest.approx(100.0) + assert durations["request_to_poses_ready"] == pytest.approx(100.0) + assert durations["queue_wait"] == pytest.approx(50.0) + assert durations["chunk_render"] == pytest.approx(200.0) + assert durations["chunk_ready_to_first_image"] == pytest.approx(20.0) + assert durations["first_image_to_present"] == pytest.approx(50.0) + assert durations["input_to_first_present"] == pytest.approx(520.0) + + +def test_summarize_chunk_history_builds_stage_statistics() -> None: + first = _chunk_times() + second = _chunk_times() + second.chunk_render_start_time = 2.0 + second.chunk_ready_time = 2.4 + + summary = summarize_chunk_history([first, second]) + + assert summary.chunk_count == 2 + assert summary.stages["chunk_render"].count == 2 + assert summary.stages["chunk_render"].avg_ms == pytest.approx(300.0) + assert summary.stages["chunk_render"].p90_ms == pytest.approx(380.0) + + +def test_video_model_timings_include_optional_decode_and_cache_durations() -> None: + timings = VideoModelTimings( + condition_start_time=1.0, + condition_ready_time=1.1, + model_start_time=1.1, + model_ready_time=1.6, + merge_start_time=1.65, + merge_ready_time=1.7, + cache_update_start_time=1.2, + cache_update_ready_time=1.3, + decode_start_time=1.4, + decode_ready_time=1.55, + ) + + durations = timings.stage_durations_ms() + + assert durations["condition"] == pytest.approx(100.0) + assert durations["model"] == pytest.approx(500.0) + assert durations["cache_update"] == pytest.approx(100.0) + assert durations["decode"] == pytest.approx(150.0) + assert durations["merge"] == pytest.approx(50.0) + assert durations["total"] == pytest.approx(700.0) + + +def test_emit_video_model_timing_ranges_adds_optional_subranges() -> None: + sink = _RecordingTraceSink() + trace_context = TraceContext.create(sink) + timings = VideoModelTimings( + condition_start_time=1.0, + condition_ready_time=1.1, + model_start_time=1.1, + model_ready_time=1.6, + merge_start_time=1.65, + merge_ready_time=1.7, + cache_update_start_time=1.2, + cache_update_ready_time=1.3, + decode_start_time=1.4, + decode_ready_time=1.55, + ) + + events = emit_video_model_timing_ranges( + trace_context, + timings=timings, + thread=trace_context.worker_thread, + depends_on=[123], + chunk_index=9, + ) + + names = [event.name for event in sink.events] + assert names == [ + "condition_raster", + "model_generate", + "cache_update", + "decode", + "frame_merge", + ] + assert sink.events[0].depends_on == [123] + assert sink.events[2].depends_on == [events.model_event_id] + assert sink.events[3].depends_on == [events.cache_update_event_id] + assert sink.events[4].depends_on == [events.decode_event_id] + assert events.final_event_id == events.merge_event_id + + +def test_emit_video_model_timing_ranges_keeps_zero_valued_event_ids() -> None: + sink = _NamedIdTraceSink( + { + "condition_raster": 10, + "model_generate": 11, + "cache_update": 0, + "decode": 12, + "frame_merge": 13, + } + ) + trace_context = TraceContext.create(sink) + timings = VideoModelTimings( + condition_start_time=1.0, + condition_ready_time=1.1, + model_start_time=1.1, + model_ready_time=1.6, + merge_start_time=1.65, + merge_ready_time=1.7, + cache_update_start_time=1.2, + cache_update_ready_time=1.3, + decode_start_time=1.4, + decode_ready_time=1.55, + ) + + events = emit_video_model_timing_ranges( + trace_context, + timings=timings, + thread=trace_context.worker_thread, + chunk_index=9, + ) + + assert events.cache_update_event_id == 0 + assert sink.events[3].depends_on == [0] + + +def test_emit_video_model_timing_ranges_uses_zero_decode_event_for_merge() -> None: + sink = _NamedIdTraceSink( + { + "condition_raster": 10, + "model_generate": 11, + "cache_update": 12, + "decode": 0, + "frame_merge": 13, + } + ) + trace_context = TraceContext.create(sink) + timings = VideoModelTimings( + condition_start_time=1.0, + condition_ready_time=1.1, + model_start_time=1.1, + model_ready_time=1.6, + merge_start_time=1.65, + merge_ready_time=1.7, + cache_update_start_time=1.2, + cache_update_ready_time=1.3, + decode_start_time=1.4, + decode_ready_time=1.55, + ) + + events = emit_video_model_timing_ranges( + trace_context, + timings=timings, + thread=trace_context.worker_thread, + chunk_index=9, + ) + + assert events.decode_event_id == 0 + assert sink.events[4].depends_on == [0] + + +def test_input_to_present_profile_window_returns_summary_on_interval() -> None: + window = InputToPresentProfileWindow(interval_s=0.25) + + assert ( + window.record( + present_time=1.0, + input_sample_time=0.9, + frame_index=0, + frame_interval_s=0.1, + ) + is None + ) + summary = window.record( + present_time=1.3, + input_sample_time=0.9, + frame_index=1, + frame_interval_s=0.1, + ) + + assert summary is not None + assert summary.samples == 2 + assert summary.avg_raw_control_to_present_ms == pytest.approx(250.0) + assert summary.avg_adj_control_to_present_ms == pytest.approx(200.0) + assert "avg_raw_control_to_present_ms=250.00" in summary.log_message() diff --git a/integrations/omnidreams/omnidreams/interactive_drive/app.py b/integrations/omnidreams/omnidreams/interactive_drive/app.py index 8ac0b0ff..0fccebab 100644 --- a/integrations/omnidreams/omnidreams/interactive_drive/app.py +++ b/integrations/omnidreams/omnidreams/interactive_drive/app.py @@ -21,7 +21,6 @@ PresenterBackend, run_main_loop, ) -from omnidreams.interactive_drive.runtime.timing import TraceContext, TraceSink from omnidreams.interactive_drive.scene_loader import ( load_scene_bundle, reseed_scene_bundle, @@ -42,6 +41,8 @@ from omnidreams.interactive_drive.video_model.chunk_pipeline import ChunkPipeline from omnidreams.interactive_drive.video_model.local import LocalVideoModelAdapter +from flashdreams.serving.realtime.timing import TraceContext, TraceSink + # Cadence for the event-pump loop that keeps the presenter alive while a # scene parses on a background thread. ~60 Hz keeps input latency low and # the loading indicator smooth without burning a core. diff --git a/integrations/omnidreams/omnidreams/interactive_drive/cli.py b/integrations/omnidreams/omnidreams/interactive_drive/cli.py index f4149bbd..2313aabe 100644 --- a/integrations/omnidreams/omnidreams/interactive_drive/cli.py +++ b/integrations/omnidreams/omnidreams/interactive_drive/cli.py @@ -21,11 +21,12 @@ WorldModelProfileConfig, ) from omnidreams.interactive_drive.log import configure_logging -from omnidreams.interactive_drive.runtime.timing import TraceSink from omnidreams.interactive_drive.synthetic_scene import build_synthetic_scene_to_temp from omnidreams.interactive_drive.world_model.manifest import load_world_model_manifest from omnidreams.scenes import local_scene_archive_path +from flashdreams.serving.realtime.timing import TraceSink + # Package root (from this file's location) so packaged-asset defaults below # resolve relative to the install, not the user's cwd. Bundled configs live at # ``interactive_drive/configs/``; scene USDZs are staged into diff --git a/integrations/omnidreams/omnidreams/interactive_drive/runtime/loop.py b/integrations/omnidreams/omnidreams/interactive_drive/runtime/loop.py index 581b0770..de553937 100644 --- a/integrations/omnidreams/omnidreams/interactive_drive/runtime/loop.py +++ b/integrations/omnidreams/omnidreams/interactive_drive/runtime/loop.py @@ -12,32 +12,31 @@ from loguru import logger from omnidreams.interactive_drive.input.backend import InputBackend from omnidreams.interactive_drive.runtime.runtime_controls import RuntimeControls -from omnidreams.interactive_drive.runtime.timing import ( +from omnidreams.interactive_drive.simulation.backend import SimulationBackend +from omnidreams.interactive_drive.types import DriverCommand, PresentedFrame +from omnidreams.interactive_drive.video_model.chunk_pipeline import ( + ChunkPipeline, + ChunkRequest, + QueuedFrame, +) + +from flashdreams.serving.realtime.timing import ( ChunkHistory, ChunkPrediction, ChunkTimes, + InputToPresentProfileWindow, TraceComponentValue, TraceContext, event_dependencies, trace_time_ns, ) -from omnidreams.interactive_drive.simulation.backend import SimulationBackend -from omnidreams.interactive_drive.types import DriverCommand, PresentedFrame -from omnidreams.interactive_drive.video_model.chunk_pipeline import ( - ChunkPipeline, - ChunkRequest, - QueuedFrame, -) _PROFILE_INPUT_TO_PRESENT_ENV = "INTERACTIVE_DRIVE_PROFILE_INPUT_TO_PRESENT" _PROFILE_INPUT_TO_PRESENT_INTERVAL_S_ENV = ( "INTERACTIVE_DRIVE_PROFILE_INPUT_TO_PRESENT_INTERVAL_S" ) -_PROFILE_E2E_SUM_RAW_MS: float = 0.0 -_PROFILE_E2E_SUM_ADJ_MS: float = 0.0 -_PROFILE_E2E_COUNT: int = 0 -_PROFILE_E2E_WINDOW_START: float | None = None +_PROFILE_E2E_WINDOW = InputToPresentProfileWindow() def _profile_input_to_present_enabled() -> bool: @@ -56,15 +55,7 @@ def _profile_input_to_present_interval_s() -> float: def reset_input_to_present_profile_window() -> None: """Clear accumulated e2e samples when the main loop starts.""" - global _PROFILE_E2E_SUM_RAW_MS - global _PROFILE_E2E_SUM_ADJ_MS - global _PROFILE_E2E_COUNT - global _PROFILE_E2E_WINDOW_START - - _PROFILE_E2E_SUM_RAW_MS = 0.0 - _PROFILE_E2E_SUM_ADJ_MS = 0.0 - _PROFILE_E2E_COUNT = 0 - _PROFILE_E2E_WINDOW_START = None + _PROFILE_E2E_WINDOW.reset(interval_s=_profile_input_to_present_interval_s()) def _chunk_frame_interval_s(chunk_times: ChunkTimes) -> float: @@ -83,42 +74,15 @@ def _record_input_to_present_for_profile( frame_index: int, frame_interval_s: float, ) -> None: - global _PROFILE_E2E_SUM_RAW_MS - global _PROFILE_E2E_SUM_ADJ_MS - global _PROFILE_E2E_COUNT - global _PROFILE_E2E_WINDOW_START - - raw_ms = (present_time - input_sample_time) * 1000.0 - scheduled_ms = frame_index * (frame_interval_s * 1000.0) - adj_ms = raw_ms - scheduled_ms - _PROFILE_E2E_SUM_RAW_MS += raw_ms - _PROFILE_E2E_SUM_ADJ_MS += adj_ms - _PROFILE_E2E_COUNT += 1 - if _PROFILE_E2E_WINDOW_START is None: - _PROFILE_E2E_WINDOW_START = present_time - - interval_s = _profile_input_to_present_interval_s() - if present_time - _PROFILE_E2E_WINDOW_START < interval_s: - return - - count = _PROFILE_E2E_COUNT - if count <= 0: - return - window_s = present_time - _PROFILE_E2E_WINDOW_START - wall_present_fps = float(count) / window_s if window_s > 1e-9 else 0.0 - avg_raw_ms = _PROFILE_E2E_SUM_RAW_MS / float(count) - avg_adj_ms = _PROFILE_E2E_SUM_ADJ_MS / float(count) - logger.info( - "[profile] e2e " - f"wall_present_fps={wall_present_fps:.1f} " - f"avg_adj_control_to_present_ms={avg_adj_ms:.2f} " - f"avg_raw_control_to_present_ms={avg_raw_ms:.2f} " - f"samples={count}", + _PROFILE_E2E_WINDOW.interval_s = _profile_input_to_present_interval_s() + summary = _PROFILE_E2E_WINDOW.record( + present_time=present_time, + input_sample_time=input_sample_time, + frame_index=frame_index, + frame_interval_s=frame_interval_s, ) - _PROFILE_E2E_SUM_RAW_MS = 0.0 - _PROFILE_E2E_SUM_ADJ_MS = 0.0 - _PROFILE_E2E_COUNT = 0 - _PROFILE_E2E_WINDOW_START = present_time + if summary is not None: + logger.info(summary.log_message()) class PresenterBackend(Protocol): diff --git a/integrations/omnidreams/omnidreams/interactive_drive/runtime/timing.py b/integrations/omnidreams/omnidreams/interactive_drive/runtime/timing.py index 2fa0dea3..cb1b0af4 100644 --- a/integrations/omnidreams/omnidreams/interactive_drive/runtime/timing.py +++ b/integrations/omnidreams/omnidreams/interactive_drive/runtime/timing.py @@ -1,186 +1,50 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -"""Timing records for latency measurement. - -``FrameTimes`` / ``ChunkTimes`` are mutable on purpose: a single instance -travels through the pipeline accumulating per-stage timestamps, so the record -created at request time is the same one that records present time (direct -correlation, no copying or re-association). -""" - -from collections import deque -from dataclasses import dataclass -from threading import Lock -from typing import Protocol - -TraceComponentValue = str | int | float | bool | None - - -def trace_time_ns(seconds: float) -> int: - return int(seconds * 1_000_000_000) - - -def event_dependencies(*events: int | None) -> list[int]: - return [event for event in events if event is not None] - - -@dataclass -class FrameTimes: - frame_index: int - intended_present_time: float - image_ready_time: float | None = None - sample_display_pose_time: float | None = None - present_time: float | None = None - - -@dataclass(frozen=True) -class ChunkPrediction: - """Predicted timestamps for a chunk's pipeline stages. - - Stage 1 only predicts ``first_present`` (when the chunk's first frame - will reach the screen). The Stage 2 design adds intermediate - EMA-summed milestones (``request -> render_start -> chunk_ready -> - decode_first -> first_present``) per ``alpasim.frame_timing``; new - fields land here as named attributes when that work arrives. - - Use :meth:`create` rather than constructing directly: the prediction - formula lives there so it stays beside the data it produces. - """ - - first_present: float - - @classmethod - def create( - cls, *, request_time: float, frame_interval_s: float - ) -> "ChunkPrediction": - """Stage 1 prediction: ``first_present = request_time + frame_interval_s``. - - Placeholder for the Stage 2 EMA-summed prediction chain; the - signature stays the same so callers don't change when Stage 2 - lands and the body grows to consume EMA latency stats. - """ - return cls(first_present=request_time + frame_interval_s) - - -@dataclass -class ChunkTimes: - chunk_index: int - input_sample_time: float - request_time: float - request_poses_ready_time: float - frames: list[FrameTimes] - prediction: ChunkPrediction | None = None - chunk_render_start_time: float | None = None - chunk_ready_time: float | None = None - - @classmethod - def create( - cls, - chunk_index: int, - input_sample_time: float, - request_time: float, - request_poses_ready_time: float, - intended_present_times: list[float], - prediction: ChunkPrediction | None = None, - ) -> "ChunkTimes": - frames = [ - FrameTimes(frame_index=index, intended_present_time=time_value) - for index, time_value in enumerate(intended_present_times) - ] - return cls( - chunk_index=chunk_index, - input_sample_time=input_sample_time, - request_time=request_time, - request_poses_ready_time=request_poses_ready_time, - frames=frames, - prediction=prediction, - ) - - -class ChunkHistory: - def __init__(self, capacity: int) -> None: - self._deque: deque[ChunkTimes] = deque(maxlen=capacity) - - def append(self, chunk: ChunkTimes) -> None: - self._deque.append(chunk) - - -class TraceSink(Protocol): - def add_thread(self, name: str) -> int: ... - - def add_instant( - self, - name: str, - *, - thread: int, - time_ns: int, - depends_on: list[int] | None = None, - **components: TraceComponentValue, - ) -> int: ... - - def add_range( - self, - name: str, - *, - thread: int, - begin_ns: int, - end_ns: int, - depends_on: list[int] | None = None, - **components: TraceComponentValue, - ) -> int: ... - - -@dataclass(frozen=True) -class TraceContext: - sink: TraceSink - main_thread: int - worker_thread: int - lock: Lock - - @classmethod - def create(cls, sink: TraceSink) -> "TraceContext": - return cls( - sink=sink, - main_thread=sink.add_thread("main"), - worker_thread=sink.add_thread("pipeline-worker"), - lock=Lock(), - ) - - def add_instant( - self, - name: str, - *, - thread: int, - time_ns: int, - depends_on: list[int] | None = None, - **components: TraceComponentValue, - ) -> int: - with self.lock: - return self.sink.add_instant( - name, - thread=thread, - time_ns=time_ns, - depends_on=depends_on, - **components, - ) - - def add_range( - self, - name: str, - *, - thread: int, - begin_ns: int, - end_ns: int, - depends_on: list[int] | None = None, - **components: TraceComponentValue, - ) -> int: - with self.lock: - return self.sink.add_range( - name, - thread=thread, - begin_ns=begin_ns, - end_ns=end_ns, - depends_on=depends_on, - **components, - ) +"""Compatibility exports for shared realtime timing helpers.""" + +from flashdreams.serving.realtime.timing import ( + ChunkHistory, + ChunkPrediction, + ChunkTimes, + FrameTimes, + InputToPresentProfileWindow, + InputToPresentSummary, + RecentTimingSummary, + RollingChunkTimingSummary, + StageDurationSummary, + TraceComponentValue, + TraceContext, + TraceSink, + VideoModelTimings, + VideoModelTraceEvents, + chunk_stage_durations_ms, + emit_video_model_timing_ranges, + event_dependencies, + summarize_chunk_history, + summarize_stage_durations, + trace_time_ns, +) + +__all__ = [ + "ChunkHistory", + "ChunkPrediction", + "ChunkTimes", + "FrameTimes", + "InputToPresentProfileWindow", + "InputToPresentSummary", + "RecentTimingSummary", + "RollingChunkTimingSummary", + "StageDurationSummary", + "TraceComponentValue", + "TraceContext", + "TraceSink", + "VideoModelTimings", + "VideoModelTraceEvents", + "chunk_stage_durations_ms", + "emit_video_model_timing_ranges", + "event_dependencies", + "summarize_chunk_history", + "summarize_stage_durations", + "trace_time_ns", +] diff --git a/integrations/omnidreams/omnidreams/interactive_drive/types.py b/integrations/omnidreams/omnidreams/interactive_drive/types.py index 1ffbabfc..efac4f5e 100644 --- a/integrations/omnidreams/omnidreams/interactive_drive/types.py +++ b/integrations/omnidreams/omnidreams/interactive_drive/types.py @@ -11,6 +11,8 @@ import numpy as np import numpy.typing as npt +from flashdreams.serving.realtime.timing import VideoModelTimings + FloatArray = npt.NDArray[np.float32] UInt8Array = npt.NDArray[np.uint8] Int32Array = npt.NDArray[np.int32] @@ -207,16 +209,6 @@ class PresentedFrame: status_message: str | None = None -@dataclass(frozen=True) -class VideoModelTimings: - condition_start_time: float - condition_ready_time: float - model_start_time: float - model_ready_time: float - merge_start_time: float - merge_ready_time: float - - @dataclass(frozen=True) class FrameChunk: frames: tuple[PresentedFrame, ...] diff --git a/integrations/omnidreams/omnidreams/interactive_drive/video_model/chunk_pipeline.py b/integrations/omnidreams/omnidreams/interactive_drive/video_model/chunk_pipeline.py index ec45e081..a1799fcf 100644 --- a/integrations/omnidreams/omnidreams/interactive_drive/video_model/chunk_pipeline.py +++ b/integrations/omnidreams/omnidreams/interactive_drive/video_model/chunk_pipeline.py @@ -9,12 +9,6 @@ from typing import Protocol from loguru import logger -from omnidreams.interactive_drive.runtime.timing import ( - ChunkTimes, - TraceContext, - event_dependencies, - trace_time_ns, -) from omnidreams.interactive_drive.types import ( FrameChunk, PresentedFrame, @@ -22,6 +16,14 @@ TrajectoryChunk, ) +from flashdreams.serving.realtime.timing import ( + ChunkTimes, + TraceContext, + emit_video_model_timing_ranges, + event_dependencies, + trace_time_ns, +) + class VideoModelBackend(Protocol): """Video-model interface called from the pipeline worker thread. @@ -238,30 +240,14 @@ def render_command(backend: VideoModelBackend) -> bool: worker_ready_event_id = chunk_render_event timings = frame_chunk.video_model_timings if timings is not None: - condition_event = trace_context.add_range( - "condition_raster", + timing_events = emit_video_model_timing_ranges( + trace_context, + timings=timings, thread=trace_context.worker_thread, - begin_ns=trace_time_ns(timings.condition_start_time), - end_ns=trace_time_ns(timings.condition_ready_time), depends_on=event_dependencies(queue_wait_event), chunk_index=chunk_times.chunk_index, ) - model_event = trace_context.add_range( - "model_generate", - thread=trace_context.worker_thread, - begin_ns=trace_time_ns(timings.model_start_time), - end_ns=trace_time_ns(timings.model_ready_time), - depends_on=event_dependencies(condition_event), - chunk_index=chunk_times.chunk_index, - ) - worker_ready_event_id = trace_context.add_range( - "frame_merge", - thread=trace_context.worker_thread, - begin_ns=trace_time_ns(timings.merge_start_time), - end_ns=trace_time_ns(timings.merge_ready_time), - depends_on=event_dependencies(model_event), - chunk_index=chunk_times.chunk_index, - ) + worker_ready_event_id = timing_events.final_event_id # Drop the output if a reset / scene switch superseded this chunk # while it was queued or rendering -- its frames belong to a # rollout the user has already moved on from. diff --git a/integrations/omnidreams/tests/interactive_drive/test_chunk_pipeline.py b/integrations/omnidreams/tests/interactive_drive/test_chunk_pipeline.py index 5f1379f9..02c327a2 100644 --- a/integrations/omnidreams/tests/interactive_drive/test_chunk_pipeline.py +++ b/integrations/omnidreams/tests/interactive_drive/test_chunk_pipeline.py @@ -12,17 +12,18 @@ make_trajectory, minimal_scene, ) -from omnidreams.interactive_drive.runtime.timing import ( - ChunkTimes, - TraceComponentValue, - TraceContext, -) from omnidreams.interactive_drive.types import FrameChunk, PresentedFrame, SceneBundle from omnidreams.interactive_drive.video_model.chunk_pipeline import ( ChunkPipeline, ChunkRequest, ) +from flashdreams.serving.realtime.timing import ( + ChunkTimes, + TraceComponentValue, + TraceContext, +) + @dataclass(frozen=True) class _TraceEvent: diff --git a/integrations/omnidreams/tests/interactive_drive/test_latency_loop.py b/integrations/omnidreams/tests/interactive_drive/test_latency_loop.py index f7140b20..190b6fba 100644 --- a/integrations/omnidreams/tests/interactive_drive/test_latency_loop.py +++ b/integrations/omnidreams/tests/interactive_drive/test_latency_loop.py @@ -20,12 +20,6 @@ present_queued_frame, run_main_loop, ) -from omnidreams.interactive_drive.runtime.timing import ( - ChunkPrediction, - ChunkTimes, - TraceComponentValue, - TraceContext, -) from omnidreams.interactive_drive.types import ( DriverCommand, PresentedFrame, @@ -37,6 +31,13 @@ QueuedFrame, ) +from flashdreams.serving.realtime.timing import ( + ChunkPrediction, + ChunkTimes, + TraceComponentValue, + TraceContext, +) + def _on_ci() -> bool: """True when running under CI (GitHub Actions et al. set ``CI=true``).""" diff --git a/integrations/omnidreams/tests/interactive_drive/test_latency_timing.py b/integrations/omnidreams/tests/interactive_drive/test_latency_timing.py index feacd19e..bdefc5f0 100644 --- a/integrations/omnidreams/tests/interactive_drive/test_latency_timing.py +++ b/integrations/omnidreams/tests/interactive_drive/test_latency_timing.py @@ -3,7 +3,7 @@ import time -from omnidreams.interactive_drive.runtime.timing import ChunkTimes +from flashdreams.serving.realtime.timing import ChunkTimes def _make_chunk(chunk_index: int = 0, chunk_size: int = 4) -> ChunkTimes: