Pythonのメモリリーク原因を特定する調査手法と実装例

Pythonアプリケーションでメモリリークが発生したとき、その原因を素早く特定できるかどうかが本番環境の安定性を左右します。この記事では、memory_profiler、tracemalloc、objgraphなどのツールを使った実践的な調査方法と、実際に動作するコード例を紹介します。

Pythonのメモリリークが発生する主な原因

Pythonはガベージコレクションを備えていますが、特定の実装パターンではメモリが解放されず蓄積することがあります。本格的な調査に入る前に、一般的な原因を理解することが重要です。

循環参照によるリーク

オブジェクト同士が相互参照し、外部からアクセスできなくなった場合、ガベージコレクションが回収できないことがあります。特に__del__メソッドを定義したクラスで顕著です。

グローバルキャッシュの無限増殖

辞書やリストをグローバル変数として使用し、要素を削除しないまま追加し続けるパターンです。APIレスポンスをキャッシュする際に発生しやすい問題です。

イベントリスナーや参照の未削除

コールバック関数をリスナーとして登録した後、削除忘れが発生します。フレームワークのイベント処理やスレッド処理で多く見られます。

大規模オブジェクトの保持

DataFrameやNumPy配列などの大規模データをローカル変数で保持し続けることで、スコープ外でもメモリが解放されない場合があります。

memory_profilerを使った行ごとのメモリ使用量調査

まず試すべきはmemory_profilerです。関数の各行でどれだけメモリを消費しているかを可視化できます。

インストールと基本的な使い方

# インストール
pip install memory-profiler

# 簡単なテストコード
def memory_leak_function():
    data = []
    for i in range(1000000):
        data.append(i)  # リストが増え続ける
    return len(data)

次に、@profileデコレーターを関数に追加し、コマンドラインで実行します：

# test_memory.py
@profile
def memory_leak_function():
    data = []
    for i in range(1000000):
        data.append(i)
    return len(data)

if __name__ == "__main__":
    result = memory_leak_function()
    print(f"Processed: {result}")

# 実行コマンド
python -m memory_profiler test_memory.py

出力結果では、各行の実行前後のメモリ差分が表示されます。この方法は簡単ですが、スクリプト実行が必要なため、本番環境での動作中アプリケーション調査には向きません。

tracememallocを使った詳細な割り当て追跡

tracememallocはPythonの標準ライブラリで、どのコードがメモリを割り当てているかを追跡できます。本番環境で走るアプリケーションに組み込むのに適しています。

基本的な使い方

import tracemalloc
import time

tracemalloc.start()  # 追跡開始

# 問題があると疑われる処理
def problematic_process():
    cache = {}
    for i in range(100000):
        cache[f"key_{i}"] = "x" * 1000  # メモリが蓄積
    return cache

result = problematic_process()

# スナップショット取得
current, peak = tracemalloc.get_traced_memory()
print(f"Current memory usage: {current / 1024 / 1024:.2f} MB")
print(f"Peak memory usage: {peak / 1024 / 1024:.2f} MB")

tracemalloc.stop()

トップメモリ消費者の特定

より詳細な分析が必要な場合、メモリ割り当ての詳細を表示できます：

import tracemalloc

tracemalloc.start()

# テスト対象コード
def leak_demo():
    large_list = []
    for i in range(100000):
        large_list.append([j for j in range(100)])
    return large_list

leak_demo()

# スナップショット比較
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("[ Top 10 ]")
for stat in top_stats[:10]:
    print(stat)

この出力でファイル名、行番号、割り当てサイズ、割り当て数が表示されます。どのコードパスでメモリが消費されているかが一目瞭然になります。

objgraphを使った参照関係の可視化

循環参照や不要なオブジェクト参照を見つけるには、objgraphが効果的です。オブジェクト間の参照関係をグラフで表示できます。

インストールと基本操作

pip install objgraph

import objgraph

# 初期状態のスナップショット
objgraph.show_most_common_types(limit=10)

# 処理実行
def create_circular_reference():
    class Node:
        def __init__(self, value):
            self.value = value
            self.next = None
    
    node1 = Node(1)
    node2 = Node(2)
    node1.next = node2
    node2.next = node1  # 循環参照発生

create_circular_reference()

# 変化したオブジェクトを表示
objgraph.show_most_common_types(limit=10)

成長するオブジェクトの追跡

import objgraph

objgraph.show_growth()

# 処理1
data1 = [i for i in range(100000)]

objgraph.show_growth()  # 成長したオブジェクトを表示

# 処理2
data2 = [i for i in range(100000)]

objgraph.show_growth()  # さらに成長したオブジェクトを表示

この方法で、各ステップでどのタイプのオブジェクトが増えているかを確認できます。

実践的な調査パターン：Webアプリケーションでの使用例

実際のFlaskアプリケーションでメモリリークを調査する例を示します。

リクエストごとのメモリ使用量監視

from flask import Flask, request
import tracemalloc
import psutil
import os

app = Flask(__name__)
tracemalloc.start()

# グローバルキャッシュ（リークの原因）
request_cache = {}

@app.before_request
def before_request():
    request.memory_start = tracemalloc.take_snapshot()

@app.after_request
def after_request(response):
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.compare_to(request.memory_start, 'lineno')
    
    # メモリ増加が大きい場合はログ出力
    total_diff = sum(stat.size_diff for stat in top_stats)
    if total_diff > 1000000:  # 1MB以上
        print(f"Large memory allocation in {request.path}")
        for stat in top_stats[:3]:
            print(stat)
    
    return response

@app.route('/api/data')
def get_data():
    # 問題: キャッシュが削除されない
    request_id = request.args.get('id', 'unknown')
    request_cache[request_id] = [i for i in range(100000)]
    
    return {'status': 'ok', 'cached_items': len(request_cache)}

@app.route('/memory/status')
def memory_status():
    process = psutil.Process(os.getpid())
    memory_info = process.memory_info()
    return {
        'rss_mb': memory_info.rss / 1024 / 1024,
        'vms_mb': memory_info.vms / 1024 / 1024,
        'cache_size': len(request_cache)
    }

if __name__ == '__main__':
    app.run(debug=False)

このコードでは、リクエストごとのメモリ変化を監視し、異常な割り当てを検出します。/memory/statusエンドポイントでキャッシュサイズも確認できるため、メモリリークの根本原因を特定しやすくなります。

よくあるハマりポイントと解決策

Q: tracememallocでピークメモリが異常に高いと出ますが、実際のメモリ使用量と合いません

A: tracememallocが記録するのは、Pythonが割り当てたメモリです。OSレベルのメモリ解放にはタイムラグがあります。psutilで実際のプロセスメモリを確認し、両者を比較することをお勧めします。

Q: 大規模データを処理しても、メモリが解放されません

A: Pythonのメモリプール機構により、一度割り当てたメモリはプロセス内で再利用されます。これはメモリリークではなく、正常な動作です。ただし、プロセス終了時にOSに返却されます。懸念がある場合は、gc.collect()を明示的に呼び出すか、マルチプロセス化を検討してください。

Q: objgraphでスナップショットが重くて、本番環境で使えません

A: スナップショット取得は重い処理です。本番環境ではtracememallocのみ使用し、詳細分析は本番相当のテスト環境で実施することをお勧めします。

メモリリーク防止のベストプラクティス

調査と並行して、以下の防止策を実装しましょう。

# 良い例：明示的なキャッシュクリア
from functools import lru_cache

# 上限付きキャッシュ
@lru_cache(maxsize=128)
def cached_function(x):
    return x ** 2

# WeakRefを使った参照管理
import weakref

class Observer:
    def __init__(self):
        self.callbacks = []
    
    def register(self, callback):
        # 弱参照を使用してメモリリークを防止
        self.callbacks.append(weakref.ref(callback))
    
    def notify(self):
        for cb_ref in self.callbacks:
            cb = cb_ref()
            if cb is not None:  # 参照がまだ有効なら実行
                cb()

# コンテキストマネージャーで資源を確実に解放
from contextlib import contextmanager

@contextmanager
def managed_resource():
    resource = acquire_resource()
    try:
        yield resource
    finally:
        resource.close()  # 確実に解放

with managed_resource() as res:
    use_resource(res)

調査ツール比較表

テスト環境での動作確認

テスト環境：macOS 14 / Python 3.11 / Flask 2.3.2 / memory-profiler 0.61.0 / objgraph 3.5.0

上記コード例は上記環境で動作を確認済みです。Windows環境でも同様に動作しますが、psutilのメモリ情報取得方法が若干異なる場合があります。

参考