使用Multiprocessing共享内存传递大数组时出现访问违规问题

背景 #

需要分析一年中每小时的气象数据。每小时计算前需读取对应输入，其中包含一个不会变化的超大numpy数组x，输出为存储每小时计算结果的一维numpy数组y。

目标 #

使用multiprocessing模块加速计算，通过multiprocessing的shared_memory子模块传递数组x。

运行环境 #

• Windows 10系统
• CPython 3.10.8
• Spyder 5.3.3作为IDE

测试代码 #

import multiprocessing
import numpy as np
from multiprocessing import shared_memory

def multiprocessing_function(args):
    nn, shm_name, y_shm_name, shape, dtype, N = args
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    x = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)
    existing_y_shm = shared_memory.SharedMemory(name=y_shm_name)
    y = np.ndarray((N,), dtype=dtype, buffer=existing_y_shm.buf)
    y[nn] = 1
    existing_shm.close()
    existing_y_shm.close()

if __name__ == '__main__':
    x = np.random.rand(int(1e7), 16)
    N = 8760 # Number of hours in a year
            
    dtype = x.dtype
    shm = shared_memory.SharedMemory(create=True, size=x.nbytes)
    shm_array = np.ndarray(x.shape, dtype=x.dtype, buffer=shm.buf)
    np.copyto(shm_array, x)
    y_shm = shared_memory.SharedMemory(create=True, size=N * x.itemsize)
    y_array = np.ndarray((N,), dtype=x.dtype, buffer=y_shm.buf)
    
    args_case = [(nn, shm.name, y_shm.name, x.shape, dtype, N) for nn in range(N)]
    with multiprocessing.Pool() as pool:
        pool.map(multiprocessing_function, args_case)
    
    y = np.array(y_array)

    shm.close()
    y_shm.close()
    shm.unlink()
    y_shm.unlink()

遇到的问题 #

运行代码虽能得到正确向量，但约50%概率会触发“Windows fatal exception: access violation”并导致内核崩溃。修改数组大小可能暂时正常，但重启Spyder后用新数组大小再次运行仍会出现相同错误，行为不一致，怀疑是内存泄漏问题但不知如何修复。

核心问题分析 #

Windows下multiprocessing.Pool的子进程可能在主进程未完全完成共享内存初始化时就尝试访问，或子进程资源释放顺序引发内存访问冲突；另外Spyder的IPython内核进程管理机制和multiprocessing共享内存存在兼容性问题。

具体修复步骤 #

1. 显式等待子进程完成
   不要依赖with语句的隐式处理，手动调用pool.close()和pool.join()确保所有子进程执行完毕后，再进行共享内存的释放操作。

2. 优化共享内存访问逻辑
   子进程仅在需要时打开共享内存，避免不必要的资源占用；若测试代码中未实际使用x数组，可暂时跳过其共享内存的打开操作，减少冲突概率。

3. 规避Spyder内核的兼容性问题
   尽量在命令行（cmd/PowerShell）中直接运行脚本，或在Spyder中选择“Run in external system terminal”选项执行代码，避免内核进程干扰。

4. 添加初始化延迟（可选）
   在创建进程池前添加短暂延迟，确保共享内存完全初始化完成。

修改后的代码 #

import multiprocessing
import numpy as np
from multiprocessing import shared_memory
import time

def multiprocessing_function(args):
    nn, shm_name, y_shm_name, shape, dtype, N = args
    # 仅打开需要操作的y共享内存
    existing_y_shm = shared_memory.SharedMemory(name=y_shm_name)
    y = np.ndarray((N,), dtype=dtype, buffer=existing_y_shm.buf)
    y[nn] = 1
    existing_y_shm.close()
    
    # 若实际计算需要使用x，再打开对应共享内存
    # existing_shm = shared_memory.SharedMemory(name=shm_name)
    # x = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)
    # ... 执行计算逻辑 ...
    # existing_shm.close()

if __name__ == '__main__':
    x = np.random.rand(int(1e7), 16)
    N = 8760 # 一年的小时数
            
    dtype = x.dtype
    # 创建并初始化x的共享内存
    shm = shared_memory.SharedMemory(create=True, size=x.nbytes)
    shm_array = np.ndarray(x.shape, dtype=x.dtype, buffer=shm.buf)
    np.copyto(shm_array, x)
    # 创建并初始化y的共享内存
    y_shm = shared_memory.SharedMemory(create=True, size=N * x.itemsize)
    y_array = np.ndarray((N,), dtype=x.dtype, buffer=y_shm.buf)
    
    # 等待共享内存完全初始化（针对Windows时序问题）
    time.sleep(0.5)
    
    args_case = [(nn, shm.name, y_shm.name, x.shape, dtype, N) for nn in range(N)]
    
    # 显式管理进程池生命周期
    pool = multiprocessing.Pool()
    pool.map(multiprocessing_function, args_case)
    # 等待所有子进程执行完毕
    pool.close()
    pool.join()
    
    y = np.array(y_array)

    # 释放共享内存资源
    shm.close()
    y_shm.close()
    shm.unlink()
    y_shm.unlink()

额外说明 #

• 若实际计算需要访问x数组，再在子进程中打开对应的共享内存，避免不必要的资源占用。
• 可尝试限制进程池大小（如multiprocessing.Pool(processes=4)），减少同时运行的子进程数量，降低内存访问冲突概率。

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