Class CollapsedStackCollector

Lib/profiling/sampling/stack_collector.py:31–63

class CollapsedStackCollector(StackTraceCollector):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stack_counter = collections.Counter()

    def process_frames(self, frames, thread_id, weight=1):
        # Extract only (filename, lineno, funcname) - opcode not needed for collapsed stacks
        # frame is (filename, location, funcname, opcode)
        call_tree = tuple(
            (f[0], extract_lineno(f[1]), f[2]) for f in reversed(frames)
        )
        self.stack_counter[(call_tree, thread_id)] += weight

    def export(self, filename):
        lines = []
        for (call_tree, thread_id), count in self.stack_counter.items():
            parts = [f"tid:{thread_id}"]
            for file, line, func in call_tree:
                # This is what pstats does for "special" frames:
                if file == "~" and line == 0:
                    part = func
                else:
                    part = f"{os.path.basename(file)}:{func}:{line}"
                parts.append(part)
            stack_str = ";".join(parts)
            lines.append((stack_str, count))

        lines.sort(key=lambda x: (-x[1], x[0]))

        with open(filename, "w") as f:
            for stack, count in lines:
                f.write(f"{stack} {count}\n")
        print(f"Collapsed stack output written to {filename}")


class FlamegraphCollector(StackTraceCollector):