1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
|
# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt
"""Raw data collector for coverage.py."""
from __future__ import annotations
import atexit
import inspect
import sys
import threading
import traceback
from types import CodeType, FrameType, ModuleType
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, cast
from coverage import env
from coverage.types import (
TArc, TFileDisposition, TLineNo, TTraceData, TTraceFileData, TTraceFn,
TTracer, TWarnFn,
)
# When running meta-coverage, this file can try to trace itself, which confuses
# everything. Don't trace ourselves.
THIS_FILE = __file__.rstrip("co")
def log(msg):
with open("/tmp/pan.out", "a") as f:
print(msg, file=f)
def panopticon(meth):
def _wrapped(self, *args, **kwargs):
assert not kwargs
log(f"{meth.__name__}{args!r}")
try:
return meth(self, *args, **kwargs)
except:
with open("/tmp/pan.out", "a") as f:
traceback.print_exception(sys.exception(), file=f)
sys.monitoring.set_events(sys.monitoring.COVERAGE_ID, 0)
raise
return _wrapped
class Pep669Tracer(TTracer):
"""Python implementation of the raw data tracer for PEP669 implementations."""
def __init__(self) -> None:
# pylint: disable=super-init-not-called
# Attributes set from the collector:
self.data: TTraceData
self.trace_arcs = False
self.should_trace: Callable[[str, FrameType], TFileDisposition]
self.should_trace_cache: Dict[str, Optional[TFileDisposition]]
self.should_start_context: Optional[Callable[[FrameType], Optional[str]]] = None
self.switch_context: Optional[Callable[[Optional[str]], None]] = None
self.warn: TWarnFn
# The threading module to use, if any.
self.threading: Optional[ModuleType] = None
self.cur_file_data: Optional[TTraceFileData] = None
self.last_line: TLineNo = 0
self.cur_file_name: Optional[str] = None
#self.context: Optional[str] = None
self.code_cache: Dict[CodeType, Tuple[bool, Optional[TTraceFileData]]] = {}
# The frame_stack parallels the Python call stack. Each entry is
# information about an active frame, a three-element tuple:
# [0] The TTraceData for this frame's file. Could be None if we
# aren't tracing this frame.
# [1] The current file name for the frame. None if we aren't tracing
# this frame.
# [2] The last line number executed in this frame.
self.frame_stack: List[Tuple[Optional[TTraceFileData], Optional[str], TLineNo]] = []
self.thread: Optional[threading.Thread] = None
self.stopped = False
self._activity = False
self.in_atexit = False
# On exit, self.in_atexit = True
atexit.register(setattr, self, "in_atexit", True)
def __repr__(self) -> str:
me = id(self)
points = sum(len(v) for v in self.data.values())
files = len(self.data)
return f"<Pep669Tracer at 0x{me:x}: {points} data points in {files} files>"
def log(self, marker: str, *args: Any) -> None:
"""For hard-core logging of what this tracer is doing."""
with open("/tmp/debug_trace.txt", "a") as f:
f.write("{} {}[{}]".format(
marker,
id(self),
len(self.frame_stack),
))
if 0: # if you want thread ids..
f.write(".{:x}.{:x}".format( # type: ignore[unreachable]
self.thread.ident,
self.threading.current_thread().ident,
))
f.write(" {}".format(" ".join(map(str, args))))
if 0: # if you want callers..
f.write(" | ") # type: ignore[unreachable]
stack = " / ".join(
(fname or "???").rpartition("/")[-1]
for _, fname, _, _ in self.frame_stack
)
f.write(stack)
f.write("\n")
def start(self) -> TTraceFn:
"""Start this Tracer.
Return a Python function suitable for use with sys.settrace().
"""
self.stopped = False
if self.threading:
if self.thread is None:
self.thread = self.threading.current_thread()
else:
if self.thread.ident != self.threading.current_thread().ident:
# Re-starting from a different thread!? Don't set the trace
# function, but we are marked as running again, so maybe it
# will be ok?
#self.log("~", "starting on different threads")
return self._cached_bound_method_trace
self.myid = sys.monitoring.COVERAGE_ID
sys.monitoring.use_tool_id(self.myid, "coverage.py")
events = sys.monitoring.events
sys.monitoring.set_events(self.myid, events.PY_START)
sys.monitoring.register_callback(self.myid, events.PY_START, self.sysmon_py_start)
# Use PY_START globally, then use set_local_event(LINE) for interesting
# frames, so i might not need to bookkeep which are the interesting frame.
sys.monitoring.register_callback(self.myid, events.PY_RESUME, self.sysmon_py_resume)
sys.monitoring.register_callback(self.myid, events.PY_RETURN, self.sysmon_py_return)
sys.monitoring.register_callback(self.myid, events.PY_YIELD, self.sysmon_py_yield)
# UNWIND is like RETURN/YIELD
sys.monitoring.register_callback(self.myid, events.LINE, self.sysmon_line)
#sys.monitoring.register_callback(self.myid, events.BRANCH, self.sysmon_branch)
def stop(self) -> None:
"""Stop this Tracer."""
sys.monitoring.set_events(self.myid, 0)
sys.monitoring.free_tool_id(self.myid)
def activity(self) -> bool:
"""Has there been any activity?"""
return self._activity
def reset_activity(self) -> None:
"""Reset the activity() flag."""
self._activity = False
def get_stats(self) -> Optional[Dict[str, int]]:
"""Return a dictionary of statistics, or None."""
return None
@panopticon
def sysmon_py_start(self, code, instruction_offset: int):
# Entering a new frame. Decide if we should trace in this file.
self._activity = True
self.frame_stack.append((self.cur_file_data, self.cur_file_name, self.last_line))
if code in self.code_cache:
tracing_code, self.cur_file_data = self.code_cache[code]
else:
tracing_code = self.cur_file_data = None
if tracing_code is None:
self.cur_file_name = filename = code.co_filename
disp = self.should_trace_cache.get(filename)
if disp is None:
frame = inspect.currentframe()
disp = self.should_trace(filename, frame)
self.should_trace_cache[filename] = disp
tracing_code = disp.trace
if tracing_code:
tracename = disp.source_filename
assert tracename is not None
if tracename not in self.data:
self.data[tracename] = set() # type: ignore[assignment]
self.cur_file_data = self.data[tracename]
else:
self.cur_file_data = None
self.code_cache[code] = (tracing_code, self.cur_file_data)
if tracing_code:
events = sys.monitoring.events
log(f"set_local_events({code!r})")
sys.monitoring.set_local_events(
self.myid,
code,
(
sys.monitoring.events.LINE |
sys.monitoring.events.PY_RETURN |
sys.monitoring.events.PY_RESUME |
sys.monitoring.events.PY_YIELD
)
)
self.last_line = -code.co_firstlineno
@panopticon
def sysmon_py_resume(self, code, instruction_offset: int):
self.frame_stack.append((self.cur_file_data, self.cur_file_name, self.last_line))
frame = inspect.currentframe()
self.last_line = frame.f_lineno
@panopticon
def sysmon_py_return(self, code, instruction_offset: int, retval: object):
if self.cur_file_data is not None:
cast(Set[TArc], self.cur_file_data).add((self.last_line, -code.co_firstlineno))
# Leaving this function, pop the filename stack.
self.cur_file_data, self.cur_file_name, self.last_line = (
self.frame_stack.pop()
)
def sysmon_py_yield(self, code, instruction_offset: int, retval: object):
...
@panopticon
def sysmon_line(self, code, line_number: int):
#assert self.cur_file_data is not None
if self.cur_file_data is not None:
if self.trace_arcs:
cast(Set[TArc], self.cur_file_data).add((self.last_line, line_number))
else:
cast(Set[TLineNo], self.cur_file_data).add(line_number)
self.last_line = line_number
return sys.monitoring.DISABLE
@panopticon
def sysmon_branch(self, code, instruction_offset: int, destination_offset: int):
...
|
