asyncio ThreadPoolExecutor メソッド内からキューに値を入れる方法

Question

websocketのデータを受け取った後eventを起こすようなプログラムを書いています。
＜現状＞
streamでキューに入れたboardを元にprocess及びeventが動いている。

＜実現したいこと＞
event内で一定条件の時に別のキューにタスクを入れて別のprocess2及びevent2(数分バックグラウンドで待機した後処理を実行するようなメソッド)を実行したい

＜問題点＞
if文の中にawaitがあると'event' was never awaitedとなってしまう？　通常のdefでThreadPoolExecutorのqueueに方法があるのでしょうか？

import asyncio
import json
import websockets
import datetime
import logging


import time
from concurrent.futures import ThreadPoolExecutor

from asyncio import Queue

# タスクを先入先出処理するためのキュー　
que = Queue()
que2 = Queue()

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

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

# ---
formatter = '%(levelname)s : %(asctime)s : %(message)s'
logging.basicConfig(level=logging.INFO, format=formatter)
logger = logging.getLogger(__name__)

# 処理
async def process():
    while working['state']:
        board = await que.get()
        logger.info(f"[TASKID:{board.get('id')}] 処理を開始します。")
        executor.submit(event, board)

# 処理
async def process2():
    while working['state']:
        board = await que2.get()
        logger.info(f"[TASKID:{board.get('id')}] process2処理を開始します。")
        executor2.submit(event2, board)


# 受信部分
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)
            await que.put(board)
            logger.info(f"[TASKID:{board.get('id')}] タスクをキューに投入しました。")

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

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


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

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

import asyncio
import datetime
import random
import websockets
import json


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)
        await asyncio.sleep(1)

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()

＜結果＞
INFO : 2021-01-02 23:59:50,151 : [TASKID: 0] タスクをキューに投入しました。
INFO : 2021-01-02 23:59:50,151 : [TASKID: 0] 処理を開始します。
C:/Users/Owner/PycharmProjects/pythonProject/async_client2.py:35: RuntimeWarning: coroutine 'event' was never awaited
executor.submit(event, board)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
INFO : 2021-01-02 23:59:51,170 : [TASKID: 1] タスクをキューに投入しました。
INFO : 2021-01-02 23:59:51,170 : [TASKID: 1] 処理を開始します。
C:\Users\Owner\anaconda3\lib\concurrent\futures\thread.py:82: RuntimeWarning: coroutine 'event' was never awaited
del work_item
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
INFO : 2021-01-02 23:59:52,172 : [TASKID: 2] タスクをキューに投入しました。
INFO : 2021-01-02 23:59:52,172 : [TASKID: 2] 処理を開始します。
INFO : 2021-01-02 23:59:53,176 : [TASKID: 3] タスクをキューに投入しました。
INFO : 2021-01-02 23:59:53,176 : [TASKID: 3] 処理を開始します。

Answer 1

（以下コルーチン等の用語は、https://docs.python.org/ja/3/library/asyncio-task.html 内の使い方に準じています）

「 RuntimeWarning: coroutine 'event' was never awaited」が出る原因ですが、
process()内で、eventコルーチンをThreadpoolExecutor.submit()関数に渡しているためです。

エラーをなくすには、submit()関数に渡すのではなく、awaitをつけて呼び出します。

またeventコルーチン内で、time.sleep(5)としていますが、これだとメインスレッドを止めてしまうため、serverから送られてくるデータの受信処理もsleepの間止まってしまいます。
したがって、想定する動作にするためには、await asyncio.sleep(5)とする必要があります。

また、boardのidは桁を揃えた文字列となっているので、仮にserver側を変更しないならば
条件判定の部分は、

int(board.get('id').lstrip()) == 1

のように、空白を取り除いて数値に変換してやらないといけません。

以上をまとめたのが下記になります。

なお、id==1の場合、event()での処理（5秒)が終わってから、その約5秒後（que2に入ってから約10秒後）にevemt2()での処理が終わる形になっています。

つまりid==1のboardは、queに入った直後にque2にも入り、最初の5秒間は同時にevent()とevent2()での処理が走っていることになります。

client.py

import asyncio
import json
import websockets
import datetime
import logging
from datetime import datetime  # これがないとエラーになる。

import time
from concurrent.futures import ThreadPoolExecutor

from asyncio import Queue

# タスクを先入先出処理するためのキュー　
que = Queue()
que2 = Queue()

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

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

# ---
formatter = '%(levelname)s : %(asctime)s : %(message)s'
logging.basicConfig(level=logging.INFO, format=formatter)
logger = logging.getLogger(__name__)

# 処理
async def process():
    while working['state']:
        board = await que.get()
        logger.info(f"[TASKID:{board.get('id')}] 処理を開始します。")
        # executor.submit(event, board)  # 削除
        await event(board)               # 代わりにこうする。

# 処理
async def process2():
    while working['state']:
        board = await que2.get()
        logger.info(f"[TASKID:{board.get('id')}] process2処理を開始します。")
        executor2.submit(event2, board)


# 受信部分
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)
            await que.put(board)
            logger.info(f"[TASKID:{board.get('id')}] タスクをキューに投入しました。")

async def event(board):
    if int(board.get('id').lstrip()) == 1:  # 修正
        await que2.put(board)
    logger.info(f"[TASKID:{board.get('id')}] これから5秒待つ" + str(board))   # 実行順番を修正（修正前だと処理時間の表示が想定通りにならない）
    await asyncio.sleep(5)  # 修正
    logger.info(f"[TASKID:{board.get('id')}] 処理が完了しました　[wait終わり sleepなし] {board}")
    logger.info(datetime.now())

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


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

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

Comments

[taro_yamada]

ありがとうございいました。　無事動きました。

ただ、一つ心配なことがあるのですが、元々eventの処理をsidの順番で行いたかったのですが、asyncにしてしまうことによって、順番どおりに処理されなくなることはないでしょうか？

[退会済みユーザー]

event()での処理自体は、process()内で、queから順番に取り出して処理されるので、順番通りに処理されなくなることはありません。

[taro_yamada]

ありがとうございました！　本当に勉強になります。
実態のソースコードがかなり複雑化していて、しっかりと反映できるか少し不安ではありますが・・・、先が見えてきました！
引き続きよろしくお願いします。