多处理星图创建重复对象

Question

我试图使用多处理来(I)读取数据，(ii)做一些分析，以及(iii)将这些分析的输出/结果保存为自定义类的实例。我的分析函数输入了多个参数，表明multiprocessing模块中的starmap应该完成这个任务。

然而，即使我在函数中输入唯一的(非重复的)参数，结果(即类实例)有时也会重复和/或丢失。

下面是一些简化代码的示例，说明了我的问题/问题:

import numpy as np                                                                  
from multiprocessing import Pool                                                    
              
# create simple class                                                               
class EgClass:                                                                      
pass                                                                            
              
              
# define function to do analysis                                                   
def fun(eg_class, val):                                                             
eg_class.val = val                                                              
return eg_class                                                                 
              
              
if __name__ == "__main__":                                                          
              
# create a list of unique inputs                                                
unique_vals = list(np.arange(12))                                               
              
# instantiate the example class                                                 
inst = EgClass()                                                                
              
# create a list of inputs                                                       
inputs = list(zip(np.repeat(inst, len(unique_vals)), unique_vals))              
              
# apply the inputs to the function via the pool                                 
p = Pool(processes=2)                                                           
results = p.starmap(fun, inputs)                                                
p.close()                                                                       
              
result_vals = [i.val for i in results]                                          
              
# notice the result values repeat (not unique) and do not match the unique_vals 
print(result_vals)                                                              
print(unique_vals)     

有趣的是，如果您减少唯一值的数量(在本例中将unique_vals更改为unique_vals = list(np.arange(7)))，代码将按照我所期望的方式工作，即只有当输入参数增加到一定长度以上时才会出现重复值。

我看这里:Python多处理池创建重复列表但我相信这篇文章是关于想要通过跨进程共享信息来创建重复列表，这就像我想做的相反:)

最后，原谅我naïveté。我是多处理新手，有一个很好的可能我错过了一些明显的东西。

Answer 1

让我们先连续运行这段代码，不进行多处理。

from itertools import starmap
results = list(starmap(fun,inputs))
# [11, 11, 11, 11, 11, 11, 11, 11, 11, 11, 11, 11]

注意到你在整个代码中只创建了一个EgClass实例(你只调用了EgClass()一次)，np.repeat只是重复指向同一个类的指针，它不创建新对象，所以所有的试验都在修改这个对象，关于Python名称和值的事实和神话

现在对于多处理部分，让我们修改代码来打印返回对象的id，让我们也允许修改starmap的chunksize参数。

results = p.starmap(fun, inputs,chunksize=2)
result_id = [id(i) for i in results]
# [3066122542240, 3066122542240, 3066122541184, 3066122541184, 3066122540896, 3066122540896, 3066122541232, 3066122541232, 3066122541376, 3066122541376, 3066122483072, 3066122483072]

这意味着另一个进程知道发送过来的两个对象是同一个对象，因为pickle可以解析对同一个对象的多个引用，这意味着两个具有chunksize=2的对象都被正确地发送给另一个进程，并作为一个对象返回，而不是2个对象，但是由于一次只有2个对象被pickle，只有2个对象可以保持相同的id，我们可以通过改变chunksize参数来改变这一点。例如，如果我们将chunksize设置为1，我们会得到您期望的行为，但这将依赖于pickle特定的行为，而不是确保所有对象具有不同id的正确方法是实际创建不同的对象开始，或对其进行深度复制。

inputs = list(zip([EgClass() for x in range(len(unique_vals))], unique_vals))
# result_vals = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]

这实际上创建了不同的对象，每个对象都有自己的id和内部属性，所以我们不会依赖于pickle的具体实现细节。

请记住，multiprocessing的参数和返回值是原始对象的pickle副本版本，因此对它们的任何更改都不会转化为原始对象的更改，您必须创建Manager对象来传播更改。