pce*_*con 8 python pool class pickle multiprocessing
我正在对我的代码应用一些并行化,我在其中使用类.我知道如果不采用Python提供的任何其他方法,就不可能选择类方法.我在这里找到了解决方案.在我的代码中,我必须使用类来并行化部分.在这里,我发布了一个非常简单的代码,只是代表我的结构(是相同的,但是我删除了方法内容,这是很多数学计算,对于我得到的输出来说是微不足道的).问题是'因为我可以腌制一种方法(shepard_interpolation),但是另一种方法(calculate_orientation_uncertainty)我得到了泡菜错误.我不知道为什么会这样,或者为什么它会起作用.
def _pickle_method(method):
func_name = method.im_func.__name__
obj = method.im_self
cls = method.im_class
if func_name.startswith('__') and not func_name.endswith('__'): #deal with mangled names
cls_name = cls.__name__.lstrip('_')
func_name = '_' + cls_name + func_name
print cls
return _unpickle_method, (func_name, obj, cls)
def _unpickle_method(func_name, obj, cls):
for cls in cls.__mro__:
try:
func = cls.__dict__[func_name]
except KeyError:
pass
else:
break
return func.__get__(obj, cls)
class ImageData(object):
def __init__(self, width=60, height=60):
self.width = width
self.height = height
self.data = []
for i in range(width):
self.data.append([0] * height)
def shepard_interpolation(self, seeds=20):
print "ImD - Sucess"
import copy_reg
import types
from itertools import product
from multiprocessing import Pool
copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)
class VariabilityOfGradients(object):
def __init__(self):
pass
@staticmethod
def aux():
return "VoG - Sucess"
@staticmethod
def calculate_orientation_uncertainty():
results = []
pool = Pool()
for x, y in product(range(1, 5), range(1, 5)):
result = pool.apply_async(VariabilityOfGradients.aux)
results.append(result.get())
pool.close()
pool.join()
if __name__ == '__main__':
results = []
pool = Pool()
for _ in range(3):
result = pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()])
results.append(result.get())
pool.close()
pool.join()
VariabilityOfGradients.calculate_orientation_uncertainty()
在运行时,我得到了"PicklingError:无法pickle:属性查找内置 .function失败".这几乎与此处相同.我看到的唯一区别是我的方法是静态的.
编辑:
我注意到在我的calculate_orientation_uncertainty中,当我调用函数时result = pool.apply_async(VariabilityOfGradients.aux()),即使用括号(在doc示例中我从未见过这个),它似乎工作.但是,当我试图得到结果时,我收到"TypeError:'int'对象不可调用"...
任何帮助,将不胜感激.先感谢您.
您可以定义在模块级纯函数和一个静态方法为好.这保留了static方法的调用语法,内省和可继承性功能,同时避免了酸洗问题:
def aux():
return "VoG - Sucess"
class VariabilityOfGradients(object):
aux = staticmethod(aux)
例如,
import copy_reg
import types
from itertools import product
import multiprocessing as mp
def _pickle_method(method):
"""
Author: Steven Bethard (author of argparse)
http://bytes.com/topic/python/answers/552476-why-cant-you-pickle-instancemethods
"""
func_name = method.im_func.__name__
obj = method.im_self
cls = method.im_class
cls_name = ''
if func_name.startswith('__') and not func_name.endswith('__'):
cls_name = cls.__name__.lstrip('_')
if cls_name:
func_name = '_' + cls_name + func_name
return _unpickle_method, (func_name, obj, cls)
def _unpickle_method(func_name, obj, cls):
"""
Author: Steven Bethard
http://bytes.com/topic/python/answers/552476-why-cant-you-pickle-instancemethods
"""
for cls in cls.mro():
try:
func = cls.__dict__[func_name]
except KeyError:
pass
else:
break
return func.__get__(obj, cls)
copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)
class ImageData(object):
def __init__(self, width=60, height=60):
self.width = width
self.height = height
self.data = []
for i in range(width):
self.data.append([0] * height)
def shepard_interpolation(self, seeds=20):
print "ImD - Success"
def aux():
return "VoG - Sucess"
class VariabilityOfGradients(object):
aux = staticmethod(aux)
@staticmethod
def calculate_orientation_uncertainty():
pool = mp.Pool()
results = []
for x, y in product(range(1, 5), range(1, 5)):
# result = pool.apply_async(aux) # this works too
result = pool.apply_async(VariabilityOfGradients.aux, callback=results.append)
pool.close()
pool.join()
print(results)
if __name__ == '__main__':
results = []
pool = mp.Pool()
for _ in range(3):
result = pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()])
results.append(result.get())
pool.close()
pool.join()
VariabilityOfGradients.calculate_orientation_uncertainty()
产量
ImD - Success
ImD - Success
ImD - Success
['VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess']
顺便说一下,result.get()会阻塞调用进程,直到完成pool.apply_async(例如ImageData.shepard_interpolation)调用的函数.所以
for _ in range(3):
result = pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()])
results.append(result.get())
真的是ImageData.shepard_interpolation按顺序呼叫,打败了游泳池的目的.
相反,你可以使用
for _ in range(3):
pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()],
callback=results.append)
results.append当函数完成时,在调用进程的线程中调用回调函数(例如).它被发送一个参数 - 函数的返回值.因此,没有什么能阻止三个pool.apply_async呼叫快速进行,并且三个呼叫完成的工作ImageData.shepard_interpolation将同时执行.
或者,pool.map在这里使用它可能更简单.
results = pool.map(ImageData.shepard_interpolation, [ImageData()]*3)
如果使用multiprocessing被调用的fork pathos.multiprocesssing,则可以直接在多处理map函数中使用类和类方法.这是因为dill使用而不是pickle或cPickle,并且dill可以序列化python中的几乎任何东西.
pathos.multiprocessing还提供了一个异步映射函数......它可以map使用多个参数(例如map(math.pow, [1,2,3], [4,5,6]))
请参阅: 多处理和莳萝可以一起做什么？
和:http: //matthewrocklin.com/blog/work/2013/12/05/Parallelism-and-Serialization/
>>> from pathos.multiprocessing import ProcessingPool as Pool
>>>
>>> p = Pool(4)
>>>
>>> def add(x,y):
... return x+y
...
>>> x = [0,1,2,3]
>>> y = [4,5,6,7]
>>>
>>> p.map(add, x, y)
[4, 6, 8, 10]
>>>
>>> class Test(object):
... def plus(self, x, y):
... return x+y
...
>>> t = Test()
>>>
>>> p.map(Test.plus, [t]*4, x, y)
[4, 6, 8, 10]
>>>
>>> p.map(t.plus, x, y)
[4, 6, 8, 10]
获取代码:https: //github.com/uqfoundation/pathos
pathos还有一个异步map(amap),以及imap.
| 归档时间： |
|
| 查看次数： |
6121 次 |
| 最近记录： |