如何在multiprocessing.Pool初始化时修改fork生成的类实例？

测试代码 #

import os
from multiprocessing import Pool

class A:
  def initialize(self):
    print('initialize', self, os.getpid())

  def run(self):
    print('run       ', self, os.getpid())

a = A()
print('root      ', a, os.getpid())

with Pool(1, initializer=a.initialize) as pool:
  pool.apply(a.run)

运行输出 #

root       <__main__.A object at 0x7b644c0af1c0> 386
initialize <__main__.A object at 0x7b644c0af1c0> 3264
run        <__main__.A object at 0x7b644c0ae080> 3264

问题解析 #

子进程中initialize和run的self地址不同，核心原因是两个方法用到的实例来源不一样：

• initialize是子进程fork完成后立即调用的，用的是父进程a实例在子进程内存里的直接副本，所以虚拟地址和父进程里的a一致。
• pool.apply(a.run)会把父进程的a实例序列化（pickle）后传给子进程，子进程反序列化时会生成一个全新的A实例，这就是run里self地址不同的原因。

解决方案 #

要实现子进程初始化时修改实例，后续run方法复用该修改后的实例，且不用全局变量，可以通过类静态属性在子进程内部保存初始化后的实例：

修改后的代码 #

import os
from multiprocessing import Pool

class A:
    # 子进程内部存储初始化后的实例
    _child_instance = None

    def initialize(self):
        print('initialize', self, os.getpid())
        # 将当前实例绑定到类静态属性，供子进程后续调用
        A._child_instance = self
        # 在这里执行大内存加载等初始化操作
        self.large_data = [i for i in range(10**6)]

    def run(self):
        print('run       ', self, os.getpid())
        # 使用初始化好的大内存数据
        print(f"Loaded large data length: {len(self.large_data)}")

def run_child_instance():
    # 调用子进程中初始化好的实例的run方法
    A._child_instance.run()

if __name__ == "__main__":
    a = A()
    print('root      ', a, os.getpid())

    with Pool(1, initializer=a.initialize) as pool:
        pool.apply(run_child_instance)

运行输出示例 #

root       <__main__.A object at 0x10f5d7d90> 4567
initialize <__main__.A object at 0x10f5d7d90> 4568
run        <__main__.A object at 0x10f5d7d90> 4568
Loaded large data length: 1000000

这种方式下，initialize和run使用的是同一个实例，完美实现子进程内实例的初始化复用，且无需依赖全局变量。

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