【python】concurrent.futuresを使った並列処理における戻り値の利用方法について

Question

前提・実現したいこと

下記のソースコードにおいて、func1とfunc2を並列処理させ、func1とfunc2それぞれの戻り値を使ってfunc3,func4を実行したいと考えています。その際、func3,func4についてもできれば並列処理させたいと考えていますが、func3,func4の中でfunc1とfunc2の戻り値をどのように利用すればよいのかがわかりません。アドバイスいただけましたらありがたいです。よろしくお願いいたします。

該当のソースコード

#MCP3208
from concurrent.futures import ThreadPoolExecutor
import RPi.GPIO as GPIO

#GPIOに関するコードは割愛

def func1(): 
    while True:
        while (GPIO.input(up_pin) == 1):
          
        while (GPIO.input(down_pin) == 1):
           
        while (GPIO.input(measurement_pin) == 1):
            num = 0
            while num <=500:
　　　　　　     num += 1
                t_seg = ...
                time.sleep(0.05)
    return num, t_seg

def func2(): 
    while True:
　　　　for i in range(5000):
           count = ...
           time.sleep(0.001)
    return count

def func3():
    x=analog_read(0)
    #func1の戻り値(num)と*func2の戻り値(count)を使って処理したい
    #calc_a=x*num*count*... をこの関数の戻り値としたい

def func4():
    y=analog_read(1)
    #func1の戻り値(num,t_seg)を使って処理したい
    #calc_b=y*count*... をこの関数の戻り値としたい

if __name__=="__main__":
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread1") as executor:
        for _ in range(1000):
            executor.submit(func1)
            executor.submit(func2)
            

試したこと

func3,func4がないソースコードでは、並列処理できていることを確認できています。
(その際、func1およびfunc2の戻り値は"なし"にしています）

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

ここにより詳細な情報を記載してください。

Answer 1

• future = executor.submit(func)
• future.add_done_callback で関数の実行が終わった後に呼び出される関数を登録
• 引数future の result() で戻り値を取り出せます
• callback は複数指定できます

import time
from concurrent.futures import ThreadPoolExecutor

def func1():
    time.sleep(3)
    return 5, 10

def func2(future):
    num, count = future.result()
    print(count)

def func3(future):
    num, count = future.result()
    print(num)

with ThreadPoolExecutor() as e:
    future = e.submit(func1)
    future.add_done_callback(func2)
    future.add_done_callback(func3)

Answer 2

concurrent.futuresのドキュメントやqiitaを参考にして書いてみました。参考になれば。

import concurrent.futures
from concurrent.futures import ThreadPoolExecutor
from time import sleep

def func1():
    print("in func1().")
    num = 1
    t_seg = 2
    sleep(3)
    print("out func1().")
    return num, t_seg

def func2():
    print("in func2().")
    count = 3
    sleep(2)
    print("out func2().")
    return count

def func3(num, t_seg):
    print("in func3(). num={} t_seg={}".format(num, t_seg))
    sleep(4)
    print("out func3().")
    return True

def func4(count):
    print("in func4(). count={}".format(count))
    sleep(2)
    print("out func4().")
    return False

if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread1") as executor:
        futures  = []
        futures.append(executor.submit(func1)) # return are num, t_seg
        futures.append(executor.submit(func2)) # return is count
        concurrent.futures.wait(futures)
        print([f.result() for f in futures])

        num = futures[0].result()[0]
        t_seg = futures[0].result()[1]
        count = futures[1].result()
        futures = []
        futures.append(executor.submit(func3, num, t_seg))
        futures.append(executor.submit(func4, count))
        concurrent.futures.wait(futures)
        print([f.result() for f in futures])

実行結果です。

in func1().
in func2().
out func2().
out func1().
[(1, 2), 3]
in func3(). num=1 t_seg=2
in func4(). count=3
out func4().
out func3().
[True, False]

Comments

[Uka]

@ikapyさん
詳細にコード教えてくださりありがとうございます。
いただいたコードを参考に組みなおしていますが、whileやfor文を入れているせいか、戻り値の受け渡しがうまくいきません（ 上記コードのprint("out func3().")と print("out func3().")が表示されません、当然print([f.result() for f in futures])も表示されません）このあたりはいろいろやってみようと思います！

[ikapy]

そうですか。わたくしの場合は、質問者さんのものを単純化したものですからね。でもこれはこれで提示した甲斐があると思っています。
リアルでは提示されたGPIOはまたこれで壁があるかもしれません。でもこのコンカレント技法は、私が暇で行っているスクレイピングに応用ができるのはでと思っていて良かったです。
concurrent.futuresのソースは見ていませんが、別プロセスとするのであれば理解が早いのですがそこらへんがよくわかってません。
マルチスレッド、マルチプロセスの処理はリソースの扱いが難しいです（concurrentが別プロセスに展開するのであれば扱いやすいと思っています）。
実際には、func1、func2を単体として確実に動作確認するのが良いかもしれません。本ケースでは、デバッガを使うことはありませんでしたが、コンカレントではデバッガの利用も難しいかもしれません。
一つ々、積み重ねていくのが大切だと思います。
頑張ってくださいませ。

[ikapy]

追記。
こういう場合は、ダサくても要所々にprintで情報出力してください。