-
Notifications
You must be signed in to change notification settings - Fork 0
/
profiler.py
85 lines (76 loc) · 3.19 KB
/
profiler.py
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
# -*- coding: utf-8 -*-
#
# Copyright 2017 dpa-infocom GmbH
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import tracemalloc
import os
import linecache
import pprint
import logging
from collections import OrderedDict
tracemalloc.start()
async def run_profiler(interval):
while 1:
await show_traces()
await asyncio.sleep(interval)
snapshot_prev = None
async def show_traces():
if tracemalloc.is_tracing():
snapshot = tracemalloc.take_snapshot()
snapshot = snapshot.filter_traces((
tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),
tracemalloc.Filter(False, "<frozen importlib._bootstrap_external>"),
tracemalloc.Filter(False, linecache.__file__),
tracemalloc.Filter(False, tracemalloc.__file__),
tracemalloc.Filter(False, "<unknown>"),
))
await display_top(snapshot)
await display_diff(snapshot)
async def display_top(snapshot, key_type='lineno', limit=15):
top_stats = snapshot.statistics(key_type)
logging.debug("\n\n")
logging.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:])
logging.debug("#%s: %s:%s: %.1f KiB %dx"
% (index, filename, frame.lineno, stat.size / 1024, stat.count))
line = linecache.getline(frame.filename, frame.lineno).strip()
if line:
logging.debug(' %s' % line)
other = top_stats[limit:]
if other:
size = sum(stat.size for stat in other)
logging.debug("%s other: %.1f KiB" % (len(other), size / 1024))
total = sum(stat.size for stat in top_stats)
logging.debug("Total allocated size: %.1f KiB\n\n" % (total / 1024))
async def display_diff(snapshot, key_type='lineno', limit=15):
global snapshot_prev
if snapshot_prev:
logging.debug("Top %s differences from start" % limit)
top_diffs= snapshot.compare_to(snapshot_prev, 'lineno')
for x, stat in enumerate([x for x in top_diffs[: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:])
logging.debug("#{}: {}:{} {:.1f} KiB {}x (+{})".format(
x, filename, frame.lineno, stat.size / 1024, stat.count, stat.count_diff))
line = linecache.getline(frame.filename, frame.lineno).strip()
if line:
logging.debug(' %s' % line)
logging.debug("\n\n")
if not snapshot_prev:
snapshot_prev = snapshot