使用multiprocessing共享内存供mpi4py进程访问numpy数组时出现资源泄漏警告

警告原因 #

你遇到的警告是multiprocessing的resource_tracker机制和mpi4py进程模型不兼容导致的误报：

• multiprocessing的资源追踪器是为自身创建的子进程设计的，会追踪父进程创建的共享内存资源，确保子进程退出时清理残留资源。
• 但mpi4py是通过MPI启动的独立进程，这些进程不属于multiprocessing的进程树，进程1的resource_tracker不知道进程0已经调用了shm.unlink()，所以在进程退出时会尝试清理一个已不存在的共享内存对象，从而抛出警告。

会不会引发严重故障？ #

当前场景下你的代码已经正确完成了共享内存的释放：进程1关闭了共享内存，进程0执行了unlink()，系统会在最后一个引用关闭后回收共享内存资源，不会残留僵尸资源，也不会引发严重故障。但这个警告会干扰日志输出，复杂场景下长期忽略可能存在潜在资源泄漏风险。

解决方法 #

方法1：禁用multiprocessing的资源追踪器 #

在所有MPI进程中关闭resource_tracker，MPI环境下不需要它管理共享内存：

import numpy as np
from multiprocessing import shared_memory
from mpi4py import MPI
# 禁用resource_tracker，避免误报
import multiprocessing.resource_tracker
multiprocessing.resource_tracker._resource_tracker.stop()

comm = MPI.COMM_WORLD
pid = comm.Get_rank()
npros = comm.Get_size()

npts = 10
dtyp = np.int64

if pid == 0:
    arr = np.arange(npts, dtype=dtyp)
    shm = shared_memory.SharedMemory(create=True, size=arr.nbytes, name='trial')
    brr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
    brr[:] = arr[:]
    del arr

comm.barrier()

if pid == 1:
    existing_shm = shared_memory.SharedMemory(name='trial')
    crr = np.ndarray((npts), dtype=dtyp, buffer=existing_shm.buf)
    print(crr)

comm.barrier()

if pid == 1:
    del crr
    existing_shm.close()

comm.barrier()

if pid == 0:
    del brr
    shm.close()
    shm.unlink()

方法2：改用MPI原生共享内存（推荐） #

既然使用mpi4py，直接用MPI自带的共享内存机制更适配，避免和multiprocessing工具冲突：

import numpy as np
from mpi4py import MPI

comm = MPI.COMM_WORLD
pid = comm.Get_rank()
npros = comm.Get_size()

npts = 10
dtyp = np.int64
dtype_size = dtyp().itemsize
total_bytes = npts * dtype_size

# 进程0创建共享内存窗口
if pid == 0:
    win = MPI.Win.Allocate_shared(total_bytes, dtype_size, comm=comm)
    buf, _ = win.Shared_query(0)
    arr = np.ndarray((npts,), dtype=dtyp, buffer=buf)
    arr[:] = np.arange(npts, dtype=dtyp)
else:
    # 其他进程获取共享内存窗口
    win = MPI.Win.Allocate_shared(0, dtype_size, comm=comm)
    buf, _ = win.Shared_query(0)
    arr = np.ndarray((npts,), dtype=dtyp, buffer=buf)

comm.barrier()

if pid == 1:
    print(arr)

# 关闭窗口释放资源
win.Free()

方法3：手动注销共享内存资源 #

在进程1中关闭共享内存后，手动从resource_tracker中注销该资源：

# 修改进程1的代码块
if pid == 1:
    del crr
    existing_shm.close()
    from multiprocessing import resource_tracker
    resource_tracker.unregister('/trial', 'shared_memory')

内容的提问来源于stack exchange，提问作者peedaruos