import pickle
import sys
import os
import io
import subprocess
import json
from functools import lru_cache
from typing import List, Tuple
cache = lru_cache(None)
__all__ = ["format_flamegraph", "segments", "memory", "compare"]
def _frame_fmt(f, full_filename=False):
i = f['line']
fname = f['filename']
if not full_filename:
fname = fname.split('/')[-1]
func = f['name']
return f'{fname}:{i}:{func}'
def format_flamegraph(flamegraph_lines, flamegraph_script=None):
if flamegraph_script is None:
flamegraph_script = f'/tmp/{os.getuid()}_flamegraph.pl'
if not os.path.exists(flamegraph_script):
import urllib.request
print(f"Downloading flamegraph.pl to: {flamegraph_script}")
urllib.request.urlretrieve(
'https://raw.githubusercontent.com/brendangregg/FlameGraph/master/flamegraph.pl', flamegraph_script)
subprocess.run(['chmod', '+x', flamegraph_script])
args = [flamegraph_script, '--countname', 'bytes']
p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf-8')
assert p.stdin is not None
assert p.stdout is not None
p.stdin.write(flamegraph_lines)
p.stdin.close()
result = p.stdout.read()
p.stdout.close()
p.wait()
assert p.wait() == 0
return result
def _write_blocks(f, prefix, blocks):
for b in blocks:
if 'history' not in b:
f.write(f'{prefix};{b["state"]} {b["size"]}\n')
continue
accounted_for_size = 0
for h in b['history']:
sz = h['real_size']
accounted_for_size += sz
if 'frames' in h:
frames = h['frames']
if frames:
frame_s = ';'.join([_frame_fmt(f) for f in reversed(frames)])
else:
frame_s = "<non-python>"
f.write(f'{prefix};{b["state"]};{frame_s} {sz}\n')
else:
f.write(f'{prefix};{b["state"]};<no-context> {sz}\n')
gaps = b['size'] - accounted_for_size
if gaps:
f.write(f'{prefix};{b["state"]};<gaps> {gaps}\n')
def segments(snapshot, format_flamegraph=format_flamegraph):
f = io.StringIO()
for seg in snapshot['segments']:
prefix = f'stream_{seg["stream"]};seg_{seg["address"]}'
_write_blocks(f, prefix, seg['blocks'])
return format_flamegraph(f.getvalue())
def memory(snapshot, format_flamegraph=format_flamegraph):
f = io.StringIO()
for seg in snapshot['segments']:
prefix = f'stream_{seg["stream"]}'
_write_blocks(f, prefix, seg['blocks'])
return format_flamegraph(f.getvalue())
def compare(before, after, format_flamegraph=format_flamegraph):
def _seg_key(seg):
return (seg['address'], seg['total_size'])
def _seg_info(seg):
return f'stream_{seg["stream"]};seg_{seg["address"]}'
f = io.StringIO()
before_segs = {_seg_key(seg) for seg in before}
after_segs = {_seg_key(seg) for seg in after}
print(f'only_before = {[a for a,_ in (before_segs - after_segs)]}')
print(f'only_after = {[a for a,_ in (after_segs - before_segs)]}')
for seg in before:
if _seg_key(seg) not in after_segs:
_write_blocks(f, f'only_before;{_seg_info(seg)}', seg['blocks'])
for seg in after:
if _seg_key(seg) not in before_segs:
_write_blocks(f, f'only_after;{_seg_info(seg)}', seg['blocks'])
return format_flamegraph(f.getvalue())
def _format_size(num):
# https://stackoverflow.com/questions/1094841/get-human-readable-version-of-file-size
for unit in ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi"]:
if abs(num) < 1024.0:
return f"{num:3.1f}{unit}B"
num /= 1024.0
return f"{num:.1f}YiB"
class Bytes:
def __init__(self, value):
self.value = value
def __add__(self, rhs):
return Bytes(self.value + rhs)
def __repr__(self):
return _format_size(self.value)
def calc_active(seg):
return sum(b['size'] for b in seg['blocks'] if b['state'] == 'active_allocated')
def _report_free(free_external, free_internal):
total = free_external + free_internal
pct = (free_internal / total) * 100
suffix = f' ({pct:.1f}% internal)'
return f'{Bytes(total)}{suffix}'
PAGE_SIZE = 1024 * 1024 * 20
legend = f"""\
Legend:
[a ] - a segment in the allocator
^-- a page {Bytes(PAGE_SIZE)} of memory in the segment
a-z: pages filled with a single block's content
' ': page is completely free
*: page if completely full with multiple blocks
0-9: page is partially full with tensors of multiple blocks (9 == 90% full)
(X% internal) - of the free memory, X% is free because we rounded the size of the allocation.
"""
def segsum(data):
"""" Visually reports how the allocator has filled its segments. This printout can help debug fragmentation issues
since free fragments will appear as gaps in this printout. The amount of free space is reported for each segment.
We distinguish between internal free memory which occurs because the allocator rounds the allocation size, and
external free memory, which are the gaps between allocations in a segment.
Args:
data: snapshot dictionary created from _snapshot()
"""
segments = []
out = io.StringIO()
out.write(f"Summary of segments >= {Bytes(PAGE_SIZE)} in size\n")
total_reserved = 0
total_allocated = 0
free_external = 0
free_internal = 0
for seg in sorted(data['segments'], key=lambda x: (x['total_size'], calc_active(x))):
total_reserved += seg['total_size']
seg_free_external = 0
seg_free_internal = 0
seg_allocated = 0
all_ranges = []
boffset = 0
for b in seg['blocks']:
active = b['state'] == 'active_allocated'
if 'history' in b:
# use the more accureate real_size to account for internal fragmenetation if we have it
for h in b['history']:
if active:
all_ranges.append((h['addr'] - seg['address'], h['real_size'], active))
seg_allocated += h['real_size']
assert len(b['history']) == 1
seg_free_internal += b['size'] - h['real_size']
else:
if active:
all_ranges.append((boffset, b['size'], True))
seg_allocated += b['size']
if not active:
seg_free_external += b['size']
boffset += b['size']
total_allocated += seg_allocated
free_external += seg_free_external
free_internal += seg_free_internal
nseg = (seg['total_size'] - 1) // PAGE_SIZE + 1
occupied = [' ' for _ in range(nseg)]
frac = [0.0 for _ in range(nseg)]
active_size = 0
for i, (start_, size, active) in enumerate(all_ranges):
active_size += size
finish_ = (start_ + size)
start = start_ // PAGE_SIZE
finish = (finish_ - 1) // PAGE_SIZE + 1
m = chr((ord('a' if active else 'A') + (i % 26)))
for j in range(start, finish):
s = max(start_, j * PAGE_SIZE)
e = min(finish_, (j + 1) * PAGE_SIZE)
frac[j] += (e - s) / PAGE_SIZE
if occupied[j] != ' ':
occupied[j] = '0123456789*'[int(frac[j] * 10)]
else:
occupied[j] = m
stream = '' if seg['stream'] == 0 else f', stream_{seg["stream"]}'
body = ''.join(occupied)
assert seg_free_external + seg_free_internal + seg_allocated == seg['total_size']
stream = f' stream_{seg["stream"]}' if seg['stream'] != 0 else ''
if seg['total_size'] >= PAGE_SIZE:
out.write(f'[{body}] {Bytes(seg["total_size"])} allocated, '
f'{_report_free(seg_free_external, seg_free_internal)} free{stream}\n')
out.write(f'segments: {len(data["segments"])}\n')
out.write(f'total_reserved: {Bytes(total_reserved)}\n')
out.write(f'total_allocated: {Bytes(total_allocated)}\n')
internal_external = f' ({Bytes(free_internal)} internal + {Bytes(free_external)} external)' if free_internal else ''
out.write(f'total_free: {_report_free(free_external, free_internal)}\n')
out.write(legend)
assert free_internal + free_external + total_allocated == total_reserved
return out.getvalue()
def trace(data):
out = io.StringIO()
def format(entries):
segment_intervals : list = []
segment_addr_to_name = {}
allocation_addr_to_name = {}
free_names : list = []
next_name = 0
def _name():
nonlocal next_name
if free_names:
return free_names.pop()
r, m = next_name // 26, next_name % 26
next_name += 1
return f'{chr(ord("a") + m)}{"" if r == 0 else r}'
def find_segment(addr):
for name, saddr, size in segment_intervals:
if addr >= saddr and addr < saddr + size:
return name, saddr
for i, seg in enumerate(data['segments']):
saddr = seg['address']
size = seg['allocated_size']
if addr >= saddr and addr < saddr + size:
return f'seg_{i}', saddr
return None, None
count = 0
out.write(f'{len(entries)} entries\n')
total_reserved = 0
for seg in data['segments']:
total_reserved += seg['total_size']
for count, e in enumerate(entries):
if e['action'] == 'alloc':
addr, size = e['addr'], e['size']
n = _name()
seg_name, seg_addr = find_segment(addr)
if seg_name is None:
seg_name = "MEM"
offset = addr
else:
offset = addr - seg_addr
out.write(f'{n} = {seg_name}[{offset}:{Bytes(size)}]\n')
allocation_addr_to_name[addr] = (n, size, count)
count += size
elif e['action'] == 'free_requested':
addr, size = e['addr'], e['size']
name, _, _ = allocation_addr_to_name.get(addr, (addr, None, None))
out.write(f'del {name} # {Bytes(size)}\n')
elif e['action'] == 'free_completed':
addr, size = e['addr'], e['size']
count -= size
name, _, _ = allocation_addr_to_name.get(addr, (addr, None, None))
out.write(f'# free completed for {name} {Bytes(size)}\n')
if name in allocation_addr_to_name:
free_names.append(name)
del allocation_addr_to_name[name]
elif e['action'] == 'segment_alloc':
addr, size = e['addr'], e['size']
name = _name()
out.write(f'{name} = cudaMalloc({addr}, {Bytes(size)})\n')
segment_intervals.append((name, addr, size))
segment_addr_to_name[addr] = name
elif e['action'] == 'segment_free':
addr, size = e['addr'], e['size']
name = segment_addr_to_name.get(addr, addr)
out.write(f'cudaFree({name}) # {Bytes(size)}\n')
if name in segment_addr_to_name:
free_names.append(name)
del segment_addr_to_name[name]
elif e['action'] == 'oom':
size = e['size']
free = e['device_free']
out.write(f'raise OutOfMemoryError() # {Bytes(size)} requested, {Bytes(free)} free in CUDA\n')
else:
out.write(f'{e}\n')
out.write(f"TOTAL MEM: {Bytes(count)}")
for i, d in enumerate(data['device_traces']):
if d:
out.write(f'Device {i} ----------------\n')
format(d)
return out.getvalue()
class PlotWriter:
def __init__(self):
string_table: List[str] = []
suffix_table: List[Tuple[int, int]] = []
elements = []
actions: List[int] = []
initially_allocated: List[int] = []
@cache
def intern_str(s):
string_table.append(s)
return len(string_table) - 1
@cache
def intern_suffix(sid, restid):
suffix_table.append((sid, restid))
return len(suffix_table) - 1
def intern_stack(frames):
sids = [intern_str(f) for f in frames]
next_id = None
for sid in reversed(sids):
next_id = intern_suffix(sid, next_id)
return next_id
def add_element(size, lines):
elements.append({'size': size, 'info': intern_stack(lines)})
return len(elements) - 1
def to_html():
Loading ...