python-memory-safe-scripts

内存安全的 Python 脚本模式

经过实战检验的模式，可帮助 Python 脚本在 systemd 的 MemoryMax 限制下持续运行。这些模式提取自 repair_direct_parquet.py（24 个工作线程并行修复）和 exness_tick_cache_seeder.py（每日为 10 个交易品种播种数据）；此前在一台配备 62 GB GPU 的工作站上进行了 5 轮 OOM 优化。

核心见解：Python 的垃圾回收器会释放对象，但 C 分配器（glibc ptmalloc2）不会将已释放的页面归还给操作系统。如果不显式调用 malloc_trim(0)，RSS 就会持续增长——即使执行了 del 和 gc.collect() 也是如此。使用 MIMALLOC_PURGE_DELAY 的 mimalloc 会有所帮助，但显式清理更快。

自我演进技能：此技能会在使用过程中不断改进。如果说明有误、参数已偏移，或需要采用变通方案，请立即修复此文件，不要推迟。仅在出现真实且可复现的问题时更新。

7 种模式

1. 缓存分配器清理句柄

最重要的模式。首次调用时缓存 ctypes 库句柄，后续清理只需调用一次 FFI，且不会产生分配开销。

import ctypes
import gc
import sys

_purge_lib = None
_purge_method = None  # "mimalloc" | "glibc" | "none"

def _force_allocator_purge():
    """Force mimalloc/glibc to return freed pages to the OS."""
    global _purge_lib, _purge_method

    if sys.platform != "linux":
        return

    if _purge_method is None:
        try:
            _purge_lib = ctypes.CDLL("libmimalloc.so.2")
            _purge_method = "mimalloc"
        except OSError:
            try:
                _purge_lib = ctypes.CDLL("libc.so.6")
                _purge_method = "glibc"
            except OSError:
                _purge_method = "none"

    if _purge_method == "mimalloc":
        _purge_lib.mi_collect(ctypes.c_bool(True))
    elif _purge_method == "glibc":
        _purge_lib.malloc_trim(0)

def _force_gc():
    """Python GC + allocator purge. Call every 50 iterations + between work units."""
    gc.collect()
    _force_allocator_purge()

缓存句柄为何重要：ctypes.CDLL("libc.so.6") 会调用 dlopen()，而该函数本身也会分配内存。在循环中调用 1400 次会适得其反。应只缓存一次。

为何优先使用 mimalloc：当 LD_PRELOAD=libmimalloc.so.2 生效时，由于 mimalloc 拦截了所有分配，glibc 的 malloc_trim 就会成为空操作。对 mimalloc 来说，mi_collect(True) 才是正确的清理方式。

2. HTTP 响应生命周期

提取所需内容后，立即关闭响应。requests 库会持有响应正文、连接池引用以及 urllib3 内部状态。

# CORRECT: extract content, close, delete
resp = requests.get(url, timeout=60)
if resp.status_code != 200:
    resp.close()
    return None

content = resp.content  # Extract what you need
resp.close()            # Release connection pool reference
del resp                # Drop the Python object

# Process content...
del content             # Release after processing

# WRONG: response lives until end of function scope
resp = requests.get(url, timeout=60)
data = parse(resp.content)  # resp still alive, holding ~18 MB
return data                 # resp GC'd eventually... maybe

这为何重要：每个 requests.Response 都会持有 content（完整响应正文）、对 urllib3.HTTPResponse 的引用，以及连接池的 PoolManager。当 4 个并发工作线程处理 1400 个 URL 时，未关闭的响应会不断累积，占用数百 MB 内存。

3. 显式删除对象

不要依赖 Python 的 GC 来处理大型对象。对象不再需要时，应立即使用 del。

# After writing a DataFrame to Parquet
_atomic_write_parquet(df, path)
del df  # Only reference gone → immediate refcount GC

# After extracting data from a ZIP
with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
    df = pl.read_csv(zf.open(zf.namelist()[0]), ...)
del zip_bytes  # Release raw ZIP content after parsing

# After processing a list of work items
results = process_all(missing_days)
del missing_days  # Release the 1400-element date list

何时使用 del：任何大于约 1 MB 且不再需要的对象，例如 DataFrame、HTTP 响应中的字节串、ZIP 内容和大型列表。

4. 定期 GC 节奏

在两个层级调用 _force_gc()：

# Level 1: Every 50 iterations within a work unit
for i, item in enumerate(items):
    process(item)
    if (i + 1) % 50 == 0:
        _force_gc()

# Level 2: Between major work units
for symbol in symbols:
    seed_symbol(symbol)
    _force_gc()  # Release all per-symbol state before next symbol

为何选择 50：已在一台 32 核工作站上通过实测验证。设为 100 时，清理前 RSS 会增长过多；设为 25 时，清理开销可被测量到（吞吐量损失约为 2%）。repair_direct_parquet.py 的实测结果表明，50 是最佳值。

5. ThreadPoolExecutor 清理

执行器退出后，显式清理残留状态。

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as ex:
    pending = {}
    # ... bounded future submission pattern ...

# After pool exits:
del pending     # Future objects hold references to results
del missing     # Work item list
_force_gc()     # Release worker thread memory + allocator pages

对于进阶场景（工作线程中的数据库连接），应显式关闭线程局部资源：

pool.shutdown(wait=False, cancel_futures=True)
for t in threading.enumerate():
    if t.name.startswith("ThreadPoolExecutor"):
        _close_worker_cache()  # Close DB connections
gc.collect()
_force_allocator_purge()

6. 线程局部连接复用

切勿在循环内创建数据库连接或 HTTP 会话。使用 threading.local()，让每个工作线程各自持有一个连接。

import threading

_thread_local = threading.local()

def _get_worker_cache():
    """One DB connection per worker thread, reused across all iterations."""
    cache = getattr(_thread_local, "cache", None)
    if cache is None:
        cache = DatabaseClient()
        _thread_local.cache = cache
    return cache

def _close_worker_cache():
    """Explicit cleanup at shutdown."""
    cache = getattr(_thread_local, "cache", None)
    if cache is not None:
        cache.close()
        _thread_local.cache = None

这为何能防止文件描述符耗尽：每个 urllib3.PoolManager(maxsize=20) 最多会持有 20 个文件描述符。在一个 24 工作线程的线程池中，每次迭代都创建新的对象，会在几分钟内耗尽 ulimit -n 1024。使用线程局部复用后，N 个工作线程的文件描述符数量会保持在约 4N+50。

7. systemd 服务配置

[Service]
# Memory limits — hard kill prevents runaway RSS
MemoryHigh=2G        # Soft limit: triggers reclaim pressure
MemoryMax=4G         # Hard limit: SIGKILL on breach
MemorySwapMax=0      # No swap escape — fail fast, don't thrash

# mimalloc: replaces glibc ptmalloc2, returns freed pages faster
Environment=LD_PRELOAD=/usr/lib/x86_64-linux-gnu/libmimalloc.so.2
Environment=MIMALLOC_PURGE_DELAY=1000

# OOM priority (lower = more likely to survive)
OOMScoreAdjust=-200
ManagedOOMMemoryPressure=kill

MemoryHigh 与 MemoryMax：MemoryHigh 会触发内核回收内存（cgroup 压力），使进程变慢但仍能存活。MemoryMax 则会直接触发硬性 SIGKILL。将 MemoryHigh 设为 MemoryMax 的 50-66%，让内核有机会在终止进程前回收内存。

---

反模式

反模式	失败原因	修复方法
在循环中使用 ctypes.CDLL("libc.so.6")	dlopen() 会分配内存；调用 1000 次会浪费约 50 MB	将句柄缓存在模块全局变量中
调用 requests.get() 后未调用 resp.close()	响应正文和连接池会一直保留到 GC	提取内容后立即调用 resp.close() + del resp
工作单元之间不调用 gc.collect()	循环引用会在不同符号／批次之间累积	每个主要工作单元之间调用 _force_gc()
每次循环迭代都新建 DB 连接	通过 urllib3 PoolManager，每个连接会占用 20 个文件描述符	使用 threading.local()，让每个线程复用一个连接
为解决 OOM 而提高 MemoryMax	这会掩盖内存泄漏；RSS 会持续增长，直到达到任意限制	先修复内存泄漏。修复方法始终是模式 1-6 之一
调用 del df 后不调用 gc.collect()	引用计数会释放对象，但 glibc 会保留内存页	del + gc.collect() + _force_allocator_purge()
未设置 MemorySwapMax	进程会换出到磁盘而不是退出；持续抖动数小时	设置 MemorySwapMax=0——快速失败，不要持续抖动

---

诊断清单

当脚本在 systemd 下收到 SIGKILL（status=9）时：

1. 确认是否为 OOM：journalctl --user -u service.service | grep -E "killed|signal|KILL"
2. 检查 RSS 峰值：systemctl --user status service.service | grep Memory（显示峰值）
3. 分析稳定状态下的 RSS：手动运行脚本，在相隔 30s 的 3 个时间点检查 /proc/PID/status 中的 VmRSS
4. 检查文件描述符数量：ls /proc/PID/fd | wc -l — 如果 >500，怀疑连接频繁创建和销毁（模式 6）
5. 检查分配器：服务文件中是否有 LD_PRELOAD=libmimalloc.so.2？如果使用 glibc，检查是否调用了 malloc_trim
6. 添加定期日志：每 50 次迭代调用 logger.info("RSS=%d MB", psutil.Process().memory_info().rss // 1048576)

---

参考实现

脚本	使用的模式	RSS 配置
scripts/repair_direct_parquet.py	全部 7 种模式	从 3 GB 开始，使用 24 个工作线程时稳定在约 13 GB
scripts/exness_tick_cache_seeder.py	模式 1-5、7	10 个标的 x 1400 天期间稳定在 163 MB
scripts/tick_cache_seeder.py	模式 3、7	使用 mimalloc 时峰值为 2.5 GB（使用 glibc 时为 4.47 GB）

执行后复盘

此技能完成后，请在结束前检查：

1. 命令是否成功？ — 如果没有，请修复导致失败的指令或错误表。
2. 参数或输出是否发生变化？ — 如果底层工具的接口有所变化，请更新 Usage 示例和 Parameters 表以保持一致。
3. 是否需要变通处理？ — 如果你不得不临时调整（使用不同标志、增加额外步骤），请更新此 SKILL.md，避免下次调用时再次需要同样的变通方法。

仅在问题真实且可复现时更新——不要基于猜测进行更新。