asyncioによる非同期処理

止まらずに待つ

データを扱うプログラムは、多くの時間を計算ではなく待機に費やします。価格サーバーの応答、注文の受付確認、ダウンロードの完了などです。10個の価格を1つずつ求めれば、10回続けて待つことになります。非同期コードでは、1つのスレッドの中で、リクエストを出して待っている間は脇に置き、別の処理を進められます。そのための Python の道具が asyncio で、async def と await の2つのキーワードでできています。

# A coroutine: calling it does nothing until it is awaited
import asyncio

async def get_price(symbol):
    await asyncio.sleep(0.01)                  # stands in for waiting on a network answer
    return {"AAPL": 184.25, "MSFT": 410.5}[symbol]

coro = get_price("AAPL")
print(type(coro).__name__)                     # coroutine  nothing has run yet
print(asyncio.run(coro))                       # 184.25     the event loop runs it

async def、await、イベントループ

async def で定義した関数はコルーチン関数で、呼び出すとコルーチンオブジェクトが返り、本体は何も実行されません。中で await 何か と書くと、その何かが用意できるまでコルーチンを一時停止し、制御をイベントループに返します。その間にループは別のコルーチンを実行できます。asyncio.run(coro) はループを起動してコルーチンを最後まで実行します。await は async def の中でしか使えません。書き忘れが典型的な誤りで、答えの代わりにコルーチンが返ってきます。

# Awaiting ten prices one after another waits ten times; gather overlaps the waits
import asyncio
import time

async def get_price(symbol):
    await asyncio.sleep(0.05)
    return len(symbol) * 10.0

async def one_by_one(symbols):
    return [await get_price(s) for s in symbols]

async def together(symbols):
    return await asyncio.gather(*(get_price(s) for s in symbols))

symbols = ["AAPL", "MSFT", "NVDA", "TSLA", "AMZN", "META", "GOOG", "NFLX", "ORCL", "INTC"]
t0 = time.perf_counter()
asyncio.run(one_by_one(symbols))
slow = time.perf_counter() - t0
t0 = time.perf_counter()
asyncio.run(together(symbols))
fast = time.perf_counter() - t0
print(f"one by one: {slow:.2f} s, gathered: {fast:.2f} s")      # about 0.50 s against 0.05 s
print(slow > 4 * fast)                                           # True

エラー、タイムアウト、手で動かす

待機は失敗することがあります。応答しないリクエストは asyncio.wait_for(coro, timeout) で打ち切り、TimeoutError を送出させます。try / except の中の await は通常のコードと同じようにエラーを処理するので、リトライは待機する呼び出しを囲むループになります。内部では、コルーチンは1歩ずつ進められるものにすぎません。coro.send(None) は次の一時停止まで実行し、終了すると結果を持つ StopIteration を送出します。イベントループは、これを多数のコルーチンについて繰り返し続けるコードで、このラボのテストがあなたのコードを動かす方法でもあります。

# Stepping a coroutine by hand: this is what an event loop does for many of them
class Pause:
    def __await__(self):
        yield

async def job():
    await Pause()
    await Pause()
    return 184.25

coro = job()
steps = 0
while True:
    try:
        coro.send(None)
    except StopIteration as stop:
        print("finished with", stop.value)                  # finished with 184.25
        break
    steps += 1
print("paused", steps, "times")                              # paused 2 times

# A timeout turns a slow answer into an error you can handle, then retry
import asyncio

async def slow_feed():
    await asyncio.sleep(1.0)
    return 184.25

async def main():
    for attempt in range(1, 4):
        try:
            return await asyncio.wait_for(slow_feed(), timeout=0.01)
        except TimeoutError:
            print(f"attempt {attempt} timed out")
    return None

print(asyncio.run(main()))                                   # three timeouts, then None

• async def はコルーチン関数を定義する。呼び出してもコルーチンが作られるだけで、まだ何も実行されない
• await は結果が用意できるまで一時停止し、その間イベントループが他を実行する。async def の中でのみ使える
• asyncio.run がループを起動し、gather が待ちを重ね、wait_for がタイムアウトを加える。エラーは通常の try / except で扱う

次はWebSocket によるストリーミングデータです。ティックのライブフィードをリアルタイムで受け取ります。

ラボ：採点される内容

• get_price で1つの気配を待つ25点
• get_prices でウォッチリストを取得する25点
• retry_price でタイムアウトを乗り切る25点
• drive でイベントループになる25点