Speicherprofiling mit tracemalloc

Was ist tracemalloc?

import tracemalloc

tracemalloc.start()

x = [i for i in range(100_000)]

current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()

print(f"Current: {current/1024:.1f} KB")
print(f"Peak:    {peak/1024:.1f} KB")

Snapshots erstellen

import tracemalloc

tracemalloc.start()

snap1 = tracemalloc.take_snapshot()

data = [{"key": str(i)} for i in range(10_000)]

snap2 = tracemalloc.take_snapshot()

stats = snap2.compare_to(snap1, "lineno")
for stat in stats[:5]:
    print(stat)

Snapshots filtern

import tracemalloc

tracemalloc.start()
leak_function()
snap = tracemalloc.take_snapshot()

filters = [tracemalloc.Filter(True, "*/myapp/*")]
stats = snap.filter_traces(filters).statistics("lineno")
for stat in stats[:10]:
    print(stat)

Größte Speicherverbraucher

import tracemalloc

tracemalloc.start(nframe=10)   # track 10 frames deep

big_list = [i**2 for i in range(500_000)]

snap = tracemalloc.take_snapshot()
for stat in snap.statistics("traceback")[:3]:
    print(stat)
    for line in stat.traceback.format():
        print("  ", line)

Speicherlecks finden

import tracemalloc

tracemalloc.start()
snap_before = tracemalloc.take_snapshot()

for _ in range(100):
    suspected_leaky_function()

snap_after = tracemalloc.take_snapshot()
diff = snap_after.compare_to(snap_before, "lineno")
for stat in diff[:5]:
    print(stat)

get_traced_memory()

import tracemalloc

tracemalloc.start()
before, _ = tracemalloc.get_traced_memory()

do_something()

after, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()

print(f"Delta: {(after-before)/1024:.1f} KB, Peak: {peak/1024:.1f} KB")

memory_profiler

# pip install memory_profiler
from memory_profiler import profile

@profile
def build_data():
    data = [i**2 for i in range(100_000)]
    filtered = [x for x in data if x % 2 == 0]
    return filtered

build_data()
# python -m memory_profiler script.py

memray

# pip install memray
# python -m memray run script.py
# python -m memray flamegraph memray-script.py.bin

# Generates an interactive HTML flamegraph
# showing allocations by call stack

Speichernutzung reduzieren

# Instead of:
data = [process(x) for x in huge_list]   # all in memory

# Use a generator:
def gen_data(lst):
    for x in lst:
        yield process(x)

for item in gen_data(huge_list):
    consume(item)   # one at a time

sys.getsizeof()

import sys

print(sys.getsizeof([]))          # 56
print(sys.getsizeof([1,2,3]))     # 88
print(sys.getsizeof("hello"))     # 54
print(sys.getsizeof({}))          # 64

# For deep size, use recursive measurement:
def deep_size(obj):
    return sys.getsizeof(obj) + sum(
        sys.getsizeof(v) for v in vars(obj).values()
        if hasattr(obj, "__dict__")
    )

Profiling in der Produktion

import tracemalloc, pickle

tracemalloc.start(nframe=5)   # low overhead

# ... serve some requests ...

snap = tracemalloc.take_snapshot()
with open("snap.pkl","wb") as f:
    pickle.dump(snap, f)
# Analyse later offline

Kurzer Test

Zusammenfassung