使用python的多处理模块分发作业[重复]

Question

问题描述

我有一些代码，我刚开始尝试在 Python 3.5 中加速。我正在尝试使用multiprocessing 模块来完成此任务。这是一个最小的示例来演示我正在尝试做的事情。

实际上，代码更简单。 Momma_Serial 类中有一个 Baby 对象列表。有时，我们想在其中的每一个上调用Baby.evolve() 方法。在实践中，会有很多 Baby 对象（本例中只有 100 个）。这是寻求并行性的最初动机。

使整个事情复杂化的是，程序的顶层通过传递函数pass_this_func() 来告诉每个Baby 对象是如何完成的。这个函数是Momma_Serial.evolve_all_elems() 的一个参数，并被传递给这个妈妈对象内的所有小婴儿对象。

class Baby:
    def __init__(self, lol):
        self.lol = lol

    def evolve(self, f):
        self.lol = f(self.lol)

def pass_this_func(thing):
    return( 2 * thing )        

class Momma_Serial:
    def __init__(self, num):
        self.my_list = [Baby(i) for i in range(num)]

    def evolve_all_elems(self, the_func):
        for baby in self.my_list:
            baby.evolve(the_func)


momma1 = Momma_Serial(100)
[baby.lol for baby in momma1.my_list]
momma1.evolve_all_elems(pass_this_func)
[baby.lol for baby in momma1.my_list]

这可以正常工作。但它很慢。这是我尝试使用多处理模块重写 Momma 类。

import multiprocessing as mp

class Momma_MP:
    def __init__(self, num):
        self.my_list = [Baby(i) for i in range(num)]

    def evolve_all_elems(self, the_func):

        num_workers = 2            

        def f(my_obj):
            my_obj.evolve(the_func)

        with mp.Pool(num_workers) as pool:        
            pool.map(f, self.my_list)

然后我尝试运行它：

momma2 = Momma_MP(100)
[baby.lol for baby in momma2.my_list]
momma2.evolve_all_elems(pass_this_func) #error comes here
# [baby.lol for baby in momma2.my_list]

但我得到了错误：

AttributeError: Can't pickle local object 'Momma_MP.evolve_all_elems.<locals>.f'

对this stackoverflow question 的回答指出“函数只有在模块的顶层定义时才可以选择。”这条语句看起来似乎实现这一点的唯一方法是在 Momma_MP 类之外定义一个函数。但我真的不想这样做，因为它会给我的代码带来更多问题。

我的问题##

（稍作修改） 有什么解决方法吗？假设我无法在类之外定义映射函数。还假设Momma() 没有在__main__ 顶级脚本环境中被实例化。另外，我不想偏离这个程序设计太多，因为我希望所有这些 Baby() 实例都被抽象掉；我不希望实例化实例或与Momma() 实例交互的地方/程序不必担心或知道与Baby() 类有关的任何事情。这些额外的限制使问题与here的情况略有不同。

顺便说一句，以下不会引发错误，但可能会进行一些复制，因为组成的 Baby 对象没有任何反应。

def outside_f(obj):
    obj.evolve(pass_this_func)       

class Momma_MP:
    def __init__(self, num):
        self.my_list = [Baby(i) for i in range(num)]

    def evolve_all_elems(self, the_func):
        num_workers = 2            

        with mp.Pool(num_workers) as pool:        
            pool.map(outside_f, self.my_list)        

momma2 = Momma_MP(100)
[baby.lol for baby in momma2.my_list]
momma2.evolve_all_elems(pass_this_func)
[baby.lol for baby in momma2.my_list] # no change here?

Answer 1

我将尝试给出一个我能找到的其他地方没有涵盖的答案（请参阅上面的我的 cmets）。我假设你有不同种类的妈妈，它们有不同的f() 功能。

你可以创建一个函数evolver():

def evolver(baby):
   momma = baby.momma
   momma.evolve(baby)

您需要在Baby 的__init__() 中分配self.momma，将Momma 实例传递给Baby：

class Baby:
    def __init__(self, lol, momma):
        self.lol = lol
        self.momma = momma

现在您将从Momma 派生以覆盖evolve() 方法以专门化evolve() 函数。

所以现在当你调用pool.map(evolver, babies) 时，它会将baby 传递给evolver()，然后它会将momma 传递给evolve() baby。

我在上面链接的答案说您也可以执行以下操作：

class Momma:
    evolver = staticmethod(evolver)

...将全局方法放入类中。