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
|
"""Raw data collector for Coverage."""
import sys, threading
try:
# Use the C extension code when we can, for speed.
from coverage.tracer import Tracer
except ImportError:
# Couldn't import the C extension, maybe it isn't built.
class Tracer:
"""Python implementation of the raw data tracer."""
def __init__(self):
self.data = None
self.should_trace = None
self.should_trace_cache = None
self.cur_filename = None
self.filename_stack = []
self.last_exc_back = None
def _global_trace(self, frame, event, arg_unused):
"""The trace function passed to sys.settrace."""
#print "global event: %s %r" % (event, frame.f_code.co_filename)
if event == 'call':
# Entering a new function context. Decide if we should trace
# in this file.
filename = frame.f_code.co_filename
tracename = self.should_trace_cache.get(filename)
if tracename is None:
tracename = self.should_trace(filename, frame)
self.should_trace_cache[filename] = tracename
if tracename:
# We need to trace. Push the current filename on the stack
# and record the new current filename.
self.filename_stack.append(self.cur_filename)
self.cur_filename = tracename
# Use _local_trace for tracing within this function.
return self._local_trace
else:
# No tracing in this function.
return None
return self._global_trace
def _local_trace(self, frame, event, arg_unused):
"""The trace function used within a function."""
#print "local event: %s %r" % (event, frame.f_code.co_filename)
if self.last_exc_back:
if frame == self.last_exc_back:
# Someone forgot a return event.
self.cur_filename = self.filename_stack.pop()
self.last_exc_back = None
if event == 'line':
# Record an executed line.
self.data[(self.cur_filename, frame.f_lineno)] = True
elif event == 'return':
# Leaving this function, pop the filename stack.
self.cur_filename = self.filename_stack.pop()
elif event == 'exception':
self.last_exc_back = frame.f_back
return self._local_trace
def start(self):
"""Start this Tracer."""
sys.settrace(self._global_trace)
def stop(self):
"""Stop this Tracer."""
sys.settrace(None)
class Collector:
"""Collects trace data.
Creates a Tracer object for each thread, since they track stack information.
Each Tracer points to the same shared data, contributing traced data points.
When the Collector is started, it creates a Tracer for the current thread,
and installs a function to create Tracers for each new thread started.
When the Collector is stopped, all active Tracers are stopped.
Threads started while the Collector is stopped will never have Tracers
associated with them.
"""
# The stack of active Collectors. Collectors are added here when started,
# and popped when stopped. Collectors on the stack are paused when not
# the top, and resumed when they become the top again.
_collectors = []
def __init__(self, should_trace):
"""Create a collector.
`should_trace` is a function, taking a filename, and returning a
canonicalized filename, or False depending on whether the file should
be traced or not.
"""
self.should_trace = should_trace
self.reset()
def reset(self):
"""Clear collected data, and prepare to collect more."""
# A dictionary with an entry for (Python source file name, line number
# in that file) if that line has been executed.
self.data = {}
# A cache of the results from should_trace, the decision about whether
# to trace execution in a file. A dict of filename to (filename or
# False).
self.should_trace_cache = {}
# Our active Tracers.
self.tracers = []
def _start_tracer(self):
"""Start a new Tracer object, and store it in self.tracers."""
tracer = Tracer()
tracer.data = self.data
tracer.should_trace = self.should_trace
tracer.should_trace_cache = self.should_trace_cache
tracer.start()
self.tracers.append(tracer)
# The trace function has to be set individually on each thread before
# execution begins. Ironically, the only support the threading module has
# for running code before the thread main is the tracing function. So we
# install this as a trace function, and the first time it's called, it does
# the real trace installation.
def _installation_trace(self, frame_unused, event_unused, arg_unused):
"""Called on new threads, installs the real tracer."""
# Remove ourselves as the trace function
sys.settrace(None)
# Install the real tracer.
self._start_tracer()
# Return None to reiterate that we shouldn't be used for tracing.
return None
def start(self):
"""Start collecting trace information."""
if self._collectors:
self._collectors[-1].pause()
self._collectors.append(self)
# Install the tracer on this thread.
self._start_tracer()
# Install our installation tracer in threading, to jump start other
# threads.
threading.settrace(self._installation_trace)
def stop(self):
"""Stop collecting trace information."""
assert self._collectors
assert self._collectors[-1] is self
for tracer in self.tracers:
tracer.stop()
self.tracers = []
threading.settrace(None)
# Remove this Collector from the stack, and resume the one underneath
# (if any).
self._collectors.pop()
if self._collectors:
self._collectors[-1].resume()
def pause(self):
"""Pause tracing, but be prepared to `resume`."""
for tracer in self.tracers:
tracer.stop()
threading.settrace(None)
def resume(self):
"""Resume tracing after a `pause`."""
for tracer in self.tracers:
tracer.start()
threading.settrace(self._installation_trace)
def data_points(self):
"""Return the (filename, lineno) pairs collected."""
return self.data.keys()
|
