Process a single thread's frame stack.
(self, frames, weight=1)
| 21 | self._seen_locations = set() |
| 22 | |
| 23 | def _process_frames(self, frames, weight=1): |
| 24 | """Process a single thread's frame stack.""" |
| 25 | if not frames: |
| 26 | return |
| 27 | |
| 28 | self._seen_locations.clear() |
| 29 | |
| 30 | # Process each frame in the stack to track cumulative calls |
| 31 | # frame.location is int, tuple (lineno, end_lineno, col_offset, end_col_offset), or None |
| 32 | for frame in frames: |
| 33 | lineno = extract_lineno(frame.location) |
| 34 | location = (frame.filename, lineno, frame.funcname) |
| 35 | if location not in self._seen_locations: |
| 36 | self._seen_locations.add(location) |
| 37 | self.result[location]["cumulative_calls"] += weight |
| 38 | |
| 39 | # The top frame gets counted as an inline call (directly executing) |
| 40 | top_lineno = extract_lineno(frames[0].location) |
| 41 | top_location = (frames[0].filename, top_lineno, frames[0].funcname) |
| 42 | self.result[top_location]["direct_calls"] += weight |
| 43 | |
| 44 | # Track caller-callee relationships for call graph |
| 45 | for i in range(1, len(frames)): |
| 46 | callee_frame = frames[i - 1] |
| 47 | caller_frame = frames[i] |
| 48 | |
| 49 | callee_lineno = extract_lineno(callee_frame.location) |
| 50 | caller_lineno = extract_lineno(caller_frame.location) |
| 51 | callee = (callee_frame.filename, callee_lineno, callee_frame.funcname) |
| 52 | caller = (caller_frame.filename, caller_lineno, caller_frame.funcname) |
| 53 | |
| 54 | self.callers[callee][caller] += weight |
| 55 | |
| 56 | def collect(self, stack_frames, timestamps_us=None): |
| 57 | weight = len(timestamps_us) if timestamps_us else 1 |
no test coverage detected