Python - asyncio.Event().waitの意義

Question

Pythonで非同期ネットワークIO処理実装にむけて勉強しています。
以下は、サーバから複数の__Websocket__配信を非同期的に受信するコードです（asyncio, aiohttp使用）。
（コード自体は私が一から書いたものではなく、色々なリソースを参考にしながらつなぎ合わせたものです。動作OK。）

わからないこと

asyncio.Event().waitが何のためにあるのか、どのようなケースで活用すべきなのか、いまいち理解できません。

以下コード#(1), #(2)はあっても無くても挙動は変わりません。
また、ドキュメントによると、以下は理解できましたが、具体的にどのような場合に活用すべきなのかイメージが湧きません。

• EventオブジェクトはBoolフラグで管理されている（デフォルトはFalse, set()が呼ばれたらTrue）
• wait()は、EventオブジェクトがTrueになるまで待つ

import aiohttp
import asyncio

async def coro(event,
               item1,
               item2):
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(url='URL') as ws:
            event.set()                                    # (2)
            await asyncio.gather(ws.send_json(item1),
                                 ws.send_json(item2))
            async for msg in ws:
                print(msg)

async def ws_connect(item1,
                     item2):
    event = asyncio.Event() 
    task = asyncio.create_task(coro(event, item1, item2))
    await event.wait()                                    # (1)
    return task

async def main():
    item1 = {some json object}
    item2 = {some json object}

    ws_task = await ws_connect(item1, item2)
    await ws_task

asyncio.run(main())

Answer 1

解決済みですが、応用的な話題で

---

以下コード#(1), #(2)はあっても無くても挙動は変わりません。

接続が成功しなかった場合の、実際の挙動が変わります。

---

作用が判り難いのは、main -> ws_connect -> coro だと「直列」な呼び出しで、
～の完了後に処理したいだけなら await した後にコード書けば同じだったり、
もしくは、add_done_callback で後処理を登録したりもできます。
（違い・使い分けは、コードのスコープだったり例外時やキャンセル時の挙動）

非同期処理においては、 双方の関数のスコープを継続したまま 、
待ち合わせする為の同期機構として使えます。

質問のコードのケースでも、有効な使い道はあり

# 接続成功後に event.set() されているので
# 接続に成功する迄の時間のタイムアウト
await asyncio.wait_for(event.wait(), timeout=10)

# イベントによる通知がないと
# タスク全体のタイムアウト（タイムアウト時の、接続成功の合否はわからない）
await asyncio.wait_for(task, timeout=10)

通常の、同期コードでは、失敗時は例外を出して中断することになりますが、
asyncio等のイベント駆動では、例外の捕捉を呼び出し元からは、
対象のコルーチン（を含むタスク）を直接 await する迄できません。

# asyncio での例外の扱い
await task1 # この間(task1完了前)に task2 内部で例外が起きても
await task2 # task2 からの例外が発生するのは、コード上ではこちら

どのようなケースで活用するかの解としては、
「コルーチン間でイベントを通知する手段」として使うことが出来ます。

同期プリミティブは、同期処理を実現するための手段の一つなので、
アイデア次第で他にも様々な用途で使うことが出来ます。

• コルーチン内でのループの一時停止・再開

async def process(event):
    for num in range(100):
        await event.wait()
        print(f"{num=}")

# 外部からの操作で event set()/clear() で、一時停止・再開

• 任意のタイミングでキャンセル可能な sleep の実装

　(wait_for と組み合わせてタイムアウト処理)

async def past(event, within):
    """
    Examples
    --------

        if await past(event, within=10):
            ... # 10 秒以内に event.set() が呼ばれた。
        else:
            ... # 10 秒以上経過。タイムアウト時
    """
    try:
        await asyncio.wait_for(event.wait(), timeout=within)
    except asyncio.TimeoutError:
        pass
    # NOTE: マルチスレッドではないので、
    # この間に await がない限り event.set() が呼ばれることはない
    return event.is_set()

参考までに、マルチプロセスやマルチスレッドでも同樣のクラス群が提供されており、
同期プリミティブは、独自の同期機構を作るための部品として扱われます。
より具体的な応用例は：「同期」や「排他制御」辺りの用語がヒント。

例えば、独自の sleep を Event を使って実装する場合、
n秒後に event.set が呼ばれるように設定された event として実装できます。

async def my_sleep(delay):
    # NOTE: event を引数で与えるようにすれば、外部からも中断可能に。

    event = asyncio.Event()
    loop = asyncio.get_running_loop()

    if delay <= 0:
        loop.call_soon(event.set)
    else:
        loop.call_later(delay, event.set)
    await event.wait()

※用例を示すもので、実際の asyncio.sleep の実装では有りません。
実際の asyncio.Event, asyncio.sleep は、どちらも より低階層の API future で実装されてます。

---

存在意義としては、他の言語・ライブラリでも同樣の概念が確立されているので、
「車輪の再発明」を避ける為にも有用です。Pythonのライブラリ内でも

threading.Event, multiprocessing.Event に同じインターフェースを持つクラスがあり、
それぞれ スレッド間、プロセス間、で同樣の役割を果たします。

asyncio.Event は、それらに対して コルーチン間でのイベント通知・同期機構を提供します。
⇛ マルチスレッド・マルチプロセスのコードを asyncio へ移植する際に、
共通の部品が使えると便利。他の部品は、Lock や Queue 等。

Comments

[koyamashinji]

検証しており返信が遅くなりました。いつも大変有難うございます。応用例を示して頂き大変勉強になりました。eventの主な役目は、「コルーチン間で 任意の場所でイベントを通知し合うこと（eventがない場合、コルーチン内で発生したイベントは 対象のコルーチンをawaitするまで捉えられない）」と理解しました。

Answer 2

asyncio.Event().waitが何のためにあるのか.どのようなケースで活用すべきなのか

使い道としては、他のコルーチン非同期タスクを待ち受ける用途があります。

例：
worker1は作業1を、worker2は作業2を同時並行で行いますが
worker2は最終作業も行う必要があり、
なおかつ最終作業は作業1が完了していないと行えないものとします。

下記のコードでは、worker2は、worker1の作業1が完了するまで(worker1がevent.set()するまで）、event.wait()のところでブロックされ、最終作業を開始できません。

import asyncio
import random

async def worker1(event):
    interval = random.randint(1,8)
    print(f"worker1:　作業1を開始します。{interval}秒かかります...")
    await asyncio.sleep(interval)
    print(f"worker1:　作業1 が完了しました。worker2に完了を通知します。」")
    event.set()

async def worker2(event):
    interval = random.randint(1,8)
    print(f"worker2:　作業2を開始します。{interval}秒かかります...")
    await asyncio.sleep(interval)
    print(f"worker2:　作業2が完了しました。")
    if not event.is_set():
        print(f"worker2:　作業1 が完了するのを待っています...")
        await event.wait()
    print(f"worker2:　作業1の完了通知を受領しました。")
    print(f"worker2:　最終作業を開始します。2秒かかります...")
    await asyncio.sleep(2)
    print(f"最終作業が完了しました。")


async def main():
    event = asyncio.Event() 
    _worker1 = asyncio.create_task(worker1(event))
    _worker2 = asyncio.create_task(worker2(event))
    await asyncio.gather(_worker1,_worker2)
    

asyncio.run(main())

Comments

[koyamashinji]

いつもありがとうございます。例が非常に分かりやすく理解できました。取り急ぎお礼申し上げます。