SharedMemory结合numpy与多进程的大内存数据读取问题

架构老林

2026-6-17

问题背景

我有大量数据，先通过SharedMemory加载至内存，再使用多个子进程通过multiprocessing.Pool.map进行读取操作。

初始代码

以下是简化版代码（不完全符合官方示例规范）：

class SharedObject: # 传递给子进程的SharedMemory包装类
  def __init__(self, blablabla):
    self.shmem = SharedMemory(create=True, size=numbers) # 引用随实例存活，避免被GC回收
    temp_arr = np.ndarray(self.shape, dtype=self.dtype, buffer=self.shmem.buf)
    temp_arr[:] = ...lot of data... # 该数组在__init__结束后销毁

  def __getitem__(self, indices) -> np.ndarray: # 被子进程调用
    selected = np.ndarray(self.shape, dtype=self.dtype, buffer=self.shmem.buf)
    return selected.__getitem__(indices)

# 主进程代码
shobj = SharedObject()
with multiprocessing.Pool() as pool:
  result= list(pool.map(f, shobj))  # f会调用shobj.__getitem__

异常现象

设备内存为64GB，数据量较小时代码运行正常，但加载约40GB数据时出现锁误用错误：

n __enter__
return self._semlock.__enter__()
       ^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\mnandcharvim\AppData\Local\Programs\Python\Python312\Lib\multiprocessing\connection.py", line 321, in _recv_bytes
waitres = _winapi.WaitForMultipleObjects(

由于仅需读取数据，希望将其加载至只读内存以避免锁操作，但对SharedMemory是否无锁存疑。

改进后的代码

尝试贴合官方文档示例修改代码：

shmems = [] # 模块级变量，用于维持SharedMemory引用
class SharedObject:
  def __init__(self, blablabla):
    shmem = SharedMemory(name=self.name, create=True, size=numbers) # 该__init__内的引用会被销毁
    shmems.append(shmem) # 模块级引用维持其存活
    temp_arr = np.ndarray(self.shape, dtype=self.dtype, buffer=self.shmem.buf)
    temp_arr[:] = ...lot of data...

  def __getitem__(self, indices) -> np.ndarray: # 被子进程调用
    shmem = SharedMemory(name=self.name) # 新增行
    selected = np.ndarray(self.shape, dtype=self.dtype, buffer=shmem.buf)
    return selected.__getitem__(indices)

# 主进程代码
shobj = SharedObject()
with multiprocessing.Pool() as pool:
  result= list(pool.map(f, shobj))  # f会调用shobj.__getitem__

改进后的异常现象

此版本在执行shmem = SharedMemory(name=self.name)时报错，提示系统资源不足无法执行mmap，但数据已加载至内存，资源本应充足。

注意事项

仅对数据进行读取操作
无法使用线程（读取后需执行不释放GIL的计算）
需要SharedMemory结合numpy数组与多进程的清晰可行方案
需以numpy数组形式读取数据

最小复现示例

需根据设备资源调整参数避免内存溢出，运行时会在主循环某迭代中出现死锁，按CTRL+C可查看错误。环境为Python 3.12.2 64位、Windows 10。
script.py：

from multiprocessing.shared_memory import SharedMemory
import numpy as np
import multiprocessing

class SharedObject:
    def __init__(self):
        self.shape = (2000000, 2049)
        self.shmem = SharedMemory(create=True, size=2000000*8196)
        self.dtype = np.float32
        _temp_arr=np.ndarray(self.shape, dtype=self.dtype, buffer=self.shmem.buf)
        _temp_arr[:] = [np.zeros(shape=2049, dtype=self.dtype) for j in range(2000000)]

def f(data: SharedObject) -> int:
    print("hello")
    return 5

if __name__ == "__main__":
    shobj = SharedObject()

    for j in range(20):
        print(j)
        with multiprocessing.Pool() as pool:
            list(pool.map(f, [shobj]*256)) # 256次调用f

解决方案

核心问题分析

初始代码锁问题：Windows下SharedMemory对象内部包含同步锁，跨进程传递时锁会被复制，多进程访问时触发锁竞争导致死锁。
改进代码资源不足问题：每次调用__getitem__都重新创建SharedMemory实例，会重复映射内存区域，大内存场景下消耗过量虚拟内存，触发资源不足错误。

可行实现方案

方案思路

主进程创建并初始化共享内存后，仅传递共享内存的名称和数组元信息（shape、dtype）给子进程，不传递带锁的SharedMemory实例。
子进程通过名称打开共享内存，映射为numpy数组后直接读取，彻底规避锁竞争。
主进程维护共享内存引用，确保所有子进程完成操作前不销毁内存。

代码实现

from multiprocessing.shared_memory import SharedMemory
import numpy as np
import multiprocessing

# 主进程维护共享内存引用，防止被GC提前回收
_shared_memory_refs = []

class SharedArrayMeta:
    """仅传递共享内存的元信息，不含锁对象"""
    def __init__(self, shm_name: str, shape: tuple, dtype: np.dtype):
        self.shm_name = shm_name
        self.shape = shape
        self.dtype = dtype

def create_shared_array(shape: tuple, dtype: np.dtype, data: np.ndarray = None) -> SharedArrayMeta:
    """创建共享内存并写入数据，返回元信息对象"""
    dtype = np.dtype(dtype)
    size = np.prod(shape) * dtype.itemsize
    shm = SharedMemory(create=True, size=size)
    _shared_memory_refs.append(shm)
    
    # 映射为numpy数组并写入数据
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    if data is not None:
        arr[:] = data
    
    return SharedArrayMeta(shm.name, shape, dtype)

def read_shared_array(meta: SharedArrayMeta) -> np.ndarray:
    """子进程通过元信息读取共享内存数组"""
    shm = SharedMemory(name=meta.shm_name)
    arr = np.ndarray(meta.shape, dtype=meta.dtype, buffer=shm.buf)
    # 返回只读视图，避免误修改（可选但推荐）
    arr.flags.writeable = False
    return arr

# 子进程执行的任务函数
def f(meta: SharedArrayMeta) -> int:
    arr = read_shared_array(meta)
    # 执行你的只读操作，示例取切片计算
    sample = arr[0:10]
    return len(sample)

if __name__ == "__main__":
    # 初始化数据（示例用大数组，实际替换为你的大数据）
    shape = (2000000, 2049)
    dtype = np.float32
    data = np.zeros(shape, dtype=dtype)
    
    # 创建共享内存数组
    meta = create_shared_array(shape, dtype, data)
    
    # 多进程读取
    for j in range(20):
        print(f"迭代 {j}")
        with multiprocessing.Pool() as pool:
            results = list(pool.map(f, [meta]*256))
    
    # 所有任务完成后清理共享内存
    for shm in _shared_memory_refs:
        shm.close()
        shm.unlink()

关键优化点

规避锁传递：子进程仅通过名称打开共享内存，不接触主进程的SharedMemory实例，彻底消除锁竞争。
只读保护：设置数组为只读视图，确保数据不会被误修改，符合只读需求。
资源复用：子进程无需重复创建SharedMemory实例，每个进程打开一次即可，大幅降低资源消耗。
显式清理：主进程在所有任务完成后显式关闭并销毁共享内存，避免内存泄漏。

Windows平台额外注意事项

Windows下共享内存的生命周期与创建进程绑定，必须确保主进程在所有子进程结束后再销毁共享内存。
使用multiprocessing.Pool时，每次循环重新创建池可自然保证子进程退出后再进行后续操作。

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

SharedMemory结合numpy与多进程的大内存数据读取问题

初始代码 #

异常现象 #

改进后的代码 #

改进后的异常现象 #

注意事项 #

最小复现示例 #

核心问题分析 #

可行实现方案 #