Method process_frames

Lib/profiling/sampling/heatmap_collector.py:531–575

Process stack frames and count samples per line. Args: frames: List of (filename, location, funcname, opcode) tuples in leaf-to-root order. location is (lineno, end_lineno, col_offset, end_col_offset). opcode is None if not gathered.

(self, frames, thread_id, weight=1)

        self.stats.update(kwargs)

    def process_frames(self, frames, thread_id, weight=1):
        """Process stack frames and count samples per line.

        Args:
            frames: List of (filename, location, funcname, opcode) tuples in
                    leaf-to-root order. location is (lineno, end_lineno, col_offset, end_col_offset).
                    opcode is None if not gathered.
            thread_id: Thread ID for this stack trace
            weight: Number of samples this stack represents (for batched RLE)
        """
        self._total_samples += weight
        self._seen_lines.clear()

        for i, (filename, location, funcname, opcode) in enumerate(frames):
            # Normalize location to 4-tuple format
            lineno, end_lineno, col_offset, end_col_offset = normalize_location(location)

            if not self._is_valid_frame(filename, lineno):
                continue

            # frames[0] is the leaf - where execution is actually happening
            is_leaf = (i == 0)
            line_key = (filename, lineno)
            count_cumulative = line_key not in self._seen_lines
            if count_cumulative:
                self._seen_lines.add(line_key)

            self._record_line_sample(filename, lineno, funcname, is_leaf=is_leaf,
                                     count_cumulative=count_cumulative, weight=weight)

            if opcode is not None:
                # Set opcodes_enabled flag when we first encounter opcode data
                self.opcodes_enabled = True
                self._record_bytecode_sample(filename, lineno, opcode,
                                             end_lineno, col_offset, end_col_offset,
                                             weight=weight)

            # Build call graph for adjacent frames (relationships are deduplicated anyway)
            if i + 1 < len(frames):
                next_frame = frames[i + 1]
                next_lineno = extract_lineno(next_frame[1])
                self._record_call_relationship(
                    (filename, lineno, funcname),
                    (next_frame[0], next_lineno, next_frame[2])
                )

    def _is_valid_frame(self, filename, lineno):
        """Check if a frame should be included in the heatmap."""