多处理池：如何调用 class 对象列表中的任意方法列表
Multiprocessing pool: How to call an arbitrary list of methods on a list of class objects
一个清理版本的代码包括 (thanks @JohanL!) can be found as a Gist on GitHub.
以下截取的代码 (CPython 3.[4,5,6]) 说明了我的意图（以及我的问题）：
from functools import partial
import multiprocessing
from pprint import pprint as pp
NUM_CORES = multiprocessing.cpu_count()
class some_class:
some_dict = {'some_key': None, 'some_other_key': None}
def some_routine(self):
self.some_dict.update({'some_key': 'some_value'})
def some_other_routine(self):
self.some_dict.update({'some_other_key': 77})
def run_routines_on_objects_in_parallel_and_return(in_object_list, routine_list):
func_handle = partial(__run_routines_on_object_and_return__, routine_list)
with multiprocessing.Pool(processes = NUM_CORES) as p:
out_object_list = list(p.imap_unordered(
func_handle,
(in_object for in_object in in_object_list)
))
return out_object_list
def __run_routines_on_object_and_return__(routine_list, in_object):
for routine_name in routine_list:
getattr(in_object, routine_name)()
return in_object
object_list = [some_class() for item in range(20)]
pp([item.some_dict for item in object_list])
new_object_list = run_routines_on_objects_in_parallel_and_return(
object_list,
['some_routine', 'some_other_routine']
)
pp([item.some_dict for item in new_object_list])
verification_object_list = [
__run_routines_on_object_and_return__(
['some_routine', 'some_other_routine'],
item
) for item in object_list
]
pp([item.some_dict for item in verification_object_list])
我正在处理 some_class 类型的对象列表。 some_class 有一个 属性，一个名为 some_dict 的字典和一些可以修改字典的方法（some_routine 和 some_other_routine）。有时，我想对列表中的所有对象调用一系列方法。因为这是计算密集型的，所以我打算将对象分布在多个 CPU 核心上（使用 multiprocessing.Pool 和 imap_unordered - 列表顺序无关紧要）。
例程__run_routines_on_object_and_return__ 负责调用单个对象上的方法列表。据我所知，这工作得很好。我使用 functools.partial 来稍微简化代码的结构 - 因此多处理池必须仅将对象列表作为输入参数来处理。
问题是……它不起作用。 imap_unordered 返回的列表中包含的对象与我输入的对象相同。对象中的字典看起来就像以前一样。我已经使用类似的机制直接处理字典列表而没有出现故障，所以我怀疑修改恰好是字典的对象 属性 有问题。
在我的示例中，verification_object_list 包含正确的结果（尽管它是在单个 process/thread 中生成的）。 new_object_list 等同于 object_list，不应该是这样。
我做错了什么？
编辑
我找到了以下 question, which has an actually working and applicable answer。我按照我在每个对象上调用方法列表的想法对其进行了一些修改，并且它有效：
import random
from multiprocessing import Pool, Manager
class Tester(object):
def __init__(self, num=0.0, name='none'):
self.num = num
self.name = name
def modify_me(self):
self.num += random.normalvariate(mu=0, sigma=1)
self.name = 'pla' + str(int(self.num * 100))
def __repr__(self):
return '%s(%r, %r)' % (self.__class__.__name__, self.num, self.name)
def init(L):
global tests
tests = L
def modify(i_t_nn):
i, t, nn = i_t_nn
for method_name in nn:
getattr(t, method_name)()
tests[i] = t # copy back
return i
def main():
num_processes = num = 10 #note: num_processes and num may differ
manager = Manager()
tests = manager.list([Tester(num=i) for i in range(num)])
print(tests[:2])
args = ((i, t, ['modify_me']) for i, t in enumerate(tests))
pool = Pool(processes=num_processes, initializer=init, initargs=(tests,))
for i in pool.imap_unordered(modify, args):
print("done %d" % i)
pool.close()
pool.join()
print(tests[:2])
if __name__ == '__main__':
main()
现在，我更进一步，将我的原创 some_class 引入到游戏中，其中包含描述的字典 属性 some_dict。它不起作用：
import random
from multiprocessing import Pool, Manager
from pprint import pformat as pf
class some_class:
some_dict = {'some_key': None, 'some_other_key': None}
def some_routine(self):
self.some_dict.update({'some_key': 'some_value'})
def some_other_routine(self):
self.some_dict.update({'some_other_key': 77})
def __repr__(self):
return pf(self.some_dict)
def init(L):
global tests
tests = L
def modify(i_t_nn):
i, t, nn = i_t_nn
for method_name in nn:
getattr(t, method_name)()
tests[i] = t # copy back
return i
def main():
num_processes = num = 10 #note: num_processes and num may differ
manager = Manager()
tests = manager.list([some_class() for i in range(num)])
print(tests[:2])
args = ((i, t, ['some_routine', 'some_other_routine']) for i, t in enumerate(tests))
pool = Pool(processes=num_processes, initializer=init, initargs=(tests,))
for i in pool.imap_unordered(modify, args):
print("done %d" % i)
pool.close()
pool.join()
print(tests[:2])
if __name__ == '__main__':
main()
工作和不工作之间的区别真的很小，但我还是不明白：
diff --git a/test.py b/test.py
index b12eb56..0aa6def 100644
--- a/test.py
+++ b/test.py
@@ -1,15 +1,15 @@
import random
from multiprocessing import Pool, Manager
+from pprint import pformat as pf
-class Tester(object):
- def __init__(self, num=0.0, name='none'):
- self.num = num
- self.name = name
- def modify_me(self):
- self.num += random.normalvariate(mu=0, sigma=1)
- self.name = 'pla' + str(int(self.num * 100))
+class some_class:
+ some_dict = {'some_key': None, 'some_other_key': None}
+ def some_routine(self):
+ self.some_dict.update({'some_key': 'some_value'})
+ def some_other_routine(self):
+ self.some_dict.update({'some_other_key': 77})
def __repr__(self):
- return '%s(%r, %r)' % (self.__class__.__name__, self.num, self.name)
+ return pf(self.some_dict)
def init(L):
global tests
@@ -25,10 +25,10 @@ def modify(i_t_nn):
def main():
num_processes = num = 10 #note: num_processes and num may differ
manager = Manager()
- tests = manager.list([Tester(num=i) for i in range(num)])
+ tests = manager.list([some_class() for i in range(num)])
print(tests[:2])
- args = ((i, t, ['modify_me']) for i, t in enumerate(tests))
+ args = ((i, t, ['some_routine', 'some_other_routine']) for i, t in enumerate(tests))
这里发生了什么？
您的问题是由两件事造成的；也就是说，您正在使用 class 变量，并且您是 运行 不同进程中的代码。
由于不同的进程不共享内存，所有对象和参数都必须被 pickle 并从原始进程发送到执行它的进程。当参数是一个对象时，它的class是而不是。相反，接收进程使用自己的蓝图（即 class）。
在您当前的代码中，您将对象作为参数传递，更新它并 return 它。但是，更新不是针对对象，而是针对 class 本身，因为您更新的是 class 变量。但是，此更新不会发送回您的主进程，因此您还没有更新 class.
您想要做的是使some_dict成为您对象的一部分，而不是class的一部分。这可以通过 __init__() 方法轻松完成。从而修改some_class为：
class some_class:
def __init__(self):
self.some_dict = {'some_key': None, 'some_other_key': None}
def some_routine(self):
self.some_dict.update({'some_key': 'some_value'})
def some_other_routine(self):
self.some_dict.update({'some_other_key': 77})
这将使您的程序按预期运行。你几乎总是希望在 __init__() 调用中设置你的对象，而不是作为 class 变量，因为在后一种情况下，数据将在所有实例之间共享（并且可以被所有实例更新）。当您将数据和状态封装在 class.
的对象中时，这通常不是您想要的
编辑： 似乎我弄错了 class 是否与腌制对象一起发送。在进一步检查发生了什么之后，我认为 class 本身及其 class 变量也被腌制了。因为，如果在将对象发送到新进程之前更新 class 变量，则更新后的值可用。 但是在新进程中完成的更新仍然没有中继回原来的class。
一个清理版本的代码包括
以下截取的代码 (CPython 3.[4,5,6]) 说明了我的意图（以及我的问题）：
from functools import partial
import multiprocessing
from pprint import pprint as pp
NUM_CORES = multiprocessing.cpu_count()
class some_class:
some_dict = {'some_key': None, 'some_other_key': None}
def some_routine(self):
self.some_dict.update({'some_key': 'some_value'})
def some_other_routine(self):
self.some_dict.update({'some_other_key': 77})
def run_routines_on_objects_in_parallel_and_return(in_object_list, routine_list):
func_handle = partial(__run_routines_on_object_and_return__, routine_list)
with multiprocessing.Pool(processes = NUM_CORES) as p:
out_object_list = list(p.imap_unordered(
func_handle,
(in_object for in_object in in_object_list)
))
return out_object_list
def __run_routines_on_object_and_return__(routine_list, in_object):
for routine_name in routine_list:
getattr(in_object, routine_name)()
return in_object
object_list = [some_class() for item in range(20)]
pp([item.some_dict for item in object_list])
new_object_list = run_routines_on_objects_in_parallel_and_return(
object_list,
['some_routine', 'some_other_routine']
)
pp([item.some_dict for item in new_object_list])
verification_object_list = [
__run_routines_on_object_and_return__(
['some_routine', 'some_other_routine'],
item
) for item in object_list
]
pp([item.some_dict for item in verification_object_list])
我正在处理 some_class 类型的对象列表。 some_class 有一个 属性，一个名为 some_dict 的字典和一些可以修改字典的方法（some_routine 和 some_other_routine）。有时，我想对列表中的所有对象调用一系列方法。因为这是计算密集型的，所以我打算将对象分布在多个 CPU 核心上（使用 multiprocessing.Pool 和 imap_unordered - 列表顺序无关紧要）。
例程__run_routines_on_object_and_return__ 负责调用单个对象上的方法列表。据我所知，这工作得很好。我使用 functools.partial 来稍微简化代码的结构 - 因此多处理池必须仅将对象列表作为输入参数来处理。
问题是……它不起作用。 imap_unordered 返回的列表中包含的对象与我输入的对象相同。对象中的字典看起来就像以前一样。我已经使用类似的机制直接处理字典列表而没有出现故障，所以我怀疑修改恰好是字典的对象 属性 有问题。
在我的示例中，verification_object_list 包含正确的结果（尽管它是在单个 process/thread 中生成的）。 new_object_list 等同于 object_list，不应该是这样。
我做错了什么？
编辑
我找到了以下 question, which has an actually working and applicable answer。我按照我在每个对象上调用方法列表的想法对其进行了一些修改，并且它有效：
import random
from multiprocessing import Pool, Manager
class Tester(object):
def __init__(self, num=0.0, name='none'):
self.num = num
self.name = name
def modify_me(self):
self.num += random.normalvariate(mu=0, sigma=1)
self.name = 'pla' + str(int(self.num * 100))
def __repr__(self):
return '%s(%r, %r)' % (self.__class__.__name__, self.num, self.name)
def init(L):
global tests
tests = L
def modify(i_t_nn):
i, t, nn = i_t_nn
for method_name in nn:
getattr(t, method_name)()
tests[i] = t # copy back
return i
def main():
num_processes = num = 10 #note: num_processes and num may differ
manager = Manager()
tests = manager.list([Tester(num=i) for i in range(num)])
print(tests[:2])
args = ((i, t, ['modify_me']) for i, t in enumerate(tests))
pool = Pool(processes=num_processes, initializer=init, initargs=(tests,))
for i in pool.imap_unordered(modify, args):
print("done %d" % i)
pool.close()
pool.join()
print(tests[:2])
if __name__ == '__main__':
main()
现在，我更进一步，将我的原创 some_class 引入到游戏中，其中包含描述的字典 属性 some_dict。它不起作用：
import random
from multiprocessing import Pool, Manager
from pprint import pformat as pf
class some_class:
some_dict = {'some_key': None, 'some_other_key': None}
def some_routine(self):
self.some_dict.update({'some_key': 'some_value'})
def some_other_routine(self):
self.some_dict.update({'some_other_key': 77})
def __repr__(self):
return pf(self.some_dict)
def init(L):
global tests
tests = L
def modify(i_t_nn):
i, t, nn = i_t_nn
for method_name in nn:
getattr(t, method_name)()
tests[i] = t # copy back
return i
def main():
num_processes = num = 10 #note: num_processes and num may differ
manager = Manager()
tests = manager.list([some_class() for i in range(num)])
print(tests[:2])
args = ((i, t, ['some_routine', 'some_other_routine']) for i, t in enumerate(tests))
pool = Pool(processes=num_processes, initializer=init, initargs=(tests,))
for i in pool.imap_unordered(modify, args):
print("done %d" % i)
pool.close()
pool.join()
print(tests[:2])
if __name__ == '__main__':
main()
工作和不工作之间的区别真的很小，但我还是不明白：
diff --git a/test.py b/test.py
index b12eb56..0aa6def 100644
--- a/test.py
+++ b/test.py
@@ -1,15 +1,15 @@
import random
from multiprocessing import Pool, Manager
+from pprint import pformat as pf
-class Tester(object):
- def __init__(self, num=0.0, name='none'):
- self.num = num
- self.name = name
- def modify_me(self):
- self.num += random.normalvariate(mu=0, sigma=1)
- self.name = 'pla' + str(int(self.num * 100))
+class some_class:
+ some_dict = {'some_key': None, 'some_other_key': None}
+ def some_routine(self):
+ self.some_dict.update({'some_key': 'some_value'})
+ def some_other_routine(self):
+ self.some_dict.update({'some_other_key': 77})
def __repr__(self):
- return '%s(%r, %r)' % (self.__class__.__name__, self.num, self.name)
+ return pf(self.some_dict)
def init(L):
global tests
@@ -25,10 +25,10 @@ def modify(i_t_nn):
def main():
num_processes = num = 10 #note: num_processes and num may differ
manager = Manager()
- tests = manager.list([Tester(num=i) for i in range(num)])
+ tests = manager.list([some_class() for i in range(num)])
print(tests[:2])
- args = ((i, t, ['modify_me']) for i, t in enumerate(tests))
+ args = ((i, t, ['some_routine', 'some_other_routine']) for i, t in enumerate(tests))
这里发生了什么？
您的问题是由两件事造成的；也就是说，您正在使用 class 变量，并且您是 运行 不同进程中的代码。
由于不同的进程不共享内存，所有对象和参数都必须被 pickle 并从原始进程发送到执行它的进程。当参数是一个对象时，它的class是而不是。相反，接收进程使用自己的蓝图（即 class）。
在您当前的代码中，您将对象作为参数传递，更新它并 return 它。但是，更新不是针对对象，而是针对 class 本身，因为您更新的是 class 变量。但是，此更新不会发送回您的主进程，因此您还没有更新 class.
您想要做的是使some_dict成为您对象的一部分，而不是class的一部分。这可以通过 __init__() 方法轻松完成。从而修改some_class为：
class some_class:
def __init__(self):
self.some_dict = {'some_key': None, 'some_other_key': None}
def some_routine(self):
self.some_dict.update({'some_key': 'some_value'})
def some_other_routine(self):
self.some_dict.update({'some_other_key': 77})
这将使您的程序按预期运行。你几乎总是希望在 __init__() 调用中设置你的对象，而不是作为 class 变量，因为在后一种情况下，数据将在所有实例之间共享（并且可以被所有实例更新）。当您将数据和状态封装在 class.
编辑： 似乎我弄错了 class 是否与腌制对象一起发送。在进一步检查发生了什么之后，我认为 class 本身及其 class 变量也被腌制了。因为，如果在将对象发送到新进程之前更新 class 变量，则更新后的值可用。 但是在新进程中完成的更新仍然没有中继回原来的class。