マルチプロセッシング、pool.readyは何をしますか？

Question

は、私がこのようないくつかの処理を、行うために使用するクラスの内部にいくつかのプロセスとプールがあるとします。

class MyClass: 

    def __init_(self): 
     self.pool = Pool(processes = NUM_PROCESSES) 
     self.pop = [] 
     self.finished = [] 

    def gen_pop(self): 
     self.pop = [ self.pool.apply_async(Item.test, (Item(),)) for _ in range(NUM_PROCESSES) ] 
     while (not self.check()): 
      continue 
     # Do some other stuff 

    def check(self): 
     self.finished = filter(lambda t: self.pop[t].ready(), range(NUM_PROCESSES)) 
     new_pop = [] 
     for f in self.finished: 
      new_pop.append(self.pop[f].get(timeout = 1)) 
      self.pop[f] = None 
      # Do some other stuff 

私はこのコードを実行すると、私は<type 'function'>ができないと述べてcPickle.PicklingErrorを取得しますピックルされる。これは、apply_async関数の1つがまだ返されていないので、実行中の関数を別のリストに追加しようとしていることを示しています。しかし、実行中のすべての呼び出しがready()関数を使用してフィルタにかけられている必要があるため、これは起こってはなりません。

Itemクラスの実際の性質は重要ではありませんが、私のItem.test関数の先頭には、デバッグの目的で起動するprint文があります。しかし、それは起こらない。これは、関数が開始されたが、実際には実行を開始していないことを示しています。

それで、ready()は、呼び出しが実行を終了したかどうかを実際に教えてくれないようです。 ready()とは何を正確に実行するのですか？まだ実行中のプロセスを除外できるようにコードを編集する方法はありますか？

Answer 1

マルチプロセッシングは はので、あなたのデータはpickle化可能なにする必要があり、プロセス間でデータを渡すために、内部pickleモジュールを使用しています。 the list of what is considered picklableを参照してください。オブジェクトメソッドはそのリストにありません。

def wrap_item_test(item): 
    item.test() 

class MyClass: 
    def gen_pop(self): 
     self.pop = [ self.pool.apply_async(wrap_item_test, (Item(),)) for _ in range(NUM_PROCESSES) ] 
     while (not self.check()): 
      continue 

Answer 2

を.ready()が本当に.get()がブロックすることがありますかどうかを語っている、あなたが尋ねた質問に答えるために：.ready()戻りTrue、.get()はない場合は、単にメソッドの周りにラッパー関数を使用してすぐにこの問題を解決するには
ブロックですが、.ready()がFalse、.get()、ブロックを返す場合（またはそうでない場合があります：を呼び出す前に非同期呼び出しが完了する可能性があります）。

だから、例えば、あなたの.get()でtimeout = 1は目的を達成しません：あなただけ.ready()がTrueを返した場合、あなたはすでに.get()がブロックしないという事実を知っている.get()を呼び出しているので。リモート呼び出しが発生した場合

、ドキュメントが言うように：ブロックしていない

しかし.get()はがは、非同期呼び出しが成功した、あるいはワーカープロセスも非同期呼び出し上で作業を開始していることだった意味するものではありませんその例外はget()によってリレイズされます。非同期呼び出しは全てでを実行できなかった場合、例えば、.ready()はTrueと.get()が返されている

は、（再）作動からの試みを防止例外が発生します。

あなたのケースでは何が起きているようですが、実行可能コードを投稿していないためにトレースバックを含めなかったので推測する必要があります。

Trueを.ready()に戻した後、非同期呼び出しが正常に完了したかどうかを実際に知りたい場合は、.successful()を呼び出してください。

Item.testが何であれ、それは.apply_async()の呼び出し可能なものとして渡すことは不可能であることは明らかです。 Item.testは何も（！それは実際に呼び出されることはありませんです）、なぜ.ready()戻りTrue（.apply_async()呼び出しが失敗した）、および（ - おそらくItem.testを.apply_async()は、引数の1つを酸洗中に例外が発生したため）.get()で例外が発生した理由を印刷したことがない理由を説明します。