Update collector interface for binary replay support

Adds optional timestamp_us parameter to Collector.collect() method.
During live profiling this is None and collectors use their own timing.
During binary replay the stored timestamp is passed through, allowing
collectors to reconstruct the original timing.

Also fixes gecko_collector to use time.monotonic() instead of time.time()
for consistency with other collectors.

Author: pablogsal

Lib/profiling/sampling/collector.py

@@ -44,8 +44,15 @@ def extract_lineno(location):
 
 class Collector(ABC):
     @abstractmethod
-    def collect(self, stack_frames):
-        """Collect profiling data from stack frames."""
+    def collect(self, stack_frames, timestamp_us=None):
+        """Collect profiling data from stack frames.
+
+        Args:
+            stack_frames: List of InterpreterInfo objects
+            timestamp_us: Optional timestamp in microseconds. If provided (from
+                binary replay), use this instead of current time. If None,
+                collectors should use time.monotonic() or similar.
+        """
 
     def collect_failed_sample(self):
         """Collect data about a failed sample attempt."""

Lib/profiling/sampling/gecko_collector.py

@@ -66,7 +66,7 @@ def __init__(self, sample_interval_usec, *, skip_idle=False, opcodes=False):
         self.sample_interval_usec = sample_interval_usec
         self.skip_idle = skip_idle
         self.opcodes_enabled = opcodes
-        self.start_time = time.time() * 1000  # milliseconds since epoch
+        self.start_time = time.monotonic() * 1000  # milliseconds since start
 
         # Global string table (shared across all threads)
         self.global_strings = ["(root)"]  # Start with root
@@ -103,6 +103,9 @@ def __init__(self, sample_interval_usec, *, skip_idle=False, opcodes=False):
         # Opcode state tracking per thread: tid -> (opcode, lineno, col_offset, funcname, filename, start_time)
         self.opcode_state = {}
 
+        # For binary replay: track base timestamp (first sample's timestamp)
+        self._replay_base_timestamp_us = None
+
     def _track_state_transition(self, tid, condition, active_dict, inactive_dict,
                                   active_name, inactive_name, category, current_time):
         """Track binary state transitions and emit markers.
@@ -138,9 +141,18 @@ def _track_state_transition(self, tid, condition, active_dict, inactive_dict,
                 self._add_marker(tid, active_name, active_dict.pop(tid),
                                current_time, category)
 
-    def collect(self, stack_frames):
+    def collect(self, stack_frames, timestamp_us=None):
         """Collect a sample from stack frames."""
-        current_time = (time.time() * 1000) - self.start_time
+        if timestamp_us is not None:
+            # Use provided timestamp (from binary replay)
+            # Track first timestamp as base for relative time calculation
+            if self._replay_base_timestamp_us is None:
+                self._replay_base_timestamp_us = timestamp_us
+            # Convert to milliseconds relative to first sample
+            current_time = (timestamp_us - self._replay_base_timestamp_us) / 1000
+        else:
+            # Live sampling - use monotonic clock
+            current_time = (time.monotonic() * 1000) - self.start_time
 
         # Update interval calculation
         if self.sample_count > 0 and self.last_sample_time > 0:

Lib/profiling/sampling/live_collector/collector.py

@@ -348,7 +348,7 @@ def collect_failed_sample(self):
         self.failed_samples += 1
         self.total_samples += 1
 
-    def collect(self, stack_frames):
+    def collect(self, stack_frames, timestamp_us=None):
         """Collect and display profiling data."""
         if self.start_time is None:
             self.start_time = time.perf_counter()

Lib/profiling/sampling/pstats_collector.py

@@ -51,7 +51,7 @@ def _process_frames(self, frames):
 
             self.callers[callee][caller] += 1
 
-    def collect(self, stack_frames):
+    def collect(self, stack_frames, timestamp_us=None):
         if stack_frames and hasattr(stack_frames[0], "awaited_by"):
             # Async frame processing
             for frames, thread_id, task_id in self._iter_async_frames(stack_frames):

Lib/profiling/sampling/sample.py

@@ -12,6 +12,7 @@
 from .stack_collector import CollapsedStackCollector, FlamegraphCollector
 from .heatmap_collector import HeatmapCollector
 from .gecko_collector import GeckoCollector
+from .binary_collector import BinaryCollector
 from .constants import (
     PROFILING_MODE_WALL,
     PROFILING_MODE_CPU,
@@ -137,6 +138,10 @@ def sample(self, collector, duration_sec=10, *, async_aware=False):
             if self.collect_stats:
                 self._print_unwinder_stats()
 
+            # Print binary I/O stats if using binary collector
+            if isinstance(collector, BinaryCollector):
+                self._print_binary_stats(collector)
+
         # Pass stats to flamegraph collector if it's the right type
         if hasattr(collector, 'set_stats'):
             collector.set_stats(self.sample_interval_usec, running_time, sample_rate, error_rate, missed_samples, mode=self.mode)
@@ -278,6 +283,56 @@ def _print_unwinder_stats(self):
         if stale_invalidations > 0:
             print(f"  {ANSIColors.YELLOW}Stale cache invalidations: {stale_invalidations}{ANSIColors.RESET}")
 
+    def _print_binary_stats(self, collector):
+        """Print binary I/O encoding statistics."""
+        try:
+            stats = collector.get_stats()
+        except (ValueError, RuntimeError):
+            return  # Collector closed or stats unavailable
+
+        print(f"  {ANSIColors.CYAN}Binary Encoding:{ANSIColors.RESET}")
+
+        # Record type counts
+        repeat_records = stats.get('repeat_records', 0)
+        repeat_samples = stats.get('repeat_samples', 0)
+        full_records = stats.get('full_records', 0)
+        suffix_records = stats.get('suffix_records', 0)
+        pop_push_records = stats.get('pop_push_records', 0)
+        total_records = stats.get('total_records', 0)
+
+        if total_records > 0:
+            repeat_pct = repeat_records / total_records * 100
+            full_pct = full_records / total_records * 100
+            suffix_pct = suffix_records / total_records * 100
+            pop_push_pct = pop_push_records / total_records * 100
+        else:
+            repeat_pct = full_pct = suffix_pct = pop_push_pct = 0
+
+        print(f"    Records:          {total_records:,}")
+        print(f"      RLE repeat:     {repeat_records:,} ({ANSIColors.GREEN}{repeat_pct:.1f}%{ANSIColors.RESET}) [{repeat_samples:,} samples]")
+        print(f"      Full stack:     {full_records:,} ({full_pct:.1f}%)")
+        print(f"      Suffix match:   {suffix_records:,} ({suffix_pct:.1f}%)")
+        print(f"      Pop-push:       {pop_push_records:,} ({pop_push_pct:.1f}%)")
+
+        # Frame efficiency
+        frames_written = stats.get('total_frames_written', 0)
+        frames_saved = stats.get('frames_saved', 0)
+        compression_pct = stats.get('frame_compression_pct', 0)
+
+        print(f"  {ANSIColors.CYAN}Frame Efficiency:{ANSIColors.RESET}")
+        print(f"    Frames written:   {frames_written:,}")
+        print(f"    Frames saved:     {frames_saved:,} ({ANSIColors.GREEN}{compression_pct:.1f}%{ANSIColors.RESET})")
+
+        # Bytes written
+        bytes_written = stats.get('bytes_written', 0)
+        if bytes_written >= 1024 * 1024:
+            bytes_str = f"{bytes_written / (1024 * 1024):.1f} MB"
+        elif bytes_written >= 1024:
+            bytes_str = f"{bytes_written / 1024:.1f} KB"
+        else:
+            bytes_str = f"{bytes_written} B"
+        print(f"    Bytes (pre-zstd): {bytes_str}")
+
 
 def sample(
     pid,

Lib/profiling/sampling/stack_collector.py

@@ -18,7 +18,7 @@ def __init__(self, sample_interval_usec, *, skip_idle=False):
         self.sample_interval_usec = sample_interval_usec
         self.skip_idle = skip_idle
 
-    def collect(self, stack_frames, skip_idle=False):
+    def collect(self, stack_frames, timestamp_us=None, skip_idle=False):
         if stack_frames and hasattr(stack_frames[0], "awaited_by"):
             # Async-aware mode: process async task frames
             for frames, thread_id, task_id in self._iter_async_frames(stack_frames):
@@ -96,7 +96,7 @@ def __init__(self, *args, **kwargs):
         # Per-thread statistics
         self.per_thread_stats = {}  # {thread_id: {has_gil, on_cpu, gil_requested, unknown, has_exception, total, gc_samples}}
 
-    def collect(self, stack_frames, skip_idle=False):
+    def collect(self, stack_frames, timestamp_us=None, skip_idle=False):
         """Override to track thread status statistics before processing frames."""
         # Increment sample count once per sample
         self._sample_count += 1
@@ -128,7 +128,7 @@ def collect(self, stack_frames, skip_idle=False):
                 self.per_thread_stats[thread_id][key] += value
 
         # Call parent collect to process frames
-        super().collect(stack_frames, skip_idle=skip_idle)
+        super().collect(stack_frames, timestamp_us=timestamp_us, skip_idle=skip_idle)
 
     def set_stats(self, sample_interval_usec, duration_sec, sample_rate,
                   error_rate=None, missed_samples=None, mode=None):