threadPoolExecutorの使い方について

Question

asyncioとwebsocketを使ってデータを受信→加工しています。

　加工処理が重いため、データ受信とは別に加工の部分はバックグラウンドで動くようにしたいです。更に加工はデータを受け取った順に実行したいです。

以下のとおりコードを書いてみたのですが、maxworkerの数だけ実行され、さらには、次のメッセージ受信も5秒待ってから実行されます。

　5秒待たずして次のresponseを受けたいのですが、どのように表現すればいいでしょうか？

async def stream():
    uri = 'ws://127.0.0.1:5679/'

    async with websockets.connect(uri, ping_timeout=None) as ws:
        while not ws.closed:


            with ThreadPoolExecutor(max_workers=1, thread_name_prefix="thread") as executor:
                response = await ws.recv()# ここを滞りなく実行したいです。
                board = json.loads(response)
　　　　　　　　print(board) # スレッドのバックグラウンド処理とは別に、このプリントだけはリアルタイムで表示したい。 

                executor.map(event, board.get('id'))
                logger.info("submit end")
            logger.info("main end")


def event(board):
    print('これから5秒待つ' + str(board))
    time.sleep(5)
    print('wait終わり sleepなし')
    print(datetime.datetime.now())


loop = asyncio.get_event_loop()
loop.create_task(stream())
try:
    loop.run_forever()
except KeyboardInterrupt:
    exit()

これから5秒待つ
これから5秒待つ
これから5秒待つ
これから5秒待つ
wait終わり sleepなし
wait終わり sleepなし
wait終わり sleepなし
wait終わり sleepなし
2020-12-30 10:56:02.123801
2020-12-30 10:56:02.123801
2020-12-30 10:56:02.123801
2020-12-30 10:56:02.123801
これから5秒待つ
これから5秒待つ0
wait終わり sleepなし
wait終わり sleepなし
2020-12-30 10:56:07.135456
2020-12-30 10:56:07.135456
これから5秒待つ
これから5秒待つ
これから5秒待つ
これから5秒待つ

async def time(websocket, path):
    count = 0
    while True:
        now = datetime.datetime.utcnow().isoformat() + "Z"
        cid = f"{count:>6}"
        char = 'longlongstr'
        content = {'id':cid, 'message': char, 'time': str(now)}
        j_content = json.dumps(content)
        count += 1
        await websocket.send(j_content)
        await asyncio.sleep(random.random() * 3)


start_server = websockets.serve(time, "127.0.0.1", 5679)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

Comments

[退会済みユーザー]

加工する処理は、バックグラウンドで受け取った順に１つ１つ処理すればよく、多重に処理する必要はないという理解でよろしいでしょうか？
たとえば、１，２，３，４・・・という加工処理が発生した場合、この１，２，３，４・・・を同時に行うまでは必要なく、順番に処理していけばよいという理解でいいでしょうか

[taro_yamada]

お見込みの通りです。
やりたいことをソースコードに追記しました。

＜今回の課題は以下の点です＞
・データ受信が遅れてしまってwebsocketがtimeoutエラーになってしまうこと
・前回qnoir様にアドバイスいただいたawaitで進めていたのですが、中のasyinc def関数にawaitを使うものが一つもなかったのでうまく動かなかったこと。（私の調査ミスです。申し訳ありません。）

[taro_yamada]

イメージとしては受信したresponseをキューに入れて、バックグラウンドで先入先出でメソッドで処理していくようなイメージだと思っています。

Answer 1

ざっとThreadPoolExecutorのドキュメントを読んでみました。
ThreadPoolExecutorは、パラメーターを変えながら同じ処理を順序を無視して複数実行するための機能です。実行順序の保証はありません。

taro_yamadaさんのやろうとしていることは、データ受け取り(receiver)、データ処理(analyzer)についてそれぞれ一個のスレッドがあれば十分ですし、それぞれのスレッドが実行するのは決まった処理ですから、途中で新しいタスクを依頼する必要はありません。

やるべきことは、receiverスレッドを立ち上げて5秒眠っては受信するのを無限ループさせ、親スレッド側はanalyzer処理を行うことです。両スレッドは、排他制御した先入れ先出しキューで受け取ったデータを受け渡しし、キューが空ならanalyzer処理は何秒かスリープさせればよいです。
このためには、threadingモジュールで十分でしょう。

Comments

[taro_yamada]

threadingのスレッドとasyncioでのスレッドとどちらが効率的なのか判断が難しいですが、とりあえず、現状、動いているasyncioのスレッドで処理してみようと思います。
ありがとうございました。

Answer 2

実行を止めるためのフラグのところはお行儀が悪いかもしれませんがとりあえずの実装です。

現状、clientでの処理（5秒間隔で１つずつ）に対して、serverからの送出が早いため、どんどんタスクはたまっていきます。

client.py

(loggerとかの定義は質問でも省略されているので略）

# タスク処理屋
executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="thread")

# 処理を止めるためのフラグ
working = {'state':True}



# 受信部分
async def stream():
    uri = 'ws://127.0.0.1:5679/'
    async with websockets.connect(uri, ping_timeout=None) as ws:
        while not ws.closed:
            response = await ws.recv()
            board = json.loads(response)
            logger.info(f"[TASKID:{board.get('id')}] 受信しました。")
            executor.submit(event, board)
            logger.info(f"[TASKID:{board.get('id')}] タスクをキューに投入しました。")
 


def event(board):
    logger.info(f"[TASKID:{board.get('id')}] これから5秒待つ" + str(board))
    time.sleep(5)
    logger.info(f"[TASKID:{board.get('id')}] 処理が完了しました　[wait終わり sleepなし]")
    logger.info(datetime.now())


loop = asyncio.get_event_loop()
loop.create_task(stream())

try:
    loop.run_forever()
except KeyboardInterrupt:
    working['state'] = False
    exit()

Comments

[taro_yamada]

ありがとうございました。無事にうまく動き出しました。
logを見ているのが楽しいです。
ずっとこの問題で苦労していたので本当に助かりました。
今後ともよろしくお願いします。