KnowHow

技術的なメモを中心にまとめます。
検索にて調べることができます。

並列処理プログラミングのメモ#6 (イベント)

登録日 ：2025/05/10 10:41
カテゴリ ：Python基礎

busy-waitingアプローチの欠点に対応する方法の一つとして「イベント」がある。
イベントベースの処理は、スレッドやプロセスではなく、イベント（メッセージ：例えばこの処理が終わったからリソース空いたよとメッセージを出すなど）を中心としてアプリケーションを構造化する。

イベントベースの並行処理や、Webサーバや、メッセージングシステム、ゲームプラットフォームなど多くのハイパフォーマンスアプリケーションで見つかる。

コールバック

コールバック：イベント駆動形プログラムにおいて、各イベントが発生した時に実行されるコード

コールバックベースのコードでは、制御フローがわかりにくくなったり、デバックが難しくなったりしがちである。（コードがもやは逐次処理ではないため。ロジックを複数のコールバックに分散すると、コードでの処理の連鎖により、コールバックが何重にも入れ子になることがあり、これをコールバック地獄という）

イベントループ

イベントループ：さまざまなイベントとそれに対応するコールバックを組み合わせて制御するエンティティ

イベントループでは、イベントは「イベントキュー」に追加される。
（busy-waitingの場合は、イベントポーリングを使っていた）

イベントループでは、イベントキューからイベントを取り出して適切なコールバックを呼び出すという処理を繰り返す。

以下のサンプルコードでは、イベントループを作成し、knockとwhoの２つのイベントを登録している（knockイベントがwhoイベントを生成できることに注意）。
続いて、knockイベントを二つ生成して、それがちょうど発生したかのような状況を再現し、イベントループを無限ループとして開始する。

"""
Chapter 11 / event_loop.py
"""
from __future__ import annotations

from collections import deque
from time import sleep
import typing as T


class Event:
    """
    Eventクラスはイベントループによって実行されるアクション
    """
    def __init__(self, name: str, action: T.Callable[..., None],
                 next_event: T.Optional[Event] = None) -> None:
        self.name = name
        self._action = action
        self._next_event = next_event

    def execute_action(self) -> None:
        self._action(self)
        if self._next_event:
            event_loop.register_event(self._next_event)


class EventLoop:
    def __init__(self) -> None:
        """
        イベントループによって実行されるイベントの格納先となるキューを作成
        """
        self._events: deque[Event] = deque()

    def register_event(self, event: Event) -> None:
        # イベントキューにイベントを追加
        self._events.append(event)

    def run_forever(self) -> None:
        """
        イベントループを無限ループとして開始
        キューに追加されたイベントをそれぞれ実行
        """
        print(f"Queue running with {len(self._events)} events")
        while True:
            try:
                event = self._events.popleft()
            except IndexError:
                continue
            event.execute_action()


def knock(event: Event) -> None:
    print(event.name)
    sleep(1)


def who(event: Event) -> None:
    print(event.name)
    sleep(1)


if __name__ == "__main__":
    event_loop = EventLoop()
    replying = Event("Who's there", who)
    knocking = Event("Knock-knock", knock, replying)
    for _ in range(2):
        event_loop.register_event(knocking)
    event_loop.run_forever()

戻る