複数の並列処理を入れ子構造で動かしたい。

Question

前提・実現したいこと

複数の並列処理を入れ子構造で動かしたい。

一様分布の乱数を任意の回数処理するのを任意の回数行う
具体的イメージ：ランダム性のあるレース(ネズミの競争)を6レース分並列処理すること
レーサーの数は３とする。

###目的
複数の並列処理を入れ子構造で動かし、なるべく短時間で処理を行えるようにしたい。

知りたい事

下記コードが進行不可能な理由。←解決
下記コードを動かす改善方法。
他の有効な並列演算方法はあるかどうか。
daemonic processesとは何か。
追記：プロセッシングしてスレッディングすれば解決の可能性があるかもしれないとの事なので、サンプルコードを組んでいただけると助かります。
自身でもやってみます。

###環境
google cocolaboratory

発生している問題・エラーメッセージ

エラーメッセージ
AssertionError: daemonic processes are not allowed to have children

該当のソースコード

ソースコード

import pandas
import random
from multiprocessing import Pool

#####一様な乱数に従って任意回数思考した後、着順を返す関数の定義#####
def try_def(try_times):
  count = 3#count:レーサーの数
  #要素数:countの乱数を生成 = ネズミがゴールするまでにかかった時間（ラップタイム）
  rap_time_list = list()#リストとして初期化
  for i in range(count):
    rap_time_list.append(random.uniform(30,60))#30～60(秒)の乱数をリストに追加。
  #rap_time_listをデータフレームにし、0列目にラップタイムを持つデータフレームを作成。
  df1_df = pandas.DataFrame(rap_time_list)
  #上記データフレームにレーサー番号をつけたいので、indexを付与。
  df1_df = df1_df.reset_index()
  #0列目のラップタイムをソートして数字の小さい順（ゴール早かった順）にソート。
  df2_df = df1_df.sort_values(by=0)
  #レーサー番号もindexとして追従してソートされているので、利用して順位を取る。
  for i in range(count):
    race_result = [df2_df.iat[i,0] for i in range(count)]
  return race_result

#####レースを任意回数分(race_times回)行って結果リストを作成する関数の定義#####
def race_def(race_num):
  if __name__ == '__main__':
    race_times = 600
    initial_num_list = list(range(race_times))
    with Pool(processes=4) as p_para1:
      race_result_list = p_para1.map(func=try_def, iterable=initial_num_list)
  return race_result_list

#実行したいプログラム
if __name__ == '__main__':
  initial_num_list2 = list(range(6))
  with Pool(processes=6) as p_para2:
    race_result_list_2 = p_para2.map(func=race_def, iterable=initial_num_list2)
print(race_result_list_2)

調べたこと

並列演算コードの書き方など
入れ子構造で並列演算しているサイトを探したが見つからなかった。

補足情報（FW/ツールのバージョンなど）

環境：google cocolaboratory

プログラミング初めて１周間程度の知識しかないので、
すみませんが、わかりやすくお願いできましたら幸いです。

よろしくお願い致します。

Comments

[退会済みユーザー]

PCの調子が悪いのでイマイチ検証できていませんが、直感的に「ん？」という部分がいくつか見つかりました。
・　try_timesがdef try_def(try_times)のなかで呼び出されてないです。
・　with Poolはプロセスを複数のプロセスで請け負う処理ですので、普通は入れ子構造の中で呼び出す（複数回呼び出す）ことはしません。
・　da1_dfとdf1_df は同じでしょうか？どちらかが定義されるより前に呼び出されています。
・　sort_valuesはdataframeのものでしょうか？import pandas as pd とか？

[hiroki91629]

ご連絡ありがとうございます。
・try_def(trytimes)は何でもいいから引数を１つ入れるとdef～returnまで実行するというイメージで作成しております。このような使い方はまずかったでしょうか。
・普通はしないのですか。。。他の方法、書き方などはございませんでしょうか。
・誤：da1_df[1] = list(range(count))　修正致します。
・import pandas入れておきます。

[退会済みユーザー]

編集ありがとうございます。こちらでも動作を試してみますね。

> ・try_def(trytimes)は何でもいいから引数を１つ入れるとdef～returnまで実行するというイメージで作成しております。このような使い方はまずかったでしょうか。
バグとは少しずれた話で、動かす分には全く問題ありません。

Answer 1

解決したソースコード

ソースコード
import concurrent.futures
import pandas
import random
from multiprocessing import Pool
import concurrent


def inner_concurrent(num: int):
  count = 3#count:レーサーの数
  #要素数:countの乱数を生成 = ネズミがゴールするまでにかかった時間（ラップタイム）
  rap_time_list = list()#リストとして初期化
  for i in range(count):
    rap_time_list.append(random.uniform(30,60))#30～60(秒)の乱数
  #rap_time_listをデータフレームにし、0列目にラップタイムを持つデータフレームを作成。
  df1_df = pandas.DataFrame(rap_time_list)
  #上記データフレームにレーサー番号をつけたいので、indexを付与。
  df1_df = df1_df.reset_index()
  #0列目のラップタイムをソートして数字の小さい順（ゴール早かった順）にソート。
  df2_df = df1_df.sort_values(by=0)
  #レーサー番号もindexとして追従してソートされているので、利用して順位を取る。
  for i in range(count):
    race_result = [df2_df.iat[i,0] for i in range(count)]
  return race_result


def outer_concurrent():
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(inner_concurrent, i) for i in range(600)]
    return futures


def main():
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(outer_concurrent) for _ in range(6)]

    for outer_f in futures:
        for inner_f in outer_f.result():
            print(f"result = {inner_f.result()}")


if __name__ == "__main__":
    main()

#皆様ありがとうございました。
こちらのコードで動きましたので、こちらの元となった記事をご紹介頂いた方をベストアンサーとしたいと思います。

素人の感想ですが、インナープロセス（入れ子の中）の方を「ProcessPoolExecutor」として、アウターのプロセス（メインの方）は「ThreadPoolExecutor」とすると、うまくいくようです。
逆にすると、元の記事にあったように、「TypeError: can't pickle _thread.RLock objects」と、エラーが出ます。

Comments

[退会済みユーザー]

試さないと詳しいことは言えませんが、ProcessPoolとThreadPoolでハードウエアの中のリソースの割き具合が違うらしく、それに起因して得意な処理不得意な処理があるそうです（重い処理を少数、軽い処理を多数等）。あとは並列であるのはjoblibでしょうか。使いやすいものを使えばいいのかな？という気もします。

Answer 2

まずは複雑な処理を抜いて動かすところと思います。

（以下シンプルなサンプル）

import pandas
import random
from multiprocessing import Pool

#####一様な乱数に従って任意回数思考した後、着順を返す関数の定義#####
def try_def(try_times):
  count = 3#count:レーサーの数
  #要素数:countの乱数を生成 = ネズミがゴールするまでにかかった時間（ラップタイム）
  # rap_time_list = []
  rap_time_list = list()# ※リストとして初期化

  for i in range(count):
    
    # rap_time_list = rap_time_list + [random.uniform(30,60)]#30～60(秒)の乱数
    rap_time_list.append(random.uniform(30,60))#30～60(秒)の乱数 # ※リストに追加
    
  #rap_time_listをデータフレームにし、0列目にラップタイムを持つデータフレームを作成。
  # df1_df = rap_time_list
  df1_df = pandas.DataFrame(rap_time_list) # ※pandasのDataFrameに格納させました。

  
  #上記データフレームにレーサー番号をつけたいので、1列目に番号をつけてラップタイムと紐付ける。
  # df1_df[1] = list(range(count)) 
  df1_df = df1_df.reset_index() #これでインデックスを付与

  
  #0列目のラップタイムをソートして数字の小さい順（ゴール早かった順）にソート。
  df2_df = df1_df.sort_values(by=0)
  print(df2_df)
  
  #レーサー番号も追従してソートされているので、利用して順位を取る。
  #レーサー番号はdf2_dfの1列目に入っている。
  for i in range(count)  :
    print("Ranking:",i+ 1,"Index:",df2_df.iat[i,0] ," Record:",df2_df.iat[i,1])

  race_result = [df2_df.iat[i,0] for i in range(count)]
  
  return race_result

import concurrent.futures

#####レースを任意回数分(race_times回)行って結果リストを作成する関数の定義#####
def race_def(race_num):
  if __name__ == '__main__':
    race_times = 600
    initial_num_list = list(range(race_times))


    # ※multiprocessingをする前に、普通の状態でエラーがない状態にしたいため、
    #   まずはmultiprocessingを解除しました。
    # with Pool(processes=4) as p_para1:
    #   race_result_list = p_para1.map(func=try_def,iterable=initial_num_list)
     race_result_list = try_def(initial_num_list)
  return race_result_list

#実行したいプログラム
if __name__ == '__main__':
  initial_num_list2 = list(range(6))


  # ※multiprocessingをする前に、普通の状態でエラーがない状態にしたいため、
  #   まずはmultiprocessingを解除しました。
  #  with Pool(processes=6) as p_para2:
  #   race_result_list_2 = p_para2.map(func=race_def, iterable=initial_num_list2)
  race_result_list_2 = race_def(initial_num_list2)

print(race_result_list_2)

私のところではMPが動かせなかったので推定での答えですが、

下記コードが進行不可能な理由。

MPの中でMPを走らせたことが怪しい

下記コードを動かす改善方法。

上のシンプルな例を参照ください。

他の有効な並列演算方法はあるかどうか。

processpoolexecutorとthreadpoolexecutorでしょうか。たとえばここ。
先の方の答えにもありますが、オーバーヘッドすると遅くなるので多用は禁物です。

daemonic processesとは何か。

親スレッドがいて、子スレッドがいます。子スレッドを「デーモン」扱いにすると、親スレッドが終了したときに子スレッドも道連れにして終了してくれるスレッドです。多分掲載のコードだと子スレッドが親スレッドになろうとしてエラーをはいているのではないか、と思います。（検証できていないため推測ですよ！）

Comments

[hiroki91629]

ご回答ありがとうございます。
コード例もご丁寧にありがとうございます。
きちんと動きました。
（そのままコピペだと「race_result_list = try_def(initial_num_list)」の前にスペースが1個入っちゃうようでしたので、そこだけ修正すれば動きました）

ところで、大変恐縮なのですが、並列処理で動かしたく存じます。
そちらもサンプルコードをお出し頂けましたらベストアンサーとしたいと存じます。

「processpoolexecutorとthreadpoolexecutorでしょうか。たとえばここ。」と、改善案を頂きましたので、こちら、自身で試行してみようと存じます。

Answer 3

説明は以下です。

"AssertionError: daemonic processes are not allowed to have children"について

ただ、このように実行しても性能は上がらないでしょう。
並列処理は魔法ではありません。
あなたのPCのハードウェア性能、サーバがあなたに割り当ててくれるCPU資源の制限があります。
無駄にプロセスを増やすことは、プロセス生成と終了のオーバーヘッドを増やし、むしろ性能を悪化させます。

Comments

[hiroki91629]

ご回答、誠にありがとうございます。

追加の質問で恐縮ですが、添付URL中にAssertionErrorの無視ができるとありますが、調べてもよくわかりませんでした。ご解説頂けますと大変助かります。

性能が悪化しても良いので、何とか入れ子構造で並列処理できませんでしょうか。

もしくは、デーモンプロセスではない並列演算の方法などございましたらお教え頂けますと幸いです。

以上よろしくお願い致します。