Class ThreadData

Lib/profiling/sampling/live_collector/collector.py:53–103

Encapsulates all profiling data for a single thread.

@dataclass
class ThreadData:
    """Encapsulates all profiling data for a single thread."""

    thread_id: int

    # Function call statistics: {location: {direct_calls: int, cumulative_calls: int}}
    result: dict = field(default_factory=lambda: collections.defaultdict(
        lambda: dict(direct_calls=0, cumulative_calls=0)
    ))

    # Thread status statistics
    has_gil: int = 0
    on_cpu: int = 0
    gil_requested: int = 0
    unknown: int = 0
    has_exception: int = 0
    total: int = 0  # Total status samples for this thread

    # Sample counts
    sample_count: int = 0
    gc_frame_samples: int = 0

    # Opcode statistics: {location: {opcode: count}}
    opcode_stats: dict = field(default_factory=lambda: collections.defaultdict(
        lambda: collections.defaultdict(int)
    ))

    def increment_status_flag(self, status_flags):
        """Update status counts based on status bit flags."""
        if status_flags & THREAD_STATUS_HAS_GIL:
            self.has_gil += 1
        if status_flags & THREAD_STATUS_ON_CPU:
            self.on_cpu += 1
        if status_flags & THREAD_STATUS_GIL_REQUESTED:
            self.gil_requested += 1
        if status_flags & THREAD_STATUS_UNKNOWN:
            self.unknown += 1
        if status_flags & THREAD_STATUS_HAS_EXCEPTION:
            self.has_exception += 1
        self.total += 1

    def as_status_dict(self):
        """Return status counts as a dict for compatibility."""
        return {
            "has_gil": self.has_gil,
            "on_cpu": self.on_cpu,
            "gil_requested": self.gil_requested,
            "unknown": self.unknown,
            "has_exception": self.has_exception,
            "total": self.total,
        }


class LiveStatsCollector(Collector):