一人で回すホール係：asyncioの第一歩

今夜からホールが変わります。ホール係はルー一人。テーブルは10卓です。シェフは人を増やしませんでした。

ルーは休まずホールを1周ずつ回ります。注文を取って厨房に通し、料理を待つ代わりにすぐ次のテーブルへ向かいます。厨房でベルが鳴ればその皿を持って出ていきます。1周、また1周。ベル（イベント）に反応しながら同じ道を回り続けるその姿に、初日から料理人たちがつけたあだ名がイベントループです。

ルーがエプロンのひもを結びながら言います。「テーブルの前に突っ立ってぼーっと待つのが、いちばん嫌なんです。待っているあいだ、ほかのテーブルが私を待ってるじゃないですか」

シェフがうなずきます。「昨日までは人を増やしてきた。今日は、一人が待つ時間をどう使うか。Pythonではasyncioって呼ぶんだ」

9-1. async defはコルーチンを作る (7分)

関数の前に async をつけるとコルーチン関数になります。まず不思議な点から見ていきましょう。呼んでも実行されないのです。

import asyncio

async def take_order(table):
    print("order", table)
    return f"T{table}"

c = take_order(1)
print(type(c).__name__)
print(c)
# coroutine
# <coroutine object take_order at 0x...>
# <sys>:0: RuntimeWarning: coroutine 'take_order' was never awaited

• take_order(1) は "order 1" を出力しませんでした。関数の本体は1行も動いていません。
• 返ってきたのはコルーチンオブジェクトです。「1番テーブルの注文を取ること」と書いておいたメモのようなものです。
• メモを一度も実行せずに捨てると、Pythonが最後に was never awaited の警告を出します。この警告が見えたら、ほぼ間違いなく await を書き忘れています。

メモを実行する方法は2つあります。いちばん外側では asyncio.run() に渡し、別のコルーチンの中では await します。

print(asyncio.run(take_order(1)))
# order 1
# T1

c = take_order(2)
asyncio.run(c)
asyncio.run(c)
# order 2
# RuntimeError: cannot reuse already awaited coroutine

コルーチンオブジェクトは一度しか使えません。同じ注文をもう一度するなら、take_order(2) をもう一度呼んで新しいメモを作ります。

9-2. awaitとasyncio.run (7分)

await は「この仕事が終わるまで、私はここで待つ。そのあいだループはほかの仕事をしていていい」という意味です。await は async def の中でしか使えません。普通の def の中に書くと SyntaxError: 'await' outside async function になります。

import asyncio
import time

async def table(n, wait):
    print(f"T{n} order")
    await asyncio.sleep(wait)   # 厨房を待っているあいだ、ルーは自由
    print(f"T{n} served")
    return f"T{n}"

async def one_by_one():
    a = await table(1, 0.3)
    b = await table(2, 0.3)
    return [a, b]

start = time.perf_counter()
print(asyncio.run(one_by_one()), f"{time.perf_counter() - start:.2f}s")
# T1 order
# T1 served
# T2 order
# T2 served
# ['T1', 'T2'] 0.62s

• asyncio.sleep(0.3) は time.sleep(0.3) のasyncio版です。待っているあいだ、ループに順番を譲ります。
• ところが結果は 0.3 + 0.3 = 約0.6秒。await table(1, ...) が終わらないと次の行に進まないからです。ルーが1番テーブルの前で、料理が出てくるまで立っているのと同じです。
• await は「同時に」ではなく「待つ」です。同時に待つには、次の節の道具が必要です。

asyncio.run(main()) はプログラムの入口です。

1. イベントループを新しく作る（ルーが出勤する）。
2. main() コルーチンを終わるまで回す。
3. 残ったタスクをキャンセルしてループを閉じる（ルーが退勤する）。

プログラムの中で普通は一度だけ、いちばん外側でだけ呼びます。すでにループが動いている場所でもう一度呼ぶと RuntimeError: asyncio.run() cannot be called from a running event loop になります。

9-3. イベントループがしていること (6分)

イベントループは、ひとつのスレッドで回るループです。していることは、ルーがホールを回るのと同じです。

1. いま進められるコルーチンをひとつ選ぶ。
2. そのコルーチンを次の await まで実行する。
3. コルーチンが何を待っているか（タイマー、ネットワークの応答）を書き留めて、次のコルーチンへ行く。
4. 進められるコルーチンがなければ、いちばん早く鳴るベルを待ち、そのコルーチンを起こし直す。

3卓を同時に任せると（次の節の gather）、ルーの動きはこうなります。1番は0.3秒、2番は0.2秒、3番は0.1秒かかる料理です。

2つのことを覚えておいてください。

• 人は一人です。スレッドはひとつだけで、どの瞬間にも実行中のコルーチンはひとつです。速くなるのは待ち時間が重なるからで、手が増えたからではありません。
• 順番は await でしか替わりません。スレッドはOSがいつでも割り込みますが（3限目）、コルーチンは自分で await したときにしか席を外しません。だから await と await のあいだのコードは、途中で切られることがありません。

9-4. gather：同時に待つ (8分)

asyncio.gather(コルーチン, コルーチン, ...) は、複数のコルーチンをいっぺんにループに載せ、全部が終わるまで待ちます。

async def together():
    return await asyncio.gather(table(1, 0.3), table(2, 0.2), table(3, 0.1))

start = time.perf_counter()
print(asyncio.run(together()), f"{time.perf_counter() - start:.2f}s")
# T1 order
# T2 order
# T3 order
# T3 served
# T2 served
# T1 served
# ['T1', 'T2', 'T3'] 0.31s

• かかった時間は合計（0.6秒）ではなく、いちばん長いもの（0.3秒）です。
• 終わった順番は T3 → T2 → T1 ですが、結果のリストは渡した順です。7限目の executor.map と同じです。
• コルーチンをいくつもリストにして渡すときは、gather(*coros) のようにアスタリスクで展開します。

下の実習台で「1卓ずつawait」と「gatherでまとめて」を切り替えながら回してみてください。1列が1卓で、バーはすべて「待ち」の色です。ルーが実際に手を動かす時間はほとんどなく、大半は厨房を待っている時間だからです。

• 1卓ずつ：テーブルの数だけ時間が延びます。10卓なら約2.6秒。
• gather：テーブルが1卓でも12卓でも約0.25秒です。1限目に料理人を8人雇って得た効果を、ホール係一人で手に入れました。
• この実習台も、サーバーで本物のasyncioループが回り、各テーブルが始まって終わった時刻を描いたものです。

1卓がしくじったら：return_exceptions

gather に渡したコルーチンのひとつが例外を出すと、gather はその例外をそのまま投げます。結果を全部受け取りたいなら return_exceptions=True を渡します。

async def table(n):
    await asyncio.sleep(0.1 * n)
    if n == 2:
        raise ValueError(f"T{n} wrong dish")
    return f"T{n}"

async def main():
    results = await asyncio.gather(table(1), table(2), table(3), return_exceptions=True)
    print(results)
    try:
        await asyncio.gather(table(1), table(2), table(3))
    except ValueError as e:
        print("ValueError:", e)

asyncio.run(main())
# ['T1', ValueError('T2 wrong dish'), 'T3']
# ValueError: T2 wrong dish

• return_exceptions=True なら、例外も結果リストの1枠に入ります。結果をひとつずつ isinstance(r, Exception) で確かめてください。
• デフォルト（False）のとき、例外が上がってきても残りのテーブルはキャンセルされず動き続けます。「ひとつが失敗したら残りを止めろ」は、10限目の TaskGroup がやってくれます。

9-5. create_task：先に任せておく (6分)

asyncio.create_task(コルーチン) は、コルーチンをタスクで包んですぐループに予約します。タスクはルーの手が空いたとき（いまのコルーチンが await したとき）に始まります。終わるのを待たずに次の行へ進むので、注文を先に通しておいて別の仕事ができます。

async def table(n, wait):
    print(f"T{n} order")
    await asyncio.sleep(wait)
    print(f"T{n} served")
    return f"T{n}"

async def main():
    start = time.perf_counter()
    t1 = asyncio.create_task(table(1, 0.3))
    t2 = asyncio.create_task(table(2, 0.2))
    print("Lou: 2 tasks created")
    await asyncio.sleep(0.1)          # ルーがちょっと別の仕事をする
    print("Lou: back", t1.done(), t2.done())
    print(await t1, await t2, f"{time.perf_counter() - start:.2f}s")

asyncio.run(main())
# Lou: 2 tasks created
# T1 order
# T2 order
# Lou: back False False
# T2 served
# T1 served
# T1 T2 0.31s

• "Lou: 2 tasks created" が T1 order より先に出ます。タスクは作った瞬間に予約されますが、初めて実行されるのは main が await で席を外したときです。
• await t1 は、すでに動いているタスクが終わるのを待ちます。2つのタスクが一緒に待ったので、0.5秒ではなく約0.3秒です。
• gather は受け取ったコルーチンを内部でタスクに包んでくれます。結果をまとめて受け取れば十分なら gather、個別に持っておいて状態を見たり後で待ったりするなら create_task を使います。

タスクは参照を持っておくこと

イベントループはタスクを弱参照でしか持っていません。変数やコレクションに入れていないタスクは、終わる前にガベージコレクションで消えてしまうことがあります。水のおかわりのように結果がいらない仕事を任せるときも、参照は握っておく必要があります。

background = set()

async def refill_water(table):
    await asyncio.sleep(0.1)
    print(f"T{table} water")

async def main():
    for n in (1, 2, 3):
        task = asyncio.create_task(refill_water(n))
        background.add(task)                         # 参照を握っておく
        task.add_done_callback(background.discard)   # 終わったら自分で抜ける
    print("pending", len(background))
    await asyncio.sleep(0.2)
    print("pending", len(background))

asyncio.run(main())
# pending 3
# T1 water
# T2 water
# T3 water
# pending 0

asyncio.create_task(refill_water(n)) のように結果を捨てる1行は、たいていうまく動きます。でも「たいてい」が問題です。ときどき消えるタスクは、いちばん見つけにくいバグです。

9-6. asyncio.sleepとtime.sleep (3分)

どちらも「ちょっと待つ」ですが、ルーにとってはまったく別のことです。

async def polite(n):
    await asyncio.sleep(0.3)   # 待っているあいだループに順番を譲る

async def stubborn(n):
    time.sleep(0.3)            # ループごと止まる

async def main(fn):
    start = time.perf_counter()
    await asyncio.gather(*(fn(n) for n in range(3)))
    print(fn.__name__, f"{time.perf_counter() - start:.2f}s")

asyncio.run(main(polite))
asyncio.run(main(stubborn))
# polite 0.32s
# stubborn 0.90s

• asyncio.sleep は、await した瞬間にループへ「0.3秒後に起こして」と頼んで席を外します。
• time.sleep はスレッドそのものを止めます。ループが回っているスレッドはひとつだけなので、ルーがテーブルの前に突っ立ってぼーっと待っているのと同じです。そのあいだ、ほかのテーブルは誰も見てくれません。
• だから3卓が0.3秒ずつ順番に、約0.9秒かかりました。gather を使っても意味がありません。

time.sleep だけでなく、時間のかかる計算や、待っているあいだに順番を譲らない古いライブラリも、同じようにループを止めます。この問題と解決策は11限目のテーマです。

9-7. コルーチン、タスク、フューチャー、そしてコスト (3分)

asyncioには似た名前が3つ出てきます。

print(issubclass(asyncio.Task, asyncio.Future))   # True

async def main():
    coro = take_order(1)
    task = asyncio.create_task(coro)
    print(type(coro).__name__, type(task).__name__)   # coroutine Task

7限目の concurrent.futures.Future と考え方は同じです。ただしasyncioのフューチャーは await でき、同じループの中でしか使いません。自分でフューチャーを作ることはまれで、普通はコルーチンとタスクで十分です。

タスクは軽い

スレッドはOSが作るものなので、1本ごとにスタックメモリとOSの管理コストがかかります。タスクはPythonのオブジェクトひとつにすぎません。

async def table(n):
    await asyncio.sleep(0.5)
    return n

async def main():
    return await asyncio.gather(*(table(n) for n in range(10_000)))

# 10,000 tasks: 約0.57秒   （0.5秒待つテーブル1万卓）
# 1,000 threads: 約0.62秒  （0.5秒待つスレッド1000本）
# このPCで測った値

タスク1万個を立ち上げるコストが、スレッド1000本より小さいのです。待つ仕事が何千、何万とあるなら（チャットサーバー、大量のURLへのリクエスト）、asyncioがよく合う理由です。

閉店後 (10分)

今夜の目標は、10卓の料理をホール係を増やさずに0.5秒以内に全部出すことでした。ルー一人、イベントループひとつ、gather ひとつで足ります。

import asyncio
import time

async def table(n):
    await asyncio.sleep(0.3)   # 厨房から料理が出てくるのを待つ
    return f"T{n}"

async def hall():
    return await asyncio.gather(*(table(n) for n in range(1, 11)))

start = time.perf_counter()
served = asyncio.run(hall())
print(served)
print(f"{time.perf_counter() - start:.2f}s")
# ['T1', 'T2', 'T3', 'T4', 'T5', 'T6', 'T7', 'T8', 'T9', 'T10']
# 0.30s

スレッドはひとつも作っていません。0.3秒のテーブル10卓が、約0.3秒で終わりました。

ルーが空のトレーを置きます。「10卓、1周で終わりました」シェフがタイマーを見ながら尋ねます。「いいね。でも明日は団体予約だよ。6卓のうち1卓がキャンセルになったら、残り5卓の料理はどうやって止める？」

練習問題

1. 9-1の take_order を await なしで呼ぶ行と、asyncio.run() で呼ぶ行を順番に実行し、出力と警告を比べてみましょう。
2. 9-5の main から await asyncio.sleep(0.1) を消すと、出力の順番はどう変わるでしょう？ 先に予想を書いてから、実行して確かめてみましょう。
3. 閉店後のコードのテーブルを100卓、1,000卓に増やして、かかる時間を測ってみましょう。同じことをスレッド1,000本でやるとどうなるかも比べてみてください。