Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def wait(self):
if self._stdin_feeder:
self._stdin_feeder.join()
self._stderr_reader.join()
returncode = self._process.wait()
logger.debug('done with %s. exit code %d', self.cmdline()[0], returncode)
return returncode
def display_top(snapshot, group_by='lineno', limit=10):
snapshot = snapshot.filter_traces((
tracemalloc.Filter(False, ""),
tracemalloc.Filter(False, ""),
))
top_stats = snapshot.statistics(group_by)
logger.debug("Top %s lines", limit)
for index, stat in enumerate(top_stats[:limit], 1):
frame = stat.traceback[0]
# replace "/path/to/module/file.py" with "module/file.py"
filename = os.sep.join(frame.filename.split(os.sep)[-2:])
logger.debug("#%s: %s:%s: %.1f KiB",
index, filename, frame.lineno, stat.size / 1024)
line = linecache.getline(frame.filename, frame.lineno).strip()
if line:
logger.debug(' %s', line)
other = top_stats[limit:]
if other:
size = sum(stat.size for stat in other)
logger.debug("%s other: %.1f KiB", len(other), size / 1024)
total = sum(stat.size for stat in top_stats)
logger.debug("Total allocated size: %.1f KiB", total / 1024)
for index, stat in enumerate(top_stats[:limit], 1):
frame = stat.traceback[0]
# replace "/path/to/module/file.py" with "module/file.py"
filename = os.sep.join(frame.filename.split(os.sep)[-2:])
logger.debug("#%s: %s:%s: %.1f KiB",
index, filename, frame.lineno, stat.size / 1024)
line = linecache.getline(frame.filename, frame.lineno).strip()
if line:
logger.debug(' %s', line)
other = top_stats[limit:]
if other:
size = sum(stat.size for stat in other)
logger.debug("%s other: %.1f KiB", len(other), size / 1024)
total = sum(stat.size for stat in top_stats)
logger.debug("Total allocated size: %.1f KiB", total / 1024)
snapshot = snapshot.filter_traces((
tracemalloc.Filter(False, ""),
tracemalloc.Filter(False, ""),
))
top_stats = snapshot.statistics(group_by)
logger.debug("Top %s lines", limit)
for index, stat in enumerate(top_stats[:limit], 1):
frame = stat.traceback[0]
# replace "/path/to/module/file.py" with "module/file.py"
filename = os.sep.join(frame.filename.split(os.sep)[-2:])
logger.debug("#%s: %s:%s: %.1f KiB",
index, filename, frame.lineno, stat.size / 1024)
line = linecache.getline(frame.filename, frame.lineno).strip()
if line:
logger.debug(' %s', line)
other = top_stats[limit:]
if other:
size = sum(stat.size for stat in other)
logger.debug("%s other: %.1f KiB", len(other), size / 1024)
total = sum(stat.size for stat in top_stats)
logger.debug("Total allocated size: %.1f KiB", total / 1024)
logger.debug("Top %s lines", limit)
for index, stat in enumerate(top_stats[:limit], 1):
frame = stat.traceback[0]
# replace "/path/to/module/file.py" with "module/file.py"
filename = os.sep.join(frame.filename.split(os.sep)[-2:])
logger.debug("#%s: %s:%s: %.1f KiB",
index, filename, frame.lineno, stat.size / 1024)
line = linecache.getline(frame.filename, frame.lineno).strip()
if line:
logger.debug(' %s', line)
other = top_stats[limit:]
if other:
size = sum(stat.size for stat in other)
logger.debug("%s other: %.1f KiB", len(other), size / 1024)
total = sum(stat.size for stat in top_stats)
logger.debug("Total allocated size: %.1f KiB", total / 1024)
def sigusr1_handler(signo, stack_frame):
logger.debug('got SIGUSR1')
snapshot = tracemalloc.take_snapshot()
logger.debug('display_top')
display_top(snapshot)
def lookup_file(self, *names):
"""Try to fetch a specific file by digging in containers."""
name, remainings = names[0], names[1:]
try:
file = self.get_member(name)
except KeyError:
return None
logger.debug('lookup_file(%s) -> %s', names, file)
specialize(file)
if not remainings:
return file
container = file.as_container
if not container:
return None
return container.lookup_file(*remainings)
def path(self):
if self._path is None:
logger.debug('unpacking %s', self._name)
assert self._temp_dir is None
self._temp_dir = get_temporary_directory()
with profile('container_extract', self.container):
self._path = self.container.extract(self._name, self._temp_dir.name)
return self._path