Python 多处理 : calling methods and passing objects in asynchronous calls

Question

我正在尝试通过 apply_async ( https://docs.python.org/3/library/multiprocessing.html#multiprocessing.pool.AsyncResult ) 调用完成两件事:

(i) 调用类方法

(ii) 传递一个对象作为参数

到目前为止，我有以下基线代码:

import multiprocessing as mp

class myClass():
  def __init__(self, id):
    self.id = id
    self.val = 1.0
    self.pool = None

  def callback(self, obj):
    self.val = obj.val

def foo(new_val):  # foo is outside myClass
    print ('foo passed with', new_val)
    c1.val = new_val
    return c1

if __name__ == '__main__':
  c1 = myClass('c1')
  c1.pool = mp.Pool(processes=1)
  c1.pool.apply_async(foo, args=(2.0, ), callback=c1.callback).wait()  
  c1.pool.close()
  c1.pool.join()
  print ('c1.val:', c1.val)  # should display 'c1 val: 2.0'

输出:

foo passed with 2.0
c1.val: 2.0

当我尝试使用下面的代码完成 (i) 时，我没有得到与上面相同的输出。

class myClass():
  def __init__(self, id):
    self.id = id
    self.val = 1.0
    self.pool = None

  def callback(self, obj):
    self.val = obj.val

  def foo(self, new_val):  # foo is inside myClass
      print ('foo passed with', new_val)
      self.val = new_val
      return self

if __name__ == '__main__':
  c1 = myClass('c1')
  c1.pool = mp.Pool(processes=1)
  c1.pool.apply_async(c1.foo, args=(2.0, ), callback=c1.callback).wait()  
  c1.pool.close()
  c1.pool.join()
  print ('c1.val:', c1.val)  # should display 'c1 val: 2.0'

输出:

c1.val: 1.0

同样，当我尝试完成 (ii) 时，foo 不会再次被调用。

class myClass():
  def __init__(self, id):
    self.id = id
    self.val = 1.0
    self.pool = None

  def callback(self, obj):
    self.val = obj.val

def foo(obj, new_val):  # foo is outside myClass
    print ('foo passed with', new_val)
    obj.val = new_val
    return obj

if __name__ == '__main__':
  c1 = myClass('c1')
  c1.pool = mp.Pool(processes=1)
  c1.pool.apply_async(foo, args=(c1, 2.0, ), callback=c1.callback).wait()  
  c1.pool.close()
  c1.pool.join()
  print ('c1.val:', c1.val)  # should display 'c1 val: 2.0'

输出:

c1.val: 1.0

知道上面的代码需要更改什么才能完成 (i) 和 (ii) 吗？

Answer 1

调用未在未引发异常的情况下完成。您可以使用 multiprocessing.pool.AsyncResult.successful 检查方法:

import multiprocessing as mp


class myClass():
    def __init__(self, id):
        self.id = id
        self.val = 1.0
        self.pool = None

    def callback(self, obj):
        self.val = obj.val

    def foo(self, new_val):
        print ('foo passed with', new_val)
        self.val = new_val
        return self

if __name__ == '__main__':
    c1 = myClass('c1')
    c1.pool = mp.Pool(processes=1)
    async_result = c1.pool.apply_async(c1.foo, args=(2.0, ), callback=c1.callback)
    async_result.wait()
    print(async_result.successful())  # this is printing False!!!
    c1.pool.close()
    c1.pool.join()
    print ('c1.val:', c1.val)

现在您可以定义一个 error_callback 来查看发生了什么:

...
async_result = c1.pool.apply_async(c1.foo, args=(2.0, ), callback=c1.callback, error_callback=lambda x: print(x))
...

这是此函数打印的错误:

pool objects cannot be passed between processes or pickled

关于 this SO question您可以找到有关发生这种情况的原因的更多信息。问题是 multiprocessing 代码必须 pickle 发送给它已启动的子进程的东西，而 pickler 不执行实例方法。