MCPcopy Create free account
hub / github.com/bloomberg/pystack / PyThread

Class PyThread

src/pystack/types.py:104–168  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

102
103@dataclass
104class 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:

Calls

no outgoing calls