1つのプログラム内で並列処理をするタイミングを決め、並列処理をしないときは1つの関数だけ動かしたい

Question

前提・実現したいこと

python3.8
ubuntu20.04を使用しています。

executor.submit(m.process)では1回のみ動画を回していて、その動画から得られた結果からexecutor.submit(m.process)を動かしつつexecutor.submit(r.run)で動画の解析をしています。また、動画の解析が終わったら終了し、executor.submit(m.process)だけにしていきたいです。

executor.submit(m.process)は動画再生のため、1回の実行だけでよいのですがexecutor.submit(r.run)は動画の結果（count_image_pathは動画から得られた画像のパス）に応じて何度も実行する必要があります。解析するときだけ並列処理になるというプログラムを書くことはできますでしょうか。

申し訳ございませんが宜しくおねがい致します。

    def process(self):
        # Defining world dimensions
        world_size = (720, 900)
        image_path ="/program/main/resize_map.jpg"
　　　　#マップの読み込み
        image = cv2.imread(image_path)
　　　　#スタート地点
        agent = Agent(Position(720,75), scan_radius=0.60, possible_moves=30)
        #第1ゴール地点
        goal = Goal(Position(588,218), sigma=math.sqrt(world_size[0]**2 + world_size[1]**2))
　　　　#描写
        agent.draw(image)
        goal.draw(image)
        cv2.imshow('Output', image)
        cv2.moveWindow('Output', 1950, 100)
        cv2.waitKey(100)
        
#ポテンシャル法による画像に点描写するプログラム
        navi  = 0
        args = self.arg_parse()
        confidence = float(args.confidence)
        nms_thesh = float(args.nms_thresh)
        CUDA = torch.cuda.is_available()
        num_classes = 80
        print("Loading network.....")
        model = Darknet(args.cfgfile)
        model.load_weights(args.weightsfile)
        print("Network successfully loaded")
        if CUDA:
            model.cuda()
#ここまでしか来ず、「進みますがターミナルに出力されません
#whileループ内のループというのが無限ループ
        # ストリーム(Color/Depth)の設定
        config = rs.config()
　　　　#撮影した動画を読み込む
        config.enable_device_from_file('/48A9A14C31288186/12_2_thu.bag',repeat_playback=False)
        config.enable_stream(rs.stream.color, 640, 480, rs.format.bgr8, 30)
        config.enable_stream(rs.stream.depth, 640, 480, rs.format.z16, 30)
        
        # ストリーミング開始
        pipeline = rs.pipeline()
        profile = pipeline.start(config)
        align = rs.align(rs.stream.color)
        print("進みます")
        playsound('/home/limlab/program/voice/go_straight.mp3')
        try:
#ゴールまで0.3以上の距離があるのでゴールに向かって点の描写が始まる
            while Position.calculate_distance(agent.position, goal.position) > 0.3:
                try:
                    frames = pipeline.wait_for_frames()
                    color_frame = frames.get_color_frame()
                    depth_frame = frames.get_depth_frame()
                    if not depth_frame or not color_frame:
                        continue
                    frames = align.process(frames)
                    profile = frames.get_profile()
                    #画像データに変換    
                    color_image = np.asanyarray(color_frame.get_data())
                    depth_color_frame = rs.colorizer().colorize(depth_frame)
                    depth_color_image = np.asanyarray(depth_color_frame.get_data())
#これ以降,点描写がされつつ、yoloによる認識が始まり認識した場合、count_image_pathが保存されます、

def main():
    if os.path.exists(count_image_path):
        print("count_image_pathが存在するので削除します")
        os.remove(count_image_path)
    if os.path.exists(text_reading_image_path):
        print("text_reading_image_pathが存在するので削除します")
        os.remove(text_reading_image_path)    
    m = Main()
    executor = concurrent.futures.ProcessPoolExecutor(max_workers=2)
    executor.submit(m.process)
    r_running = False

    while True:
        print("ループ")
        # プロセスrが動作していなくて、プロセスmの情報ができたらプロセスrを起動
        if not r_running and os.path.exists(count_image_path)== True:
            r = Run_image()
            executor.submit(r.run)
            r_running = True
      
        # プロセスrが終了したら、フラグを落す。
        if os.path.exists(text_reading_image_path):
            r_running = False
               

        # プロセスmが終章したらループを抜ける
        #if not r_running and #プロセスmの終了検知:
            #break

Comments

[y_waiwai]

で、しつもんはなんでしょうか

[退会済みユーザー]

返信ありがとうございます。
また、拙い質問で申し訳ございません。
1つのプログラム内で並列処理をするしないを条件分岐したいと考えていますがどのように変えればいいでしょうか。そもそも、このうようなことは可能でしょうか。

[y_waiwai]

質問文にそれを追記しましょう

今の質問文ではなにをしたいのかわかりません

[退会済みユーザー]

質問の題名で「1つのプログラム内で並列処理をするしないを条件分岐したい」を書き、実現したいでその詳細を書いたつもりでしたが説明不足で申し訳ございません。

はじめは1つ目の関数だけ実行し、条件により1つ目と2つ目の関数を同時に起動し、2つ目の関数の処理が終わり次第終了し、1つ目の関数のみの実行に切り替えたいです。
特にエラーがでているということではないんですがなぜファイルが存在するのに並列処理に入らないのかわからない状態です。

Answer 1

こういう動作にすればいいのではないかな。

まず、プロセスは3種類あります。
・ メインのプロセス
このプログラムのプロセス
・ プロセスm
m.processを実行するプロセス
プロセスrで処理するファイルを作成する
・ プロセスr
r.runを実行するプロセス
m.processが作った情報を処理する。

ざっくり書くと、メインプロセスはこう動きます

- プロセスmを起動する。
- ループに入る
  - プロセスmが終了していたらループを抜ける
  - ファイルができていたら、プロセスrを起動する
    - プロセスrの終了を待ち合わせる
  - ちょっと待つ

ようするに、メインのプロセスで他のプロセスの起動と終了をちゃんと管理する必要があるということです。

---

質問を受けて追記します。

プロセスrが動作しているかどうかを示す情報、そういうのをフラグと呼びます、をつかって状態を管理すればいいです。

フラグとして、r_runningを作ってFalseで初期化します。
あとは、そのフラグも見ながら動作を調整していきます。

    executor.submit(m.process)

    r_running = False

    while True:

        # プロセスrが動作していなくて、プロセスmの情報ができたらプロセスrを起動
        if not r_running and os.path.exists(count_image_path)== True:
            executor.submit(r.run)
            r_running = True
            # ここで消してしまっていいのか不明 プロセスrで使うのでは？
            os.remove(count_image_path)
        
        # プロセスrが終了したら、フラグを落す。
        if os.path.exists(text_reading_image_path):
            r_running = False
            os.remove(text_reading_image_path)    

        # プロセスmが終章したらループを抜ける
        if not r_running and #プロセスmの終了検知
            break

        #ここに少しスリープを入れたほうがいいでしょう。

Comments

[退会済みユーザー]

ご丁寧な返信ありがとうございます。
初歩的な質問で申し訳ございません。
こちらはconcurrent.futures.ProcessPoolExecutorこちらを使わずに実行する形でしょうか。
また、その場合はプロセスrを起動中、プロセスｍも同時に実行されますか。

[TakaiY]

concurrent.futuresを使うのであれば、PooExecuterを使うのがいいと思います。
基本的なプロセスの呼び出しは今のやりかたでいいと思います。

逆に、使わないってのがどういうやりかたを想定しているのかわかりません。

[退会済みユーザー]

ご返信ありがとうございます。
下記のコードですとループに入る前は必ずFalseとなっているのでTrueになるまで待つということが必要？なのだと思いますがわからない状態です。
申し訳ございませんがご教示いただけないでしょうか。
よろしくお願いいたします。
def main():
if os.path.exists(count_image_path):
print("count_image_pathが存在するので削除します")
os.remove(count_image_path)
if os.path.exists(text_reading_image_path):
print("text_reading_image_pathが存在するので削除します")
os.remove(text_reading_image_path)
m = Main()
r = Run_image()
　 executor.submit(m.process)
　 while True:
if os.path.exists(count_image_path)== True:
executor.submit(r.run)
os.remove(count_image_path)

[退会済みユーザー]

ご丁寧な解説、大変感謝しております。
また、質問させてもらう形になり申し訳ございません。
質問の修正で追加させてもらいましたがなぜか動画が再生されない状態です。ですので、動画が再生されないため解析もできず画像(count_image_path)が保存されず、ずっとr.runがループしている状態となっています。
なぜ、このようなことが起きているのでしょうか。
申し訳ございませんがよろしくお願いいたします。

[TakaiY]

「なぜか動画が再生されない状態」というのが何のことなのかよくわからないのですがたぶん、プロセスmで処理されるんだと思います。
そもそも、その、プロセスmは単体で動作させたときにちゃんと動くのでしょうか？
また、サブプロセスとして動作させたとき「executor.submit(m.process)」にはどうでしょう？

[退会済みユーザー]

度々申し訳ございません。
単体で起動は可能です。
また、下記のようにしますとr.runはcount_image_pathがないためかすぐ終了する形となっています。
m = Main()
r = Run_image()
executor = concurrent.futures.PoolExecutor(max_workers=2)
executor.submit(m.process)
executor.submit(r.run)

[TakaiY]

上記のときにはm.processは動作しているのでしょうか？

これが動作していないとすると、m.processの動かしかたたがよくないのではないかと思います。

[退会済みユーザー]

process関数内での下記の出力は出力されるので動いているのではないかと思います。
print("Network successfully loaded")
また、ThreadPoolExecutorこちらを使用しますと、executor.submit(m.process)起動し、認識後にこちらもexecutor.submit(r.run)起動します。ただすごく遅い状態です。

[TakaiY]

ProcessPoolにすると「print("進みます")」の表示は出ないんですよね？
ThreadPoolだと出るんですか？

そこを解決しないと先には進まないでしょうね。

[退会済みユーザー]

>ProcessPoolにすると「print("進みます")」の表示は出ないんですよね？
そうですね、進みますまで出ない状態です。

>ThreadPoolだと出るんですか？
質問に掲載しているコードをThreadPoolに変えますと進みますの表示もでて動いています。

このような問題はパソコンの性能に依存しますでしょうか。
また、ただプログラムの問題なのでしょうか。
申し訳ございませんが宜しくおねがい致します。

[TakaiY]

残念ながら、そこになると僕にはよくわかりません。

並列処理にしたいという話と離れてくるので、Threadだと動くのにProcessだと動かないということで別質問にしてみてはいかがでしょう。
それにあたっては、Mainクラスのソースも上げたほうがいいでしょうね。

[退会済みユーザー]

承知しました。
私の拙い質問でも親切に対応していただきありがとうございました。

[退会済みユーザー]

先日は、お世話になりました。
容量の大きい、gpuに変えたとこと無地思ったように動くことが出来ました。
ありがとうございました！

Answer 2

ThreadPoolExecutorの代わりに、ProcessPoolExecutorを使ってください。

スレッドの場合はglobal interpreter lockがあるので同時実行できません。

グローバルインタプリタロック (Global Interpreter Lock) を取り除くことはできないのですか？も合わせてお読みください。

Comments

[退会済みユーザー]

ご返信ありがとうございます
こちらを
concurrent.futures.ThreadPoolExecutor
こちらに変更しますと
concurrent.futures.ProcessPoolExecutor
どちらの関数も1回のみの起動となってしまいます。
executor.submit(r.run)こちらを必要に応じて何度も呼び出すにはどうしたら良さそうでしょうか。
申し訳ございませんが宜しくおねがい致します。