使用多进程时出现PicklingError

Question

我在使用 Pool.map_async()（还有 Pool.map()）这个多进程模块时遇到了一些麻烦。我实现了一个并行循环的功能，只要传给 Pool.map_async 的是一个“普通”的函数，它就能正常工作。但是当这个函数是某个类的方法时，我就会遇到 PicklingError 错误：

cPickle.PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

我主要用Python进行科学计算，所以对“序列化”这个概念不太熟悉，今天刚学到一点。我看过一些之前的回答，比如使用多进程的 Pool.map() 时无法序列化 <type 'instancemethod'>，但即使跟着回答中的链接，我也搞不定。

我的代码是为了模拟一个正态随机变量的向量，目的是利用多个核心来运行。请注意，这只是一个例子，可能在多个核心上运行并不划算。

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)

在回答中提到的链接里，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_reg 正常工作，这样就能充分利用多进程，而不必每次都绕过这个问题。

代码调试并行计算序列化实例方法多进程科学计算 picklingerror 正态随机变量

使用多进程时出现PicklingError

1 个回答

撰写回答