如何恢复传递给 multiprocessing.Process 的函数的返回值？

Posted 2023-02-16

【中文标题】如何恢复传递给 multiprocessing.Process 的函数的返回值？

【英文标题】：How to recover the return value of a function passed to multiprocessing.Process?

【发布时间】：2021-11-24 20:33:49

【问题描述】：

我已经看过这个问题并开始使用它，它工作得很好How can I recover the return value of a function passed to multiprocessing.Process?

但就我而言，我想编写一个小工具，它可以连接到多台计算机并收集一些统计信息，每个统计信息都将收集在一个进程中以使其快速运行。但是，一旦我尝试将多处理命令包装在机器的一个类中，它就会失败。

这是我的代码

import multiprocessing 
import pprint


def run_task(command):
    p = subprocess.Popen(command, stdout = subprocess.PIPE, universal_newlines = True, shell = False)
    result = p.communicate()[0]
    return result


MACHINE_NAME = "cptr_name"
A_STAT = "some_stats_A"
B_STAT = "some_stats_B"

class MachineStatsGatherer():
    def __init__(self, machineName):
        self.machineName = machineName
        manager = multiprocessing.Manager() 
        self.localStats = manager.dict() # creating a shared ressource for the sub processes to use
        self.localStats[MACHINE_NAME] = machineName

    def gatherStats(self):
        self.runInParallel(
            self.GatherSomeStatsA,
            self.GatherSomeStatsB,
            )
        self.printStats()

    def printStats(self):
        pprint.pprint(self.localStats)

    def runInParallel(self, *fns):
        processes = []
        for fn in fns:
            process = multiprocessing.Process(target=fn, args=(self.localStats))
            processes.append(process)
            process.start()
        for process in processes:
            process.join()

    def GatherSomeStatsA(self, returnStats):
        # do some remote command, simplified here for the sake of debugging
        result = "Windows"
        returnStats[A_STAT] = result.find("Windows") != -1
 
    def GatherSomeStatsB(self, returnStats):
        # do some remote command, simplified here for the sake of debugging
        result = "Windows"
        returnStats[B_STAT] = result.find("Windows") != -1
 

def main():
    machine = MachineStatsGatherer("SOMEMACHINENAME")
    machine.gatherStats()
    return

if __name__ == '__main__':
    main()

这是错误信息

Traceback (most recent call last):
  File "C:\Users\mesirard\AppData\Local\Programs\Python\Python37\lib\multiprocessing\process.py", line 297, in _bootstrap
    self.run()
  File "C:\Users\mesirard\AppData\Local\Programs\Python\Python37\lib\multiprocessing\process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "d:\workdir\trunks6\Tools\VTKAppTester\Utils\NXMachineMonitorShared.py", line 45, in GatherSomeStatsA
    returnStats[A_STAT] = result.find("Windows") != -1
TypeError: 'str' object does not support item assignment
Process Process-3:
Traceback (most recent call last):
  File "C:\Users\mesirard\AppData\Local\Programs\Python\Python37\lib\multiprocessing\process.py", line 297, in _bootstrap
    self.run()
  File "C:\Users\mesirard\AppData\Local\Programs\Python\Python37\lib\multiprocessing\process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "d:\workdir\trunks6\Tools\VTKAppTester\Utils\NXMachineMonitorShared.py", line 50, in GatherSomeStatsB
    returnStats[B_STAT] = result.find("Windows") != -1
TypeError: 'str' object does not support item assignment

【问题讨论】：

为什么要使用多处理来完成这项工作？连接到计算机并收集统计数据是网络瓶颈，而不是 CPU 瓶颈，因此序列化/反序列化数据以跨进程边界传递数据的成本是不必要的浪费。这是线程的工作，而不是多处理。

（也就是说：错误消息明确地告诉您当前的问题是什么：您的代码假定传递给 GatherSomeStatsA 的参数是一个可变字典，而是一个字符串。但是，修复它以通过那个位置的 dict 不是一个好主意，因为 dict 的属性，即对一个副本的更改会更改所有其他副本不跨越进程边界 - 当一个对象被复制到子进程时子进程的副本独立于父进程的副本，并且更改不会向后传播——因此尝试的一般方法存在致命缺陷）

@CharlesDuffy 感谢您的回答。 1）我正在使用由 multiprocessing.Manager() 创建的字典，但我认为这会使其安全 2）为什么当我在进程的 args 中传递字典时代码认为它正在接收字符串

我可以回答第 2 点并且它现在可以工作，在“process = multiprocessing.Process(target=fn, args=(self.localStats))”行中，我没有在末尾添加逗号的 args 列表。应该是 process = multiprocessing.Process(target=fn, args=(self.localStats,))

【参考方案1】：

问题出在这一行

process = multiprocessing.Process(target=fn, args=(self.localStats))

它应该在 args 的末尾有一个额外的逗号，就像这样

process = multiprocessing.Process(target=fn, args=(self.localStats,))