| 102 | |
| 103 | @dataclass |
| 104 | class PyThread: |
| 105 | tid: int |
| 106 | frame: Optional[PyFrame] |
| 107 | native_frames: List[NativeFrame] |
| 108 | holds_the_gil: int |
| 109 | is_gc_collecting: int |
| 110 | python_version: Optional[Tuple[int, int]] |
| 111 | name: Optional[str] = None |
| 112 | |
| 113 | @property |
| 114 | def frames(self) -> Iterable[PyFrame]: |
| 115 | yield from filter(lambda frame: not frame.is_shim, self.all_frames) |
| 116 | |
| 117 | @property |
| 118 | def first_frame(self) -> Optional[PyFrame]: |
| 119 | return next(iter(self.frames), None) |
| 120 | |
| 121 | @property |
| 122 | def all_frames(self) -> Iterable[PyFrame]: |
| 123 | current_frame = self.frame |
| 124 | while current_frame is not None: |
| 125 | yield current_frame |
| 126 | current_frame = current_frame.next |
| 127 | |
| 128 | @property |
| 129 | def status(self) -> str: |
| 130 | status = [] |
| 131 | gil_status = self.gil_status |
| 132 | gc_status = self.gc_status |
| 133 | if self.tid == 0: |
| 134 | status.append("Thread terminated") |
| 135 | if gil_status: |
| 136 | status.append(gil_status) |
| 137 | if gc_status: |
| 138 | status.append(gc_status) |
| 139 | return "[" + ",".join(status) + "]" |
| 140 | |
| 141 | @property |
| 142 | def gc_status(self) -> str: |
| 143 | if self.native_frames: |
| 144 | gc_symbols = {"gc_collect", "collect", "collect.constrprop"} |
| 145 | if any( |
| 146 | gc_symbol in frame.symbol |
| 147 | for gc_symbol in gc_symbols |
| 148 | for frame in self.native_frames |
| 149 | ): |
| 150 | return "Garbage collecting" |
| 151 | return "" |
| 152 | |
| 153 | if self.is_gc_collecting != -1: |
| 154 | if self.is_gc_collecting == 1 and self.holds_the_gil: |
| 155 | return "Garbage collecting" |
| 156 | return "" |
| 157 | |
| 158 | return "" |
| 159 | |
| 160 | @property |
| 161 | def gil_status(self) -> str: |
no outgoing calls