/usr/lib/python3/dist-packages/pex/tracer.py is in python3-pex 1.1.14-2ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | # Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import sys
import threading
import time
from contextlib import contextmanager
from .variables import ENV
__all__ = ('TraceLogger',)
class Trace(object):
__slots__ = ('msg', 'verbosity', 'parent', 'children', '_clock', '_start', '_stop')
def __init__(self, msg, parent=None, verbosity=1, clock=time):
self.msg = msg
self.verbosity = verbosity
self.parent = parent
if parent is not None:
parent.children.append(self)
self.children = []
self._clock = clock
self._start = self._clock.time()
self._stop = None
def stop(self):
self._stop = self._clock.time()
def duration(self):
assert self._stop is not None
return self._stop - self._start
class TraceLogger(object):
"""
A multi-threaded tracer.
"""
def __init__(self, predicate=None, output=sys.stderr, clock=time, prefix=''):
"""
If predicate specified, it should take a "verbosity" integer and determine whether
or not to log, e.g.
def predicate(verbosity):
try:
return verbosity < int(os.environ.get('APP_VERBOSITY', 0))
except ValueError:
return False
output defaults to sys.stderr, but can take any file-like object.
"""
self._predicate = predicate or (lambda verbosity: True)
self._length = None
self._output = output
self._isatty = getattr(output, 'isatty', False) and output.isatty()
self._lock = threading.RLock()
self._local = threading.local()
self._clock = clock
self._prefix = prefix
def should_log(self, V):
return self._predicate(V)
def log(self, msg, V=1, end='\n'):
if not self.should_log(V):
return
if not self._isatty and end == '\r':
# force newlines if we're not a tty
end = '\n'
trailing_whitespace = ''
with self._lock:
if self._length and self._length > (len(self._prefix) + len(msg)):
trailing_whitespace = ' ' * (self._length - len(msg) - len(self._prefix))
self._output.write(''.join([self._prefix, msg, trailing_whitespace, end]))
self._output.flush()
self._length = (len(self._prefix) + len(msg)) if end == '\r' else 0
def print_trace_snippet(self):
parent = self._local.parent
parent_verbosity = parent.verbosity
if not self.should_log(parent_verbosity):
return
traces = []
while parent:
if self.should_log(parent.verbosity):
traces.append(parent.msg)
parent = parent.parent
self.log(' :: '.join(reversed(traces)), V=parent_verbosity, end='\r')
def print_trace(self, indent=0, node=None):
node = node or self._local.parent
with self._lock:
self.log(' ' * indent + ('%s: %.1fms' % (node.msg, 1000.0 * node.duration())),
V=node.verbosity)
for child in node.children:
self.print_trace(indent=indent + 2, node=child)
@contextmanager
def timed(self, msg, V=1):
if getattr(self._local, 'parent', None) is None:
self._local.parent = Trace(msg, verbosity=V, clock=self._clock)
else:
parent = self._local.parent
self._local.parent = Trace(msg, parent=parent, verbosity=V, clock=self._clock)
self.print_trace_snippet()
yield
self._local.parent.stop()
if self._local.parent.parent is not None:
self._local.parent = self._local.parent.parent
else:
self.print_trace()
self._local.parent = None
TRACER = TraceLogger(
predicate=lambda verbosity: verbosity <= ENV.PEX_VERBOSE,
prefix='pex: '
)
|