Gejala Masalah: Lonjakan Memori dan OOM-Kill pada Batch Analitik

Pada arsitektur sistem analitik risiko perbankan, Directed Acyclic Graph (DAG) lazim digunakan untuk memodelkan ketergantungan perhitungan portofolio, pricing derivatif, dan scoring kredit. Masalah timbul saat worker pemroses batch (misalnya Celery worker atau Kubernetes job) mengalami kenaikan alokasi Random Access Memory (RAM) secara monoton hingga menerima sinyal SIGKILL akibat Out-Of-Memory (OOM-kill).

Karakteristik utama dari insiden ini:

  • Konsumsi Resident Set Size (RSS) proses Python tidak turun kembali ke batas dasar (baseline) setelah eksekusi batch selesai dan fungsi pemrosesan mengembalikan nilai (return).
  • Batch berikutnya mengalokasikan memori di atas sisa memori batch sebelumnya, mempercepat saturasi limit container worker.
  • Panggilan fungsi del pada objek root DAG gagal mengembalikan memori sistem operasi.

Analisis Alasan: Batasan Reference Counting CPython

CPython mengelola siklus hidup memori menggunakan dua mekanisme: Reference Counting dan Generational Cyclic Garbage Collector.

Secara default, jika reference count suatu objek menyentuh angka 0, deallokasi terjadi seketika via deallokator pymalloc atau allocator C runtime. Namun, pada implementasi DAG yang naif, simpul (node) anak menyimpan referensi langsung ke simpul induk (parent-child circular binding), sementara simpul induk memegang referensi ke daftar simpul anak.

Ketika kalkulasi selesai, penghapusan referensi terluar ke DAG hanya mengurangi reference count root sebesar 1. Siklus referensi internal antara node parent dan child mencegah reference count menyentuh 0. Objek-objek ini bertahan dan dipromosikan ke Generation 2 (Gen 2) pada Cyclic GC.

CPython GC Gen 2 hanya berjalan secara periodik berdasarkan selisih ambang batas alokasi/deallokasi (threshold). Pada kalkulasi finansial berkecepatan tinggi yang mengalokasikan buffer data besar (seperti list objek transaksi atau matriks), batas memori OS terlampaui jauh sebelum cyclic GC sempat memeriksa Gen 2.

Isolasi Root Cause dengan tracemalloc dan objgraph

Langkah pertama isolasi adalah memvalidasi baris alokasi yang bocor menggunakan modul standar tracemalloc, kemudian memetakan siklus referensi menggunakan paket objgraph.

1. Deteksi Alokasi Tertinggal Menggunakan tracemalloc

import tracemalloc
import gc

tracemalloc.start()

# Titik snapshot awal sebelum kalkulasi
snapshot_sebelum = tracemalloc.take_snapshot()

# Eksekusi kalkulasi DAG analitik
run_credit_risk_dag()

# Paksa pembersihan lokal di luar cyclic GC
gc.collect(0) 

# Titik snapshot akhir setelah kalkulasi selesai
snapshot_sesudah = tracemalloc.take_snapshot()

top_stats = snapshot_sesudah.compare_to(snapshot_sebelum, 'lineno')
for stat in top_stats[:5]:
    print(stat)

Output tracemalloc akan menunjukkan bahwa baris inisialisasi class DAGNode mendominasi kenaikan memori bersih pasca-eksekusi fungsi, menandakan bahwa instance class tersebut gagal dibebaskan oleh deallokator.

2. Memvisualisasikan Siklus Referensi dengan objgraph

import objgraph

# Ambil sampel salah satu simpul DAG yang masih hidup di heap
leaked_nodes = objgraph.by_type('DAGNode')
print(f"Jumlah node tertinggal di heap: {len(leaked_nodes)}")

if leaked_nodes:
    # Render diagram hubungan referensi siklik ke format file gambar
    objgraph.show_backrefs(
        leaked_nodes[:1],
        max_depth=4,
        filename='dag_reference_cycle.png'
    )

Diagram yang dihasilkan akan mengonfirmasi siklus: DAGNode (Parent) -> list -> DAGNode (Child) -> list -> DAGNode (Parent).

Solusi Teknis: Refactoring Lifecycle dan weakref

Penyelesaian masalah ini dilakukan dalam tiga tahap: implementasi referensi lemah (weak reference) pada pointer anak-ke-induk, penyediaan antarmuka pembersihan eksplisit, serta penyesuaian threshold garbage collector.

Implementasi Kode Sebelum Fix (Mengandung Siklus Referensi)

class DAGNode:
    def __init__(self, node_id: str, payload: list):
        self.node_id = node_id
        self.payload = payload  # Alokasi data analitik perbankan
        self.children = []
        self.parents = []

    def add_child(self, child_node: "DAGNode"):
        self.children.append(child_node)
        # Strong reference cycle terjadi di sini:
        child_node.parents.append(self)

Implementasi Kode Sesudah Fix (Menggunakan weakref)

Mengubah relasi parents menjadi weakref.ref memastikan bahwa simpul anak tidak meningkatkan nilai reference count dari simpul induk. Jika simpul induk keluar dari scope, reference count langsung turun ke 0.

import weakref

class DAGNode:
    __slots__ = ('node_id', 'payload', 'children', '_parents')

    def __init__(self, node_id: str, payload: list):
        self.node_id = node_id
        self.payload = payload
        self.children = []
        self._parents = []

    def add_child(self, child_node: "DAGNode"):
        self.children.append(child_node)
        # Gunakan weak reference untuk referensi balik ke parent
        child_node._parents.append(weakref.ref(self))

    @property
    def parents(self):
        # Resolve referensi yang masih hidup, eliminasi pointer mati
        valid_parents = []
        for ref in self._parents:
            parent_instance = ref()
            if parent_instance is not None:
                valid_parents.append(parent_instance)
        return valid_parents

    def destroy(self):
        """Putus seluruh relasi dan bebaskan payload kalkulasi secara eksplisit."""
        self.payload = None
        self.children.clear()
        self._parents.clear()

Penggunaan __slots__ juga mengeliminasi __dict__ internal pada setiap node, memangkas konsumsi overhead memori dasar per node hingga 40-60%.

Konfigurasi dan Threshold Garbage Collection

Pada pemrosesan batch besar berulang, siklus GC otomatis CPython perlu disesuaikan agar tidak menunggu alokasi Gen 2 membengkak.

import gc

def configure_gc_for_batch():
    # Cek threshold bawaan (default: 700, 10, 10)
    # Sesuaikan threshold Gen 0 dan Gen 1 untuk data throughput tinggi
    gc.set_threshold(50000, 15, 15)

def execute_batch_lifecycle(dag_root: DAGNode):
    try:
        # Jalankan eksekusi perhitungan analitik
        dag_root.compute()
    finally:
        # 1. Bersihkan referensi internal
        dag_root.destroy()
        # 2. Putus variabel lokal eksplisit
        del dag_root
        # 3. Pemicu manual GC jika batch memproses volume besar
        gc.collect()

Langkah Verifikasi Konsumsi Memori

Lakukan verifikasi regresi untuk memastikan bahwa memori kembali ke baseline pasca-perbaikan:

import tracemalloc
import gc

def verify_dag_cleanup():
    gc.collect()
    tracemalloc.start()
    
    baseline_mem = tracemalloc.get_traced_memory()[0]

    # Simulasi eksekusi batch DAG 100.000 node
    root = DAGNode("root", [x for x in range(1000)])
    for i in range(10000):
        child = DAGNode(f"child_{i}", [x for x in range(50)])
        root.add_child(child)
    
    # Lifecycle cleanup
    root.destroy()
    del root
    del child
    
    gc.collect()
    
    current_mem, peak_mem = tracemalloc.get_traced_memory()
    tracemalloc.stop()

    memory_delta = current_mem - baseline_mem
    print(f"Peak Memory: {peak_mem / (1024 * 1024):.2f} MB")
    print(f"Leaked Delta Memory: {memory_delta / 1024:.2f} KB")
    
    # Asersi bahwa delta kembali mendekati nol (di bawah ambang batas toleransi 50KB)
    assert memory_delta < 50 * 1024, "Memory leak terdeteksi pada lifecycle DAG!"

verify_dag_cleanup()

Dengan menerapkan weakref untuk referensi balik, membuang struktur internal via hook destroy(), dan mengendalikan pemanggilan gc.collect(), engine DAG analitik dapat beroperasi secara kontinu tanpa risiko OOM-kill akibat siklus referensi yang tertinggal.