MemProfilerX: como funciona por baixo dos panos

Criei o MemProfilerX porque precisava debugar memory leaks em produção e as ferramentas existentes eram ou muito pesadas ou não tinham as features que eu precisava. Aqui está como cada parte funciona.

Arquitetura em 3 camadas

┌─────────────────────────────────────────┐
│  Camada de Integração                    │
│  CLI, pytest plugin, middlewares web     │
├─────────────────────────────────────────┤
│  Camada Avançada                         │
│  tracemalloc, snapshots, leak detection  │
├─────────────────────────────────────────┤
│  Camada Core                             │
│  psutil tracking, decorators básicos     │
└─────────────────────────────────────────┘

Cada camada resolve um problema diferente. Vamos de baixo pra cima.

Core: O decorator que monitora em background

O @track_memory é enganosamente simples:

@track_memory(interval=1.0)
def process_data():
    return [i**2 for i in range(10_000_000)]

Por baixo dos panos, ele:

1. Spawna uma daemon thread que roda em paralelo
2. A thread coleta process.memory_info().rss a cada intervalo
3. Quando a função termina, a thread para
4. Retorna os dados + resultado original

# Simplificado
def track_memory(interval=1.0):
    def decorator(func):
        def wrapper(*args, **kwargs):
            samples = []
            stop_event = threading.Event()

            def monitor():
                while not stop_event.is_set():
                    mem = psutil.Process().memory_info().rss
                    samples.append((time.time(), mem))
                    time.sleep(interval)

            thread = threading.Thread(target=monitor, daemon=True)
            thread.start()

            result = func(*args, **kwargs)  # Executa a função real

            stop_event.set()
            thread.join()

            return {'result': result, 'memory_usage': samples}
        return wrapper
    return decorator

O pulo do gato é a daemon thread , ela morre automaticamente quando o processo principal termina. Sem cleanup manual, sem resource leaks.

Avançado: tracemalloc , O raio-X do Python

psutil te dá o consumo total do processo. Mas e se você quiser saber qual linha de código está alocando memória?

Entra o tracemalloc , um profiler em nível C embutido no Python:

with AdvancedMemoryProfiler() as profiler:
    data = [x for x in range(1_000_000)]

for alloc in profiler.get_top_allocations(5):
    print(f"{alloc.filename}:{alloc.lineno} - {alloc.size_mb:.2f}MB")

Output:

script.py:2 - 38.15MB  # A list comprehension!

Como funciona:

class AdvancedMemoryProfiler:
    def __enter__(self):
        tracemalloc.start(25)  # Rastreia até 25 stack frames
        self._start_snapshot = tracemalloc.take_snapshot()
        return self

    def __exit__(self, *args):
        self._end_snapshot = tracemalloc.take_snapshot()
        tracemalloc.stop()

    def get_top_allocations(self, n=10):
        stats = self._end_snapshot.compare_to(
            self._start_snapshot,
            'lineno'  # Agrupa por arquivo:linha
        )
        return stats[:n]

tracemalloc.start() instrui o Python a registrar cada alocação com seu traceback. ~10-15% de overhead, mas te dá informação cirúrgica.

Snapshots: Fotografando o heap

Às vezes você quer comparar o estado da memória em dois pontos no tempo. Snapshots fazem isso.

ProcessSnapshot , métricas do sistema:

snapshot = ProcessSnapshot.capture()
# RSS, VMS, CPU%, threads, file descriptors

ObjectSnapshot , raio-X dos objetos Python:

snapshot = ObjectSnapshot.capture()

Esse é interessante. Usa gc.get_objects() pra percorrer todos os objetos vivos no heap:

def capture():
    counts = defaultdict(int)
    sizes = defaultdict(int)

    for obj in gc.get_objects():  # TODOS os objetos
        obj_type = type(obj).__name__
        obj_size = sys.getsizeof(obj)
        counts[obj_type] += 1
        sizes[obj_type] += obj_size

    return ObjectSnapshot(counts, sizes)

Com dois snapshots, você calcula deltas:

before = ObjectSnapshot.capture()
# ... código suspeito ...
after = ObjectSnapshot.capture()

# Quais tipos cresceram?
for obj_type in after.counts:
    delta = after.counts[obj_type] - before.counts[obj_type]
    if delta > 1000:
        print(f"⚠️ {obj_type}: +{delta} instâncias")

Detecção de Leaks: As heurísticas

Detectar leaks automaticamente é difícil. O MemProfilerX usa uma combinação de heurísticas:

detector = LeakDetector(interval=0.5)
detector.start_monitoring(duration=30)
report = detector.analyze_leaks()

O algoritmo:

1. Coleta snapshots a cada interval segundos
2. Calcula taxa de crescimento pra cada tipo de objeto
3. Pontuação de confiança:

if growth_rate > 0.1 MB/s and instances > 1000:
    confidence = 'high'
elif growth_rate > 0.01 MB/s and instances > 100:
    confidence = 'medium'
else:
    confidence = 'low'

4. Gera recomendações baseadas no tipo:

if 'dict' in suspects:
    recommendations.append("Verifique caches sem limite")
if 'list' in suspects:
    recommendations.append("Procure append() sem cleanup")
if 'socket' in suspects:
    recommendations.append("Garanta que conexões são fechadas")

Não é perfeito , falsos positivos acontecem. Mas é um bom ponto de partida.

Context Managers: Sync e Async

A versão síncrona usa threading:

class MemoryContext:
    def __enter__(self):
        self._stop_event = threading.Event()
        self._thread = threading.Thread(target=self._monitor)
        self._thread.start()

    def __exit__(self, *args):
        self._stop_event.set()
        self._thread.join()

A versão async usa asyncio.Task:

class AsyncMemoryContext:
    async def __aenter__(self):
        self._stop_event = asyncio.Event()
        self._task = asyncio.create_task(self._monitor())

    async def _monitor(self):
        while not self._stop_event.is_set():
            # Coleta métrica
            try:
                await asyncio.wait_for(
                    self._stop_event.wait(),
                    timeout=self.interval
                )
            except asyncio.TimeoutError:
                pass  # Continua monitorando

A diferença é sutil mas importante: async não bloqueia o event loop.

Middlewares Web: Monitorando cada request

Para FastAPI (async):

class FastAPIMemoryMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        mem_before = psutil.Process().memory_info().rss

        response = await call_next(request)

        mem_after = psutil.Process().memory_info().rss
        delta_mb = (mem_after - mem_before) / 1024 / 1024

        response.headers['X-Memory-Delta-MB'] = f"{delta_mb:.2f}"
        return response

Cada response ganha headers com o delta de memória. Útil pra identificar endpoints problemáticos.

CLI: Rich pra output bonito

A CLI usa a biblioteca Rich pra tabelas coloridas:

memx check my_script.py --duration 10

┌──────────────────────────────────────────┐
│         Memory Leak Analysis             │
├────────────┬────────────┬───────────────┤
│ Type       │ Growth     │ Confidence    │
├────────────┼────────────┼───────────────┤
│ list       │ +15.2 MB   │ 🔴 high       │
│ dict       │ +3.1 MB    │ 🟡 medium     │
│ str        │ +0.5 MB    │ 🟢 low        │
└────────────┴────────────┴───────────────┘

Overhead por módulo

Por que o overhead?

• tracemalloc: Rastreia TODAS as alocações
• gc.get_objects(): Percorre TODOS os objetos vivos
• Background threads: CPU extra

A stack completa

Coleta:         psutil, tracemalloc, gc, sys.getsizeof
Concorrência:   threading (sync), asyncio (async)
Serialização:   json, csv, pickle
Visualização:   Rich (terminal), matplotlib (gráficos)
Integração:     pytest hooks, middlewares ASGI/WSGI

TL;DR

MemProfilerX combina:

• psutil pra métricas de processo
• tracemalloc pra profiling em nível Python
• gc pra inspeção de objetos
• daemon threads pra monitoramento não-invasivo
• heurísticas pra detecção de leaks

Tudo empacotado em uma API de decorators e context managers que (espero) é fácil de usar.

O código está no GitHub se quiser explorar a implementação.