質問

ループ内で多くの作業を行う必要があるクラス内のメソッドがあり、その作業をすべてのコアに分散させたいと考えています。

私は以下のコードを書きました。 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()

エラー1

AttributeError: ローカルオブジェクト 'SomeClass.some_method..single' を pickle できません。

なぜ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: モジュール''single''の属性が取得できません。 メイン ' from '.../test.py'.

どのように解決するのですか？

エラー1。

AttributeError: ローカル・オブジェクトをピクルスにできません 'SomeClass.some_method..single'です。

このエラーを自分で解決するには、ネストしたターゲット関数 single() をトップレベルに移動することで解決しました。

背景を

プールは、ワーカープロセスに送信するすべてのものを pickle (シリアライズ) する必要があります ( IPC ). pickling は実際には関数名を保存するだけで、unpickling は関数名で関数を再インポートする必要があります。そのためには、関数はトップレベルで定義されている必要があり、ネストされた関数は子プロセスではインポートできず、すでにそれらをピクルスにしようとすると例外が発生します ( より ).

エラー2です。

AttributeError: Can't get attribute 'single' on module 'main' from '.../test.py' から取得できません。

プールを起動している 前に そうすれば、子プロセスはコードを継承することができません。プールの起動を一番下に移動して、保護する ( なぜ？ で保護します。 if __name__ == '__main__':

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())

付録

...すべてのコアに作業を分散させたいのですが。

どのように役立つかについての潜在的な背景 multiprocessing.Pool がどのようにチャンキング作業をしているのかについて、役に立つかもしれません。

Pythonのマルチプロセシング：チャンクサイズの背後にあるロジックを理解する