Python中如何嵌套多进程并使用共享变量？报错求助

我有一个函数（称为task()，对应原描述的parentFunction），通过两个并行进程运行两次，每个进程最终生成一个字典并添加到共享列表中，该部分已通过Manager创建的预设列表实现。现在我想在task()内部再启动两个并行进程，分别向字典存入变量，尝试用Manager创建的预设字典实现，但运行代码后出现报错。最终我需要将字典列表转换为Pandas DataFrame。

def I(D, a):
        D["a"] = a
def II(D, b):
        D["a"] = b

def task(L, x):
            x = 0
            a = 1
            b = 2
            manager = Manager()
            D = manager.dict()  # <-- can be shared between processes.
            pI = Process(target=I, args=(D, 0))
            pII = Process(target=II, args=(D, 0))
            pI.start()
            pII.start()
            pI.join()
            pII.join()

            L.append(D)

if __name__ == "__main__":
    with Manager() as manager:
        L = manager.list()  # <-- can be shared between processes.
        p1 = Process(target=task, args=(L, 0))  # Passing the list
        p2 = Process(target=task, args=(L, 0))  # Passing the list
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        
        print(L)

TypeError: task() missing 1 required positional argument: 'L'

Traceback (most recent call last):
  File "C:\Users\user\AppData\Roaming\JetBrains\PyCharmCE2021.2\scratches\scratch_8.py", line 88, in <module>
    print(list(L))
  File "<string>", line 2, in __getitem__
  File "C:\Users\user\Local\Programs\Python\Python39\lib\multiprocessing\managers.py", line 810, in _callmethod
    kind, result = conn.recv()
  File "C:\Users\user\Local\Programs\Python\Python39\lib\multiprocessing\connection.py", line 256, in recv
    return _ForkingPickler.loads(buf.getbuffer())
  File "C:\Users\user\Local\Programs\Python\Python39\lib\multiprocessing\managers.py", line 934, in RebuildProxy
    return func(token, serializer, incref=incref, **kwds)
  File "C:\Users\user\Local\Programs\Python\Python39\lib\multiprocessing\managers.py", line 784, in __init__
    self._incref()
  File "C:\Users\user\Local\Programs\Python\Python39\lib\multiprocessing\managers.py", line 838, in _incref
    conn = self._Client(self._token.address, authkey=self._authkey)
  File "C:\Users\user\Local\Programs\Python\Python39\lib\multiprocessing\connection.py", line 505, in Client
    c = PipeClient(address)
  File "C:\Users\user\Local\Programs\Python\Python39\lib\multiprocessing\connection.py", line 707, in PipeClient
    _winapi.WaitNamedPipe(address, 1000)
FileNotFoundError: [WinError 2] The system cannot find the file specified

核心错误原因 #

task()函数内部创建的manager = Manager()是局部对象，当task进程执行完毕后，这个局部Manager会被销毁，导致它创建的D（共享字典代理对象）失去了对应的服务进程。主进程尝试访问L中的这个代理对象时，就会出现找不到管道的错误（WinError 2）。

另外，代码中的TypeError大概率是代码版本遗留问题，当前给出的代码中task的参数传递是正确的，不会触发该错误。

解决方法 #

方法一：将共享字典转换为普通字典后存入列表 #

在task函数中，等内部子进程执行完毕后，把Manager字典转换成普通Python字典，再添加到共享列表L中，这样存入的是独立的字典对象，不依赖局部Manager：

from multiprocessing import Process, Manager
import pandas as pd

def I(D, a):
    D["a"] = a
def II(D, b):
    D["b"] = b  # 修正：避免覆盖同一个key，原代码中两个函数都赋值D["a"]，会导致值被覆盖

def task(L, x):
    a = 1
    b = 2
    manager = Manager()
    D = manager.dict()
    pI = Process(target=I, args=(D, a))
    pII = Process(target=II, args=(D, b))
    pI.start()
    pII.start()
    pI.join()
    pII.join()

    # 将共享字典转为普通字典后添加到列表
    L.append(dict(D))

if __name__ == "__main__":
    with Manager() as manager:
        L = manager.list()
        p1 = Process(target=task, args=(L, 0))
        p2 = Process(target=task, args=(L, 0))
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        
        # 转换为DataFrame
        df = pd.DataFrame(L)
        print(df)

方法二：使用主进程的Manager创建内部共享字典 #

不在task内部创建新的Manager，而是将主进程的Manager传递给task，用它来创建内部需要的共享字典，这样字典代理对象依赖的是主进程的Manager（在with块内一直存活）：

from multiprocessing import Process, Manager
import pandas as pd

def I(D, a):
    D["a"] = a
def II(D, b):
    D["b"] = b

def task(L, manager, x):
    a = 1
    b = 2
    D = manager.dict()  # 使用主进程传递过来的Manager创建字典
    pI = Process(target=I, args=(D, a))
    pII = Process(target=II, args=(D, b))
    pI.start()
    pII.start()
    pI.join()
    pII.join()

    L.append(D)

if __name__ == "__main__":
    with Manager() as manager:
        L = manager.list()
        # 传递主进程的manager给task
        p1 = Process(target=task, args=(L, manager, 0))
        p2 = Process(target=task, args=(L, manager, 0))
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        
        df = pd.DataFrame(L)
        print(df)

额外说明 #

原代码中I和II函数都对D["a"]赋值，会导致后执行的进程覆盖前一个的值，建议修改为不同的key（如D["a"]和D["b"]），避免数据丢失。

内容的提问来源于stack exchange，提问作者ניסן פייגי�</think_never_used_51bce0c785ca2f68081bfa7d91973934>