相关疑难解决方法(0)

很抱歉,我无法用更简单的示例重现错误,而且我的代码太复杂而无法发布.如果我在IPython shell而不是常规Python中运行程序,那么事情就会很顺利.

我查看了之前关于这个问题的一些注意事项.它们都是由在类函数中定义的pool to call函数引起的.但对我来说情况并非如此.

Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib64/python2.7/threading.py", line 552, in __bootstrap_inner
    self.run()
  File "/usr/lib64/python2.7/threading.py", line 505, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 313, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

我将不胜感激任何帮助.

更新:我挑选的功能是在模块的顶层定义的.虽然它调用包含嵌套函数的函数.即f()要求g()调用h()具有嵌套函数i(),和我打电话pool.apply_async(f).f(),g(),h()都在顶层定义.我用这个模式尝试了更简单的例子,但它确实有效.

我想使用multiprocessing的Pool.map()功能,同时划分出工作.当我使用以下代码时,它工作正常:

import multiprocessing

def f(x):
    return x*x

def go():
    pool = multiprocessing.Pool(processes=4)        
    print pool.map(f, range(10))


if __name__== '__main__' :
    go()

但是,当我在面向对象的方法中使用它时,它不起作用.它给出的错误信息是:

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

当以下是我的主程序时会发生这种情况:

import someClass

if __name__== '__main__' :
    sc = someClass.someClass()
    sc.go()

以下是我的someClass课程:

import multiprocessing

class someClass(object):
    def __init__(self):
        pass

    def f(self, x):
        return x*x

    def go(self):
        pool = multiprocessing.Pool(processes=4)       
        print pool.map(self.f, range(10))

任何人都知道问题可能是什么,或者一个简单的方法呢？

当我运行类似的东西:

from multiprocessing import Pool

p = Pool(5)
def f(x):
     return x*x

p.map(f, [1,2,3])

它工作正常.但是,将此作为类的函数:

class calculate(object):
    def run(self):
        def f(x):
            return x*x

        p = Pool()
        return p.map(f, [1,2,3])

cl = calculate()
print cl.run()

给我以下错误:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/sw/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/sw/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/sw/lib/python2.6/multiprocessing/pool.py", line 225, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

我看过Alex Martelli的一篇文章处理同样的问题,但它不够明确.

我必须像这样挑选一个对象数组:

import cPickle as pickle
from numpy import sin, cos, array
tmp = lambda x: sin(x)+cos(x)
test = array([[tmp,tmp],[tmp,tmp]],dtype=object)
pickle.dump( test, open('test.lambda','w') )

它会出现以下错误:

TypeError: can't pickle function objects

有办法吗？

我知道在处理作为类的数据成员的函数时(由于Pickling问题),多处理模块的局限性的各种 讨论.

但是,在多处理中是否存在另一个模块或任何类型的解决方案,它允许特定的类似下面的内容(特别是不强制并行应用函数的定义以存在于类外)？

class MyClass():

    def __init__(self):
        self.my_args = [1,2,3,4]
        self.output  = {}

    def my_single_function(self, arg):
        return arg**2

    def my_parallelized_function(self):
        # Use map or map_async to map my_single_function onto the
        # list of self.my_args, and append the return values into
        # self.output, using each arg in my_args as the key.

        # The result should make self.output become
        # {1:1, 2:4, 3:9, 4:16}


foo = MyClass()
foo.my_parallelized_function()
print foo.output

注意:我可以通过移动my_single_function到类之外并将类似的东西传递foo.my_args给map或map_async …

我正在对我的代码应用一些并行化,我在其中使用类.我知道如果不采用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] …

嗨,我一直在为这个早上的大部分时间而苦苦挣扎,希望有人能指出我正确的方向.

这是我目前的代码:

def f(tup):
    return some_complex_function(*tup)

def main():

    pool = Pool(processes=4) 
    #import and process data omitted 
    _args = [(x.some_func1, .05, x.some_func2) for x in list_of_some_class]
    results = pool.map(f, _args)
    print results

我得到的第一个错误是:

> Exception in thread Thread-2: Traceback (most recent call last):  
> File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner
>     self.run()   File "/usr/lib/python2.7/threading.py", line 504, in run
>     self.__target(*self.__args, **self.__kwargs)   File "/usr/lib/python2.7/multiprocessing/pool.py", line 319, in
> _handle_tasks
>     put(task) PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup __builtin__.instancemethod failed

任何帮助将非常感激.

我有汤,BeautifulSoup因为我不能泡菜.当我尝试pickle对象时,python解释器静默崩溃(这样它就不能作为例外处理).我必须能够pickle对象,以便使用multiprocessing包返回对象(pickles对象在进程之间传递它们).我该如何解决/解决问题？不幸的是,我无法发布该页面的html(它不公开),我一直无法找到问题的可重现的例子.我试图通过循环汤和酸洗单个组件来隔离问题,产生错误的最小的东西是<class 'BeautifulSoup.NavigableString'>.当我打印对象时,它打印出来u'\n'.

我目前正在使用python中的标准多处理来生成一系列无限期运行的进程.我并不特别关心表现; 每个线程只是在观察文件系统上的不同更改,并在修改文件时采取适当的操作.

目前,我有一个适合我的需求的解决方案,适用于Linux.我有一个函数和参数的字典,如下所示:

 job_dict['func1'] = {'target': func1, 'args': (args,)}

对于每一个,我创建一个过程:

 import multiprocessing
 for k in job_dict.keys():
     jobs[k] = multiprocessing.Process(target=job_dict[k]['target'],
                                       args=job_dict[k]['args'])

有了这个,我可以跟踪每个正在运行的,并在必要时重新启动因任何原因崩溃的作业.

这在Windows中不起作用.我正在使用的许多函数都是包装器,使用各种functools函数,我得到的消息是无法序列化函数(请参阅多处理和dill可以一起做什么？).我还没弄清楚为什么我在Linux中没有得到这个错误,但在Windows中也是如此.

如果我dill在Windows中启动进程之前导入,则不会出现序列化错误.但是,这些过程实际上并没有做任何事情.我无法弄清楚为什么.

然后我切换到多处理实现pathos,但没有找到Process标准multiprocessing模块中的简单类的模拟.我能够使用每个作业生成线程pathos.pools.ThreadPool.这不是map的预期用途,我敢肯定,但它启动了所有线程,并且它们在Windows中运行:

import pathos
tp = pathos.pools.ThreadPool()
for k in job_dict.keys():
    tp.uimap(job_dict[k]['target'], job_dict[k]['args'])

但是,现在我不确定如何监视一个线程是否仍处于活动状态,我正在寻找这样,以便我可以重新启动由于某种原因崩溃的线程.有什么建议？

我正在处理经常需要使用 pythonmultiprocessing Pool类的代码。这会产生大量如下所示的代码：

import time
from multiprocessing import Pool
from functools import partial

def test_func(x):
    time.sleep(1)
    return x

def test_func_parallel(iterable, processes):
    p = Pool(processes=processes)
    output = p.map(test_func, iterable)
    p.close()
    return output

这可以变得更一般：

def parallel(func, iterable, **kwargs):
    func = partial(func, **kwargs)
    p = Pool(processes=6)
    out = p.map(func, iterable)
    p.close()
    return out

这是可行的，但向其他每个函数添加并行包装器会使代码复杂化。我真正想要的是让它作为装饰器工作。像这样的东西：

def parallel(num_processes):
    def parallel_decorator(func, num_processes=num_processes):
        def parallel_wrapper(iterable, **kwargs):
            func = partial(func, **kwargs)
            p = Pool(processes=num_processes)
            output = p.map(func, iterable)
            p.close()
            return output

        return parallel_wrapper
    return parallel_decorator

可以按如下方式使用： …

我正在尝试使用多处理来处理我的内存问题，但是我无法获取要腌制的函数，而且我不知道为什么。我的主要代码开始于

def main():
    print "starting main"
    q = Queue()
    p = Process(target=file_unpacking,args=("hellow world",q))
    p.start()
    p.join()
    if p.is_alive():
        p.terminate()
    print "The results are in"
    Chan1 = q.get()
    Chan2 = q.get()
    Start_Header = q.get()
    Date = q.get()
    Time = q.get()
    return Chan1, Chan2, Start_Header, Date, Time

def file_unpacking(args, q):
    print "starting unpacking"
    fileName1 = "050913-00012"
    unpacker = UnpackingClass()
    for fileNumber in range(0,44):
        fileName = fileName1 + str(fileNumber) + fileName3
        header, data1, data2 = UnpackingClass.unpackFile(path,fileName)

        if header == None:
            logging.warning("curropted file found at …

我在一个类中有一个方法来返回一个参数可能会改变的函数。

Interface函数接受两个参数，f和它的args。我想用mp.pool来加速它。但是，它返回一个错误。

from multiprocessing import Pool
# from multiprocess import Pool
# from pathos.multiprocessing import ProcessingPool as Pool
import pickle
import dill


class Temp:
    def __init__(self, a):
        self.a = a

    def test(self):
        def test1(x):
            return self.a + x

        return test1


def InterfaceFunc(f, x):
    mypool = Pool(4)
    return list(mypool.map(f, x))


if __name__ == "__main__":
    t1 = Temp(1).test()
    x = [1, 2, 3, 1, 2]

    res1 = list(map(t1, x))
    print(res1)

    res2 = InterfaceFunc(t1, x)

它引发了同样的错误：

AttributeError: Can't pickle local object 'Temp.test.<locals>.test1'

我尝试了 …