diff --git a/scripts/bench_episode_byte_cache.py b/scripts/bench_episode_byte_cache.py index a6f9b5549..4b64432c9 100644 --- a/scripts/bench_episode_byte_cache.py +++ b/scripts/bench_episode_byte_cache.py @@ -618,6 +618,8 @@ def _print_range_timing_summary(fetch_pool: dict[str, float]) -> None: ("range_header_s", "http response headers"), ("range_first_byte_s", "http first body byte"), ("range_body_s", "http body drain"), + ("range_chunk_gap_s", "http chunk wait"), + ("range_join_s", "join response chunks"), ("range_retry_sleep_s", "http retry sleep"), ): value = fetch_pool.get(key) @@ -635,6 +637,14 @@ def _print_range_timing_summary(fetch_pool: dict[str, float]) -> None: if status_counts: summary = ", ".join(f"{status}={count:.0f}" for status, count in sorted(status_counts.items())) print(f"| http status counts | {summary} |") + chunks = fetch_pool.get("range_chunks", 0.0) + if chunks > 0: + bytes_read = fetch_pool.get("range_bytes", 0.0) + body_s = fetch_pool.get("range_body_s", 0.0) + print(f"| http chunks/range | {chunks / range_jobs:.1f} |") + print(f"| http avg KiB/chunk | {bytes_read / chunks / 1024:.1f} |") + if body_s > 0: + print(f"| http body MiB/s | {bytes_read / body_s / 1024**2:.1f} |") print(f"| range reads | {range_jobs:.0f} |") print(f"| avg MiB/range | {fetch_pool.get('range_bytes', 0.0) / range_jobs / 1024**2:.1f} |") diff --git a/src/lerobot/datasets/episode_video_streaming.py b/src/lerobot/datasets/episode_video_streaming.py index 503e5ca31..4f4b8a47f 100644 --- a/src/lerobot/datasets/episode_video_streaming.py +++ b/src/lerobot/datasets/episode_video_streaming.py @@ -387,20 +387,32 @@ class NativeHTTPRangeFetcher: chunks = [] first_byte_s = 0.0 first_chunk = True - body_start = time.perf_counter() + chunk_gap_s = 0.0 + chunk_count = 0.0 + previous_chunk_at = body_start = time.perf_counter() for chunk in response.iter_bytes(): + now = time.perf_counter() if first_chunk: - first_byte_s = time.perf_counter() - body_start + first_byte_s = now - body_start first_chunk = False + chunk_gap_s += now - previous_chunk_at + previous_chunk_at = now + chunk_count += 1.0 chunks.append(chunk) body_s = time.perf_counter() - body_start + join_start = time.perf_counter() + payload = b"".join(chunks) + join_s = time.perf_counter() - join_start return ( - b"".join(chunks), + payload, response.status_code, { "range_header_s": header_s, "range_first_byte_s": first_byte_s, "range_body_s": body_s, + "range_join_s": join_s, + "range_chunks": chunk_count, + "range_chunk_gap_s": chunk_gap_s, }, )