在多处理模块中使用Pool.map_async()(以及Pool.map())时遇到问题。我实现了一个parallel for loop函数,只要Pool.map_async的函数输入是一个“常规”函数,它就可以正常工作。当函数是类的方法时,我得到一个PicklingError:
cPickle.PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed
我只把Python用于科学计算,所以我对pickling的概念不太熟悉,今天才了解了一点。我已经看了前面的几个答案,比如Can't pickle <type 'instancemethod'> when using python's multiprocessing Pool.map(),但是我不知道如何让它工作,即使在遵循答案中提供的链接时也是如此。
我的代码,目标是用多核模拟一个正常的r.v矢量。请注意,这只是一个例子,也许在多个内核上运行甚至没有回报。
import multiprocessing as mp
import scipy as sp
import scipy.stats as spstat
def parfor(func, args, static_arg = None, nWorkers = 8, chunksize = None):
"""
Purpose: Evaluate function using Multiple cores.
Input:
func - Function to evaluate in parallel
arg - Array of arguments to evaluate func(arg)
static_arg - The "static" argument (if any), i.e. the variables that are constant in the evaluation of func.
nWorkers - Number of Workers to process computations.
Output:
func(i, static_arg) for i in args.
"""
# Prepare arguments for func: Collect arguments with static argument (if any)
if static_arg != None:
arguments = [[arg] + static_arg for arg in list(args)]
else:
arguments = args
# Initialize workers
pool = mp.Pool(processes = nWorkers)
# Evaluate function
result = pool.map_async(func, arguments, chunksize = chunksize)
pool.close()
pool.join()
return sp.array(result.get()).flatten()
# First test-function. Freeze location and scale for the Normal random variates generator.
# This returns a function that is a method of the class Norm_gen. Methods cannot be pickled
# so this will give an error.
def genNorm(loc, scale):
def subfunc(a):
return spstat.norm.rvs(loc = loc, scale = scale, size = a)
return subfunc
# Second test-function. The same as above but does not return a method of a class. This is a "plain" function and can be
# pickled
def test(fargs):
x, a, b = fargs
return spstat.norm.rvs(size = x, loc = a, scale = b)
# Try it out.
N = 1000000
# Set arguments to function. args1 = [1, 1, 1,... ,1], the purpose is just to generate a random variable of size 1 for each
# element in the output vector.
args1 = sp.ones(N)
static_arg = [0, 1] # standarized normal.
# This gives the PicklingError
func = genNorm(*static_arg)
sim = parfor(func, args1, static_arg = None, nWorkers = 12, chunksize = None)
# This is OK:
func = test
sim = parfor(func, args1, static_arg = static_arg, nWorkers = 12, chunksize = None)
根据Can't pickle <type 'instancemethod'> when using python's multiprocessing Pool.map()中问题的答案中提供的链接,Steven Bethard(几乎在最后)建议使用copy_reg模块。他的密码是:
def _pickle_method(method):
func_name = method.im_func.__name__
obj = method.im_self
cls = method.im_class
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)
import copy_reg
import types
copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)
我真不明白我怎么能利用这个。我唯一能想到的就是把它放在我的代码前面,但是没有用。当然,一个简单的解决方案是只使用有效的解决方案,避免卷入copy_reg。我更感兴趣的是让copy嫒u reg正常工作,充分利用多处理,而不必每次都绕过问题。
谢谢你的帮助,非常感谢。
马蒂亚斯
这里的问题不是“pickle”错误消息,而是概念性错误消息: 多进程确实将代码分叉到“worker”不同的进程中,以便执行 它的魔力。
然后,它通过无缝地序列化和反序列化数据(即使用pickle的部分)来向不同的进程发送数据。
当来回传递的数据中有一部分是函数时,它假定被调用者进程中存在同名的函数,并且(我猜)将函数名作为字符串传递。由于函数是无状态的,被调用的工作进程只是用它接收到的数据调用同一个函数。 (Python函数不能通过pickle序列化,因此只在主进程和工作进程之间传递引用)
当您的函数是实例中的一个方法时——尽管当我们编写python代码时,它与函数非常相似,使用一个“automatic”
self
变量,但它在下面是不同的。因为实例(对象)是有状态的。这意味着工作进程没有对象的副本,该对象是要在另一端调用的方法的所有者。把方法作为函数传递给map_async调用的方法也行不通,因为多进程只是使用函数引用,而不是传递函数时使用的实际函数。
因此,您应该(1)更改代码,以便将函数(而不是方法)传递给工作进程,将对象保持的任何状态转换为要调用的新参数。 (2) 为映射异步调用创建一个“target”函数,该函数在工作进程端重建所需的对象,然后调用其中的函数。Python中最简单的类本身是可挑选的,因此可以在map_async调用中传递函数所有者本身的对象,而“target”函数将在工作端调用适当的方法本身。
(2)可能听起来“困难”,但可能只是这样-除非对象的类不能被pickle:
*免责声明:我没有测试过这个
相关问题 更多 >
编程相关推荐