Dask：遺伝子探索アルゴリズムを効率的に配布する方法は？

Question

私は遺伝子検索アルゴリズムを実装し、それを並列化しようとしましたが、ひどいパフォーマンス（シングルスレッドよりも悪い）を得ました。私はこれが通信オーバーヘッドのためだと思う。

の多くの反復実行次に、私は以下の擬似コードを提供してきたが、本質的に遺伝的アルゴリズムが「染色体」オブジェクトの大きなプールを作成：

1. はそれで実行する方法に基づいて、各個体の染色体のスコアを世界。'世界は反復の間、静的なままです。
2. はランダムので、それが自然に思えた、n個の反復
スコアリングアルゴリズム

（ステップ1）主要なボトルネックがあるためステップ1に、前のステップ

ゴーで計算し、そのスコアに基づいて、新たな人口を選択しますこのコードの処理を配布します。

私は私が助けを得ることができ望ん問題のカップルに遭遇してきた：

私はすなわち、各Futureをリンク、map()によるスコアリング関数に渡されたオブジェクトと算出されたスコアをリンクするにはどうすればよい

1. スコアをChromosomeに戻していますか？私はcalculate_scores()メソッドでオブジェクトを返すことで、非常に厄介な方法でこれを行っていますが、実際にはリンクを維持するための良い方法があればfloatを返信するだけです。
2. スコアリング関数の並列処理は問題ありませんが、map()がすべてのオブジェクトを反復処理するのに時間がかかります。しかし、その後のdraw_chromosome_from_pool()の呼び出しは、非常にで、私はまだ完了していないことが分かりました。私は、このメソッドが常にシングルスレッド版ですばやく完了するため、何が原因であるかわからない。すべての先物が完了した後であっても、染色体をローカルプロセスに引き戻すIPCがありますか？ローカルプロセスは何らかの形で優先順位を引き下げられていますか？
3. 私は、毎回プールを構築/再構築するという全体的な反復的な性質が、労働者に膨大な量のデータを送信することになると心配しています。この懸念の根源となる質問：Dask は実際にはが何時にワーカープールにデータを送りますか？つまり、Environment（）がChromosome（）に対して配布されたとき、そして結果がどのように/いつ戻ってくるのでしょうか？私はdocsを読んだが、正しい細部を見つけられなかったか、または理解するにはあまりにも愚かである。私がしたいことは、各労働者が「パーマ」に基づき、ローカルEnvironment()データを保持し、その後、Chromosome()インスタンスデータが少し戻って、重複して得点するために配布される分散型アーキテクチャである

理想的に、私は思う（しかし、補正にオープン）反復間の変化していないChromosome（）データの/

非常に長い投稿ですので、これを読んでいる時間があれば、すでにありがとうございます！

class Chromosome(object): # Small size: several hundred bytes per instance 
    def get_score(): 
      # Returns a float 
    def set_score(i): 
      # Stores a a float 

class Environment(object): # Large size: 20-50Mb per instance, but only one instance 
     def calculate_scores(chromosome): 
      # Slow calculation using attributes from chromosome and instance data 
      chromosome.set_score(x) 
      return chromosome 

class Evolver(object): 
    def draw_chromosome_from_pool(self, max_score): 
     while True: 
      individual = np.random.choice(self.chromosome_pool) 
      selection_chance = np.random.uniform() 
      if selection_chance < individual.get_score()/max_score: 
       return individual 

    def run_evolution() 
     self.dask_client = Client() 
     self.chromosome_pool = list() 
     for i in range(10000): 
      self.chromosome_pool.append(Chromosome()) 

     world_data = LoadWorldData() # Returns a pandas Dataframe 
     self.world = Environment(world_data) 

     iterations = 1000 
     for i in range(iterations): 
      futures = self.dask_client.map(self.world.calculate_scores, self.chromosome_pool) 
      for future in as_completed(futures): 
        c = future.result() 
        highest_score = max(highest_score, c.get_score()) 

      new_pool = set() 
      while len(new_pool)<self.pool_size: 
       mother = self.draw_chromosome_from_pool(highest_score) 
        # do stuff to build a new pool 

Answer 1

はい、それぞれの時間はあなたが大きいself.worldを、直列化されている

futures = self.dask_client.map(self.world.calculate_scores, self.chromosome_pool) 

ラインを呼び出します。あなたは

future_world = client.scatter(self.world, broadcast=True) 

とループの前に一度だけこの操作を行うと、ループ内

futures = self.dask_client.map(lambda ch: Environment.calculate_scores(future_world, ch), self.chromosome_pool) 

すでに労働者にコピー（または同じことを簡単な関数）を使用しますことができます。ポイントは、future_worldは既に配布されているものへのポインタに過ぎませんが、daskはこれを処理します。

as_completedを使用すると、あなたが提出した注文がmapに壊れていますが、このコードは必要ありません。すべての作業が完了した時点で処理する場合は、waitを使用したか、future.result()（各作業が完了するまで待つ）を繰り返して、chromosome_pool内でその順序を保持している可能性があります。