Function sample_live

Lib/profiling/sampling/sample.py:439–533

Sample a process in live/interactive mode with curses TUI. Args: pid: Process ID to sample collector: LiveStatsCollector instance duration_sec: How long to sample for (seconds) all_threads: Whether to sample all threads realtime_stats: Whether to print re

(
    pid,
    collector,
    *,
    duration_sec=None,
    all_threads=False,
    realtime_stats=False,
    mode=PROFILING_MODE_WALL,
    async_aware=None,
    native=False,
    gc=True,
    opcodes=False,
    blocking=False,
)

def sample_live(
    pid,
    collector,
    *,
    duration_sec=None,
    all_threads=False,
    realtime_stats=False,
    mode=PROFILING_MODE_WALL,
    async_aware=None,
    native=False,
    gc=True,
    opcodes=False,
    blocking=False,
):
    """Sample a process in live/interactive mode with curses TUI.

    Args:
        pid: Process ID to sample
        collector: LiveStatsCollector instance
        duration_sec: How long to sample for (seconds)
        all_threads: Whether to sample all threads
        realtime_stats: Whether to print real-time sampling statistics
        mode: Profiling mode - WALL (all samples), CPU (only when on CPU),
              GIL (only when holding GIL), ALL (includes GIL and CPU status),
              EXCEPTION (only when thread has an active exception)
        native: Whether to include native frames
        gc: Whether to include GC frames
        opcodes: Whether to include opcode information
        blocking: Whether to stop all threads before sampling for consistent snapshots

    Returns:
        The collector with collected samples
    """
    import curses

    # Check if process is alive before doing any heavy initialization
    if not _is_process_running(pid):
        print(f"No samples collected - process {pid} exited before profiling could begin.", file=sys.stderr)
        return collector

    # Get sample interval from collector
    sample_interval_usec = collector.sample_interval_usec

    # PROFILING_MODE_ALL implies no skipping at all
    if mode == PROFILING_MODE_ALL:
        skip_non_matching_threads = False
    else:
        skip_non_matching_threads = True

    profiler = SampleProfiler(
        pid,
        sample_interval_usec,
        all_threads=all_threads,
        mode=mode,
        native=native,
        gc=gc,
        opcodes=opcodes,
        skip_non_matching_threads=skip_non_matching_threads,