Python multiprocessing.Pool : AttributeError

Question

我在一个类中有一个方法需要在一个循环中做很多工作，我想将这些工作分散到我的所有核心上。

我写了下面的代码，如果我使用普通的 map()，它可以工作，但是使用 pool.map() 会返回一个错误。

import multiprocessing
pool = multiprocessing.Pool(multiprocessing.cpu_count() - 1)

class OtherClass:
  def run(sentence, graph):
    return False

class SomeClass:
  def __init__(self):
    self.sentences = [["Some string"]]
    self.graphs = ["string"]

  def some_method(self):
      other = OtherClass()

      def single(params):
          sentences, graph = params
          return [other.run(sentence, graph) for sentence in sentences]

      return list(pool.map(single, zip(self.sentences, self.graphs)))


SomeClass().some_method()

错误一:

AttributeError: Can't pickle local object 'SomeClass.some_method..single'

为什么它不能 pickle single()？我什至尝试将 single() 移动到全局模块范围(不在类内部 - 使其独立于上下文):

import multiprocessing
pool = multiprocessing.Pool(multiprocessing.cpu_count() - 1)

class OtherClass:
  def run(sentence, graph):
    return False


def single(params):
    other = OtherClass()
    sentences, graph = params
    return [other.run(sentence, graph) for sentence in sentences]

class SomeClass:
  def __init__(self):
    self.sentences = [["Some string"]]
    self.graphs = ["string"]

  def some_method(self):
      return list(pool.map(single, zip(self.sentences, self.graphs)))


SomeClass().some_method()

我得到以下...

错误2:

AttributeError: Can't get attribute 'single' on module 'main' from '.../test.py'

Answer 1

错误一:

AttributeError: Can't pickle local object 'SomeClass.some_method..single'

您通过将嵌套的目标函数 single() 移到顶层自行解决了这个错误。

背景:

Pool 需要 pickle(序列化)它发送给它的工作进程的所有东西(IPC)。 Pickling 实际上只保存函数的名称，而 unpickling 需要按名称重新导入函数。为此，函数需要在顶层定义，嵌套函数将不能被子函数导入，并且已经尝试 pickle 它们会引发异常(more)。

---

错误2:

AttributeError: Can't get attribute 'single' on module 'main' from '.../test.py'

您在定义函数和类之前启动池，这样子进程就不能继承任何代码。将你的池开始移动到底部并使用 if __name__ == '__main__':

保护 ( why? )

import multiprocessing

class OtherClass:
  def run(self, sentence, graph):
    return False


def single(params):
    other = OtherClass()
    sentences, graph = params
    return [other.run(sentence, graph) for sentence in sentences]

class SomeClass:
   def __init__(self):
       self.sentences = [["Some string"]]
       self.graphs = ["string"]

   def some_method(self):
      return list(pool.map(single, zip(self.sentences, self.graphs)))

if __name__ == '__main__':  # <- prevent RuntimeError for 'spawn'
    # and 'forkserver' start_methods
    with multiprocessing.Pool(multiprocessing.cpu_count() - 1) as pool:
        print(SomeClass().some_method())

---

附录

...I would like to spread the work over all of my cores.

关于 multiprocessing.Pool 如何进行分 block 工作的可能有用的背景知识:

Python multiprocessing: understanding logic behind chunksize