ReflexKR

Reflex公式ドキュメント

深夜厨房の並行処理

講座の概要

初日の行列：並行と並列

料理人を増やす：threadingの基本

黒板の数が合わない：競合状態とLock

コンロは3つだけ：同期の道具

包丁とまな板：デッドロック

伝票差し：queue

出張料理人：concurrent.futures

生地はなぜ速くならない：GILと並列

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

団体予約のキャンセル：構造化された並行処理

止まったホール：ブロッキングと組み合わせ

評論家の夜：総まとめ

深夜厨房の並行処理

10限目

時間 7:30–8:20

団体予約のキャンセル：構造化された並行処理

今夜の注文書

#10 · 十日目の夜

団体予約のキャンセル

ピンチ

団体予約で6卓。3番テーブルがアレルギーの問題でコースを取り消すと、団体は全員で別の日に出直すと言う。ところが残り5卓の鍋はまだ煮え続けている。

目標

1卓が失敗したら残りの料理をすぐに止め、どの料理も決められた時間を超えないようにする。

深夜食堂ドンシ

夜10時半、団体客の6卓が一斉に席に着きます。ルーが注文を取って厨房に通し、厨房は6つのコースを同時に始めます。10分後、3番テーブルで手が挙がります。ピーナッツアレルギーを伝えるのが遅れたというのです。団体の代表が言います。「では今日は、みんなで失礼します」

ミンホはそれが聞こえず、5番テーブルのチゲを煮続けています。スアがミンホの火を止めながら言います。「1卓が抜けたら全部止める、なんてルール、どこにもなかったですよね」

ユン・ソヨンシェフがタイマーを押します。「一緒に始めた仕事は、一緒に終わらなきゃいけない。それをコードで約束するのが今夜の仕事」

10-1. gatherのすき間：1卓が失敗しても残りは煮え続ける (5分)

9限目の asyncio.gather で団体予約をさばいてみましょう。2番テーブルが0.1秒で失敗します。

import asyncio

async def table(n, sec, fail=False):
    await asyncio.sleep(sec)
    if fail:
        raise ValueError(f"T{n} allergy")
    print(f"T{n} served")

async def main():
    try:
        await asyncio.gather(table(1, 0.3), table(2, 0.1, fail=True), table(3, 0.5))
    except ValueError as e:
        print("caught:", e)
    await asyncio.sleep(0.6)   # 店をもう少し開けておく

asyncio.run(main())
# caught: T2 allergy
# T1 served
# T3 served

gather は最初の例外をすぐに上へ送ります。だから caught: が先に表示されます。
ところが残りの2卓はキャンセルされません。誰も待っていない料理が裏で煮え続け、served を表示します。
実際のプログラムなら、もう帰ったお客さんに料理を出したり、誰も結果を受け取らないリクエストが回り続けたりします。

gather(return_exceptions=True) は例外を結果のリストに入れてくれるだけで、残りを止めはしません。「1つが失敗したら全部止める」がほしいなら、別の道具が必要です。

10-2. TaskGroup：一緒に始めた仕事は一緒に終わる (12分)

asyncio.TaskGroup（3.11）はタスクをひとまとまりとして扱います。このようにまとめて管理するやり方を構造化された並行処理（structured concurrency）と呼びます。

import asyncio

async def table(n, sec):
    await asyncio.sleep(sec)
    return f"T{n}"

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(table(n, 0.1 * n)) for n in range(1, 4)]
    # ここに来たときには3つのタスクがすべて終わっている
    print([t.result() for t in tasks])

asyncio.run(main())
# ['T1', 'T2', 'T3']

TaskGroup の約束は3つです。

async with ブロックを抜けるとき、その中で作ったタスクはすべて終わっています。別に await しなくてかまいません。
タスクのひとつが例外を出すと、まだ動いている残りをすべてキャンセルします。
キャンセルが済んだら、失敗した例外を ExceptionGroup ひとつにまとめて送出します。

tg.create_task がタスクをつかまえておいてくれるので、9限目で見た「参照を失って消えるタスク」の心配もありません。では、3番テーブルが失敗する団体予約です。

import asyncio

async def table(n, sec, fail=False):
    try:
        await asyncio.sleep(sec)
        if fail:
            raise ValueError(f"T{n} allergy")
        print(f"T{n} served")
    except asyncio.CancelledError:
        print(f"T{n} cancelled")
        raise                      # 受け取ったキャンセルは送出し直す

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(table(1, 0.05))
            tg.create_task(table(2, 0.3))
            tg.create_task(table(3, 0.1, fail=True))
            tg.create_task(table(4, 0.5))
    except* ValueError as eg:
        print("group failed:", eg.exceptions)
    await asyncio.sleep(0.6)

asyncio.run(main())
# T1 served
# T2 cancelled      (T2とT4の順番は実行のたびに変わる)
# T4 cancelled
# group failed: (ValueError('T3 allergy'),)

T1は失敗より先に終わったので、そのまま提供されます。
T3が0.1秒で失敗した瞬間、まだ煮えていたT2とT4にキャンセルが入ります。キャンセルのメッセージが表示される順番は、実行のたびに変わることがあります。
0.6秒余分に待っても、served はもう表示されません。gatherと違って、裏でこっそり煮え続ける鍋はありません。
キャンセルされたタスクの中では、await の場所で asyncio.CancelledError が送出されます。だから except で受け取って後片づけができます。受け取ったキャンセルは必ず送出し直します（10-5）。

下のタイムラインは、サーバーで本物の TaskGroup を動かします。テーブルの番号が大きいほど、料理に少しずつ時間がかかります（T1が0.3秒、T2が0.345秒、… T6が約0.53秒）。「失敗するテーブル」に選んだテーブルは、自分の料理が終わるころに失敗します。

厨房タイムライン

注文

失敗するテーブル

なし

制限時間(秒)

オフ

調理

待ち

キャンセル

失敗

団体の6卓。失敗するテーブルが3なら、T1とT2が先に終わり、T3が赤い棒で終わった瞬間にT4〜T6が灰色（キャンセル）で断ち切られます。失敗するテーブルを1や6にも変えてみてください。「なし」ならすべて完了、制限時間を0.4秒ほどに縮めると、それまでに終わらなかったテーブルがキャンセルされます。

初期設定（失敗するテーブル3）では、T1とT2が先に終わり、約0.4秒でT3が失敗した瞬間、T4〜T6の3本の棒が同じ瞬間に灰色で途切れます。すでに終わったT1とT2はそのまま残ります。
失敗するテーブルを1に変えると、0.3秒で残りの5卓がすべて断ち切られ、6に変えると、5卓が全部終わったあとにT6だけが失敗します。
失敗を「なし」にすると、6本の棒が最後まで伸びます。いちばん遅いT6が約0.53秒で終わります。
制限時間を0.4秒ほどに縮めると、前のほうのテーブルがいくつか終わるだけで、制限時刻に残っていたテーブルは灰色で断ち切られます。すでにできた料理が取り消されることはありません。

10-3. ExceptionGroupとexcept* (7分)

複数のタスクが同時に失敗することもあります。だからTaskGroupは、例外ひとつではなく例外の束を送出します。

import asyncio

async def boom(exc, sec):
    await asyncio.sleep(sec)
    raise exc

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(boom(ValueError("T2 allergy"), 0.1))
            tg.create_task(boom(KeyError("T5"), 0.1))
    except Exception as e:
        print(type(e).__name__, "|", e)
        print(e.exceptions)

asyncio.run(main())
# ExceptionGroup | unhandled errors in a TaskGroup (2 sub-exceptions)
# (ValueError('T2 allergy'), KeyError('T5'))

ExceptionGroup（3.11）は、例外のリストを .exceptions に持つ例外です。束の中から種類ごとに選んで受け取るには except* を使います。

eg = ExceptionGroup("kitchen", [ValueError("T2 allergy"), KeyError("T5"), ValueError("T6 salt")])

try:
    raise eg
except* ValueError as g:
    print("ValueError:", [str(e) for e in g.exceptions])
except* KeyError as g:
    print("KeyError:", g.exceptions)
# ValueError: ['T2 allergy', 'T6 salt']
# KeyError: (KeyError('T5'),)

ふつうの except は1つしか実行されません。except* は当てはまるブロックがすべて実行されます。上の例では2つのブロックがどちらも動きます。
g はいつも ExceptionGroup です。当てはまる例外だけを選んで入れた小さな束です。
どの except* にも当てはまらなかった例外は、残りの束として再び送出されます。

eg = ExceptionGroup("kitchen", [ValueError("T2"), KeyError("T5")])
try:
    try:
        raise eg
    except* ValueError:
        print("handled ValueError")
except ExceptionGroup as rest:
    print("left:", rest.exceptions)
print(isinstance(eg, Exception), eg.subgroup(ValueError).exceptions)
# handled ValueError
# left: (KeyError('T5'),)
# True (ValueError('T2'),)

except* にはルールが2つあります。どちらも3.14では、実行前に構文エラーとして弾かれます。

src1 = '''
def f():
    try:
        pass
    except* ValueError:
        return 1
'''
src2 = '''
try:
    pass
except ValueError:
    pass
except* KeyError:
    pass
'''
for src in (src1, src2):
    try:
        compile(src, "<menu>", "exec")
    except SyntaxError as e:
        print("SyntaxError:", e.msg)
# SyntaxError: 'break', 'continue' and 'return' cannot appear in an except* block
# SyntaxError: cannot have both 'except' and 'except*' on the same 'try'

except* ブロックの中では return、break、continue が使えません。複数のブロックが順に実行されなければならないのに、途中で抜けられては困るからです。結果は変数に入れておき、try の外で返します。
ひとつの try に except と except* を混ぜることはできません。両方必要なら try を二重に分けます。この講座の実験台のエンジンもそう書いています。

シェフのひとこと

except*の中でreturnしたくなったら、結果を変数に書いておいて、tryの外で返して。厨房も同じだよ。事故の報告は全部聞いてから帰るもの。最初の報告だけ聞いて店を閉めて出ていっちゃだめ。

10-4. 制限時間：asyncio.timeoutとwait_for (8分)

今夜の2つ目のルールは「どの料理も決められた時間を超えない」です。asyncio.timeout()（3.11）はブロック全体に制限時間をかけます。

import asyncio, time

async def stew(sec):
    await asyncio.sleep(sec)
    return "stew"

async def main():
    start = time.perf_counter()
    try:
        async with asyncio.timeout(0.2):
            await stew(1.0)
    except TimeoutError:
        print(f"timeout after {time.perf_counter() - start:.2f}s")

    try:
        print(await asyncio.wait_for(stew(0.1), timeout=0.2))
        print(await asyncio.wait_for(stew(1.0), timeout=0.2))
    except TimeoutError:
        print("wait_for: timeout")

    print(asyncio.TimeoutError is TimeoutError)

asyncio.run(main())
# timeout after 0.20s
# stew
# wait_for: timeout
# True

時間を超えるとブロックの中の仕事がキャンセルされ、ブロックの外に組み込みの TimeoutError が送出されます。3.11からは asyncio.TimeoutError と組み込みの TimeoutError は同じものです。
asyncio.wait_for(コルーチン, timeout=) は、コルーチンひとつに制限をかける昔からある方法です。結果をすぐ受け取りたいときに便利です。
asyncio.timeout は複数の await をひとつのブロックにまとめられます。「注文を取ってから提供までまとめて20分」のように。

制限は「何秒後」ではなく「何時何分」でかけることもできますし、ブロックに入ったあとで変えることもできます。

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    start = time.perf_counter()
    deadline = loop.time() + 0.3            # いまから0.3秒後という「時刻」
    try:
        async with asyncio.timeout_at(deadline):
            await asyncio.sleep(1)
    except TimeoutError:
        print(f"timeout_at: {time.perf_counter() - start:.1f}s")

    start = time.perf_counter()
    try:
        async with asyncio.timeout(None) as cm:   # 最初は制限なし
            cm.reschedule(loop.time() + 0.2)      # あとから0.2秒に決める
            print(cm.when() is not None)
            await asyncio.sleep(1)
    except TimeoutError:
        print(f"reschedule: {time.perf_counter() - start:.1f}s, expired={cm.expired()}")

asyncio.run(main())
# timeout_at: 0.3s
# True
# reschedule: 0.2s, expired=True

timeout_at(時刻) は loop.time() 基準の絶対時刻を受け取ります。いくつもの段階が「締め切り時刻」ひとつを共有するときに便利です。
timeout(None) は制限なしで始まります。cm.reschedule(時刻) で、あとから制限を決めたり延ばしたりできます。

10-5. キャンセル：cancel、CancelledError、finally (6分)

TaskGroupも制限時間も、結局は同じことをしています。タスクにキャンセルを送ることです。自分で送るときは task.cancel() を使います。

import asyncio

async def braise():
    print("stove on")
    try:
        await asyncio.sleep(10)
    finally:
        print("stove off")          # キャンセルされても火は消す

async def main():
    task = asyncio.create_task(braise())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancelled:", task.cancelled())

asyncio.run(main())
# stove on
# stove off
# cancelled: True

cancel() はタスクをその場で殺すわけではありません。タスクが次に止まっている await の場所で CancelledError が送出されるようにするだけです。
火を消す、接続を閉じるといった後片づけは try/finally に置きます。キャンセルでも成功でも失敗でも、必ず実行されます。
CancelledError は Exception ではなく BaseException を継承しています。except Exception: がうっかりキャンセルを握りつぶさないようにするための設計です。

キャンセルを握りつぶすとどうなるでしょうか？

import asyncio, time

async def stubborn():
    try:
        await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        pass                          # キャンセルを握りつぶした
    await asyncio.sleep(0.5)          # そして料理を続ける
    return "late"

async def main():
    start = time.perf_counter()
    try:
        async with asyncio.timeout(0.1):
            print(await stubborn())
    except TimeoutError:
        print("timeout")
    print(f"{time.perf_counter() - start:.1f}s")

asyncio.run(main())
# late
# 0.6s

0.1秒の制限をかけたのに0.6秒かかり、TimeoutError も出ませんでした。制限時間はキャンセルを使って働くのに、そのキャンセルをコルーチンが食べてしまったからです。TaskGroupの「残りのキャンセル」もまったく同じように崩れます。だから CancelledError を受け取ったら、後片づけだけして送出し直します（raise）。

シェフのひとこと

キャンセルが来たら火を消して、まな板を拭いて、そのまま手を挙げる。「あと少しでできるのに」って煮続けた瞬間、厨房のルールが全部崩れるの。ミンホ、あんたのことだよ。

10-6. キャンセルを防ぐshield、キャンセルを数えるcancelling (2分)

レシートの保存のように、途中で切れてはいけない仕事もあります。asyncio.shield() は、外から来たキャンセルが内側のタスクまで届かないように防ぎます。

import asyncio

async def save_receipt():
    await asyncio.sleep(0.3)
    print("receipt saved")
    return "ok"

async def main():
    inner = asyncio.create_task(save_receipt())
    try:
        async with asyncio.timeout(0.1):
            await asyncio.shield(inner)
    except TimeoutError:
        print("waiter gave up")
    print(await inner)              # 内側のタスクは動き続けていた

asyncio.run(main())
# waiter gave up
# receipt saved
# ok

待っていた側は0.1秒であきらめましたが、内側のタスクは最後まで動きました。内側のタスクへの参照を持っておかないと、あとで結果を受け取れません。

Task.cancelling() はそのタスクに届いたキャンセル要求の数を、Task.uncancel() はその数をひとつ減らします。asyncio.timeout やTaskGroupが「このキャンセルは自分が送ったものか」を見分けるために内部で使う道具です。自分で使うことはめったにありません。

import asyncio

async def worker():
    try:
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        me = asyncio.current_task()
        print("cancelling:", me.cancelling())
        me.uncancel()
        print("after uncancel:", me.cancelling())
        return "rescued"

async def main():
    t = asyncio.create_task(worker())
    await asyncio.sleep(0.05)
    t.cancel()
    print(await t)

asyncio.run(main())
# cancelling: 1
# after uncancel: 0
# rescued

閉店後 (10分)

今夜の2つのルールを、ひとつの厨房にかけます。テーブルごとに0.6秒の制限をかけ、6卓をTaskGroupひとつにまとめます。

import asyncio, time

async def table(n, sec, allergy, stopped):
    try:
        async with asyncio.timeout(0.6):     # どの料理も0.6秒を超えない
            await asyncio.sleep(sec)
            if n == allergy:
                raise ValueError(f"T{n} allergy")
            return f"T{n}"
    except asyncio.CancelledError:
        stopped.append(f"T{n}")             # 火を消して手を挙げる
        raise

async def service(menu, allergy=None):
    start, stopped = time.perf_counter(), []
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(table(n, s, allergy, stopped)) for n, s in menu.items()]
        print("served:", [t.result() for t in tasks])
    except* ValueError as eg:
        print("cancelled by:", [str(e) for e in eg.exceptions], "stopped:", sorted(stopped))
    except* TimeoutError as eg:
        print("too slow:", len(eg.exceptions))
    print(f"{time.perf_counter() - start:.1f}s")

MENU = {1: 0.30, 2: 0.35, 3: 0.20, 4: 0.40, 5: 0.45, 6: 0.50}
asyncio.run(service(MENU))
asyncio.run(service(MENU, allergy=3))
asyncio.run(service({**MENU, 6: 0.9}))
# served: ['T1', 'T2', 'T3', 'T4', 'T5', 'T6']
# 0.5s
# cancelled by: ['T3 allergy'] stopped: ['T1', 'T2', 'T4', 'T5', 'T6']
# 0.2s
# too slow: 1
# 0.6s

ふだんの夜は6卓が一緒に煮えて、いちばん長いT6に合わせて約0.5秒で終わります。
3番テーブルが0.2秒で失敗すると、残りの5卓がその場で火を消します。
ひとつの料理が0.6秒を超えると、そのテーブルの TimeoutError が束になって上がってきて、やはり残りが止まります。

閉店間際、窓際の客が麺の器を空にして、小さな手帳を閉じます。会計を済ませながら、シェフに短く言います。「来週の金曜日に、また来ますね」

シェフはうなずいて、副料理長のほうを振り返ります。「キャンセルも制限時間も、結局はawaitの場所で起きる。でも厨房の隅の古いオーブンは、awaitを知らない。ルーがその前に立って待ったら、ホールはどうなると思う？」

営業日誌 — 十日目の夜。団体の6卓をTaskGroupひとつにまとめたので、3番テーブルが抜けると残りの鍋が同時に火を止めた。テーブルごとに制限時間をかけ、キャンセルを握りつぶさないルールを厨房の壁に貼った。

練習問題

タイムラインで、失敗するテーブルを1から6まで変えてみてください。失敗したテーブルより番号の小さいテーブルだけが生き残る理由を説明してみてください。次に失敗を「なし」にして、制限時間を0.3秒から0.55秒まで上げながら、最後まで行くテーブルが増えていく様子を見てください。
10-1の gather の例を TaskGroup に書き換えて、2番テーブルが失敗したら1番と3番が cancelled を表示するようにしてみてください。
閉店後のコードの table から raise を消し、その場所に await asyncio.sleep(0.3)（キャンセルを無視して煮続ける）を入れてから、allergy=3 で回してみてください。かかった時間がどう変わるかを見て、何がまずかったのかを書いてみてください。

シェフのレシピカード

gatherは1つが失敗しても残りを止めない。一緒に終わるべき仕事はTaskGroup（3.11）でまとめる。
async with asyncio.TaskGroup() as tg: tg.create_task(...) — ブロックを出たら全部終わっていて、1つが失敗すれば残りはキャンセル。
失敗はExceptionGroupにまとまって届く。except* ValueError で種類ごとに受け取る（returnは禁止、exceptとの混在も禁止）。
制限時間：async with asyncio.timeout(秒) または wait_for(コルーチン, timeout=)。超えたら組み込みの TimeoutError。
キャンセルはawaitの場所にCancelledErrorとして届く。finallyで後片づけし、受け取ったキャンセルはraiseし直す。
切れてはいけない仕事は asyncio.shield(task)。cancelling()/uncancel() は道具を作る側の仕掛け。

← 前へ

9限目 · 一人で回すホール係：asyncioの第一歩

11限目 · 止まったホール：ブロッキングと組み合わせ

新しい授業・講座のお知らせを受け取る

この講座に授業が追加されたり、続きの講座が公開されたりしたらメールでお知らせします。