アイドルasyncioイベントループを検出する

Question

asyncioイベントループが次の意味でアイドルになったときに検出できるプログラミングパターンがありますか？私の実行パスは、asyncio.gather（）を使用して複雑な方法で分岐するとしますが、各ブランチが最終的にはソケットやサブプロセスなどのアイドルコルーチンを待っていることを知っています。私はこれらのコルーチンが実際には得られないことを知っていると言えば、イベントループは実行可能なPythonコードを実行しますが、最終的にはそれらのアイドルコルーチンを待つだけです。このような状態を検出してループを停止するプログラム的な方法はありますか？

Answer 1

「アイドル」と呼ばれるものは、「IOまたはタイムアウトを待っている」と正確に記述することができます。正しく書かれたasyncioコードでは、という内容のがループしていることを検出する必要はありません。ループはその仕事をしており、asyncio.gather,asyncio.wait、loop.run_until_completeのようなツール適切なタイミングで終了するようにしてください。しかし、物事はいつも完璧というわけではありません。あなたが本当にそれをしたいのなら、確かに可能です。

イベントループの各ステップで、実行準備が整っているタスクをチェックします。もしあれば、そのステップが呼び出されます。タスクがもう準備できなくなると、イベントループはIOイベントまたは最も早いタイムアウトのいずれか早い方を待つ。重要なことは、実行中のタスクが常にIOを待つことよりも優先されることです。したがって、タスクが準備できていない場合を検出するために、すぐに発火することが知られているダミーIOイベントをスケジュールすることができます。

以下コルーチンは、このようなイベントを設定し、トリガすることを待つ：

import socket, asyncio 

async def detect_iowait(): 
    loop = asyncio.get_event_loop() 
    rsock, wsock = socket.socketpair() 
    wsock.close() 
    await loop.sock_recv(rsock, 1) 
    rsock.close() 

それが1つのソケットから読み出しが他のソケットに書き込まれたデータを返すsocket pairを設定します。これはただちにソケットの1つを閉じます。したがって、もう一方の読み取りからすぐに空のbytearrayとして表されるEOFが返されます。そのソケットからの読み取りを待つのは基本的に非ブロックですが、asyncioはそのことを知らないので、IO待機リストにソケットを配置します。上記のように、実行可能なタスクが存在しなくなると、asyncioはIOを待機し、detect_iowaitはソケットの読み込みと終了を待ちます。したがって、detect_iowait()が待機中であることを待機しています。 detect_iowait()を使用しています

テストコードは次のようになります。

# stop loop.run_forever once iowait is detected 
async def stop_on_iowait(): 
    await detect_iowait() 
    print('iowait detected, stopping!') 
    asyncio.get_event_loop().stop() 

# a dummy calculation coroutine, emulating your execution path 
async def calc(n): 
    print('calc %d start' % n) 
    async def noop(): 
     pass 
    for i in range(n): 
     await noop() 
    print('calc %d end' % n) 

# coroutine that waits on IO forever, also (ab)using a socket pair, 
# this time creating a socket whose recv will never complete 
async def io_forever(): 
    loop = asyncio.get_event_loop() 
    sock, _ = socket.socketpair() 
    sock.setblocking(False) 
    await loop.sock_recv(sock, 1) 

loop = asyncio.get_event_loop() 
for t in calc(1000), calc(10000), calc(100000), io_forever(): 
    loop.create_task(t) 
loop.create_task(stop_on_iowait()) 
loop.run_forever()