ReflexKR

深夜厨房の並行処理

講座の概要

初日の行列：並行と並列

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

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

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

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

伝票差し：queue

出張料理人：concurrent.futures

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

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

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

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

評論家の夜：総まとめ

深夜厨房の並行処理

11限目

時間 8:20–9:10

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

今夜の注文書

#11 · 十一日目の夜

止まったホール

ピンチ

中古で仕入れた古いオーブンは、扉を閉めたら焼き上がるまで取っ手を放してくれない。ルーがその前に立っているあいだ、ホール全体が止まる。

目標

オーブンが焼いているあいだも、ほかのテーブルには0.3秒で料理を出す。

深夜食堂ドンシ

夜11時。3番テーブルがグラタンを注文します。グラタンは、シェフが中古で仕入れた古いオーブンでしか焼けません。このオーブンは扉を閉めると焼き上がるまで取っ手がロックされ、誰かが取っ手を握って立っていなければなりません。

ルーがオーブンの前に立ちます。その瞬間、ホールが止まります。水がほしい1番テーブルも、会計したい5番テーブルも、ルーが戻ってくるのをひたすら待つばかりです。

タクさんが舌打ちします。「オーブンの前に立ってるのは、仕事じゃない。ただ立ってるだけだ」シェフがうなずきます。「ルーは待っちゃだめ。待つのはほかの人に任せて」

11-1. ルーが止まればホールが止まる (7分)

asyncioのイベントループはひとつのスレッドで回っています。コルーチンが await で譲ってはじめて、ループはほかのタスクに移れます。ところが await を知らない関数、たとえば time.sleep や古い同期ライブラリを呼ぶと、ループはその関数が終わるまで身動きが取れません。こうした呼び出しをブロッキング呼び出しと呼びます。

import asyncio, time

start = time.perf_counter()

async def lu_rounds():                 # ルーは0.1秒ごとにホールを1周する
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven():                        # awaitを知らない古い関数
    time.sleep(0.35)
    return "bread"

async def bake():
    await asyncio.sleep(0.15)
    print("oven", old_oven())          # ここでループ全体が0.35秒止まる

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# oven bread
# round 0.5s
# round 0.6s
# round 0.7s

0.2、0.3、0.4秒の巡回がそっくり消えました。ルーがオーブンの前に立っていた0.15〜0.5秒のあいだ、どのタスクも動けなかったのです。
ブロッキング呼び出しは、そのコルーチンだけを遅らせるのではありません。ループ上のすべてのタスクをまとめて止めます。複数スレッドとのいちばん大きな違いです。
よくある犯人：time.sleep、requests.get のような同期のネットワークライブラリ、大きなファイルの読み込み、重い計算。

下のタイムラインでは、遅いテーブルがひとつ古いオーブン（time.sleep、0.6秒）を使います。ほかのテーブルは0.1秒ずつ区切って asyncio.sleep で待ちます。

厨房タイムライン

注文

遅いテーブル

to_threadに任せる

調理

待ち

キャンセル

失敗

テーブル6卓、3番が古いオーブンを使います。オレンジの棒（オーブン）が動いているあいだ、ほかのテーブルの棒が一緒に伸びたり、始まることすらできなかったりします。「to_threadに任せる」をオンにすると、全部がまた重なります。

スイッチをオフのまま回すと約0.9秒かかります。T1とT2は0.3秒の料理なのに約0.76秒まで伸び、T4〜T6はオーブンが終わった0.6秒になってようやく始まります。ルーがそのテーブルにまだ行けてもいなかったのです。
遅いテーブルを1番に変えると、残りの5卓がすべて0.6秒後にようやく始まります。
スイッチをオンにするとオーブンがスレッドに移り、ほかのテーブルは約0.3秒で終わって、全体もオーブンの時間である約0.6秒に縮みます。

11-2. デバッグモードで犯人を探す (6分)

大きなプログラムでは、どのコルーチンがループをつかんで放さなかったのかを知るのは難しいものです。asyncioのデバッグモードは、一度に長く実行されたコールバックやタスクをログで知らせてくれます。

import asyncio, logging, time

logging.basicConfig(level=logging.WARNING, format="%(name)s %(levelname)s: %(message)s")

def old_oven():
    time.sleep(0.3)

async def bake():
    old_oven()

async def main():
    await asyncio.create_task(bake(), name="T3-oven")

asyncio.run(main(), debug=True)
# 例：asyncio WARNING: Executing <Task finished name='T3-oven' coro=<bake() done, defined at ...oven.py:8>
#     result=None created at ...\asyncio\tasks.py:395> took 0.301 seconds

asyncio.run(main(), debug=True) でオンにします。環境変数 PYTHONASYNCIODEBUG=1 や python -X dev でもオンにできます。
警告1行に、タスク名（T3-oven）、コルーチンが定義されたファイルと行、つかんでいた時間（took 0.301 seconds）が出ます。タスクに name= を付けておくと犯人を見つけやすくなります。
メッセージは asyncio ロガーのWARNINGとして出ます。パスは実行するPCごとに違います。

基準の時間は loop.slow_callback_duration で、デフォルトは0.1秒です。もっと敏感に捕まえたいなら減らします。

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    print(loop.get_debug(), loop.slow_callback_duration)
    loop.slow_callback_duration = 0.05     # 0.05秒を超えただけでも知らせて
    time.sleep(0.07)

asyncio.run(main(), debug=True)
# True 0.1
# 例：Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 0.070 seconds

デバッグモードはほかにも、await されずに捨てられたコルーチンがどこで作られたかを教えてくれたり、別のスレッドからループをまちがって触るとエラーを出したりします。開発中はオンにしておき、本番ではオフにします（遅くなるので）。

11-3. 待つのはスレッドに：to_threadとrun_in_executor (8分)

古いオーブンを直せないなら、オーブンの前に立つ人を別に置きます。asyncio.to_thread()（3.9）はブロッキング関数を別のスレッドで動かし、ループには待てるオブジェクトを返します。

import asyncio, time

start = time.perf_counter()

async def lu_rounds():
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven(item, *, minutes):
    time.sleep(0.35)
    return f"{item} {minutes}m"

async def bake():
    await asyncio.sleep(0.15)
    bread = await asyncio.to_thread(old_oven, "bread", minutes=40)
    print("oven", bread)

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# round 0.2s
# round 0.3s
# round 0.4s
# oven bread 40m

ルーは0.1秒ごとに欠かさずホールを回ります。オーブンはスレッドが見張り、ルーは結果を await で受け取るだけです。
to_thread(関数, *args, **kwargs) は引数をそのまま渡します。7限目のスレッドプール（ループのデフォルトexecutor）で動きます。
スレッドなので、GILのルール（8限目）はそのままです。待つブロッキングコードにはよく効きますが、計算の重い仕事は速くなりません。

計算の重い仕事は支店の厨房、つまりプロセスプールに送ります。asyncio.get_running_loop() でいま動いているループを取得し、loop.run_in_executor(executor, 関数, *args) を使います。プロセスを使うので、ファイルに保存して実行します。

import asyncio
from concurrent.futures import ProcessPoolExecutor

def knead(n):                          # 手がふさがる仕事（CPU）
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        doughs = await asyncio.gather(
            loop.run_in_executor(pool, knead, 2_000_000),
            loop.run_in_executor(pool, knead, 3_000_000),
        )
    print(doughs)

if __name__ == "__main__":
    asyncio.run(main())
# [2666664666667000000, 8999995500000500000]

run_in_executor の最初の引数に None を渡すと、デフォルトのスレッドプールを使います。to_thread は、まさにそれを手軽にしてくれる関数です。
run_in_executor はキーワード引数を受け取りません。必要なら functools.partial(knead, n=...) で包んで渡します。
プロセスプールなので、8限目のルールもそのままです。if __name__ == "__main__": が必要で、渡す関数と値はpickleできなければなりません。

シェフのひとこと

async defの中で、名前にasyncもawaitも付いてない関数を呼ぶときは、一度立ち止まって聞いて。「これ、時間かかる？」 0.1秒でもかかるならto_thread。ルーをオーブンの前に立たせないで。

11-4. asyncioの同期の道具 (8分)

4限目と6限目の道具には、asyncio版が別にあります。名前は同じですが、待つときにスレッドを止める代わりにawait で譲ります。

threading	asyncio	使い方
`threading.Lock`	`asyncio.Lock`	`async with lock:`
`threading.Semaphore`	`asyncio.Semaphore`	`async with sem:`
`threading.Event`	`asyncio.Event`	`await event.wait()`
`queue.Queue`	`asyncio.Queue`	`await q.put(x)`, `await q.get()`

セマフォは「同時接続数の制限」にいちばんよく使います。仕入れ業者に電話を5本かけなければならないのに、業者は同時に2本までしか受けられません。

import asyncio, time

start = time.perf_counter()
supplier = asyncio.Semaphore(2)        # 仕入れ業者への電話は同時に2本まで

async def call(n):
    async with supplier:
        print(f"call #{n} at {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.2)

async def main():
    await asyncio.gather(*(call(n) for n in range(1, 6)))

asyncio.run(main())
# call #1 at 0.0s
# call #2 at 0.0s
# call #3 at 0.2s
# call #4 at 0.2s
# call #5 at 0.4s

asyncio.Queue は6限目の伝票差しと同じ形です。maxsize、task_done()、join() もそのままあります。

import asyncio

async def hall(q):
    for n in range(1, 6):
        await q.put(f"#{n}")           # 伝票差しがいっぱいならここで待つ
    print("hall done")

async def cook(name, q, done):
    while True:
        ticket = await q.get()
        await asyncio.sleep(0.1)
        done.append((name, ticket))
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=2)
    done = []
    cooks = [asyncio.create_task(cook(c, q, done)) for c in ("A", "B")]
    await hall(q)
    await q.join()                     # 差した注文がすべてtask_doneされるまで
    for c in cooks:
        c.cancel()
    print(len(done), sorted(t for _, t in done))

asyncio.run(main())
# hall done
# 5 ['#1', '#2', '#3', '#4', '#5']

ループはひとつのスレッドなので競合状態はなさそうに思えますが、読み込みと書き込みのあいだに await があると、3限目の黒板の事故がそのまま起きます。

import asyncio

async def main():
    board = {"count": 0}

    async def cook():
        for _ in range(100):
            v = board["count"]
            await asyncio.sleep(0)     # ここでほかのタスクに移る
            board["count"] = v + 1

    await asyncio.gather(*(cook() for _ in range(4)))
    print(board["count"])

asyncio.run(main())
# 100

400になるはずの黒板が100です。4つのタスクが同じ値を読み、譲り、同じ値を書くのです。asyncio.Lock でその区間をくくり、asyncio.Event で「オープン！」の合図を待たせると、こうなります。

import asyncio

async def main():
    lock = asyncio.Lock()
    opened = asyncio.Event()
    board = {"count": 0}

    async def cook():
        await opened.wait()            # 「オープン！」まで待つ
        for _ in range(100):
            async with lock:           # 読み込みと書き込みのあいだにawaitがあるならロックが必要
                v = board["count"]
                await asyncio.sleep(0)
                board["count"] = v + 1

    tasks = [asyncio.create_task(cook()) for _ in range(4)]
    await asyncio.sleep(0.1)
    opened.set()
    await asyncio.gather(*tasks)
    print(board["count"])

asyncio.run(main())
# 400

2つの世界の道具を混ぜてはいけません。コルーチンで threading.Lock を待つと、ループがまるごと止まります。ロックを握ったタスクが戻ってきて解放する機会も失われます。

import asyncio, threading, time

lock = threading.Lock()                # スレッド用のロックをコルーチンで使うと？
start = time.perf_counter()

async def cook(n):
    if not lock.acquire(timeout=0.2):  # 待っているあいだ、ループ全体が止まる
        print(f"#{n} gave up at {time.perf_counter() - start:.1f}s")
        return
    try:
        await asyncio.sleep(0.1)       # ロックしたまま譲る
        print(f"#{n} done")
    finally:
        lock.release()

async def main():
    await asyncio.gather(cook(1), cook(2))

asyncio.run(main())
# #2 gave up at 0.2s
# #1 done

timeout がなければ、#2は永遠に待ち続け、#1は二度と目を覚ましません。逆に、asyncioの道具はスレッドセーフではありません。別のスレッドから asyncio.Queue に入れてはいけません。スレッドからループへ仕事を渡す方法は11-6で見ます。

11-5. async with、async for、非同期ジェネレーター (7分)

with と for にも、待つ版があります。

async with：入るときと出るときに await が必要な資源（__aenter__、__aexit__）。ロック、セマフォ、ネットワーク接続。
async for：次の値を受け取るのに待つ必要がある繰り返し（__aiter__、__anext__）。ひとつずつ届く伝票、ストリーミングのレスポンス。
非同期ジェネレーター：async def の中で yield を使うと、async for で回すイテレーターになります。

import asyncio
from contextlib import asynccontextmanager

async def tickets(n):                  # 非同期ジェネレーター：async def + yield
    for i in range(1, n + 1):
        await asyncio.sleep(0.05)      # 次の伝票が来るまで待つ
        yield f"#{i}"

@asynccontextmanager
async def stove(name):                 # async with で使うコンロ
    await asyncio.sleep(0.01)
    print(f"{name} on")
    try:
        yield name
    finally:
        print(f"{name} off")

async def main():
    async with stove("S1"):
        async for t in tickets(3):
            print("cook", t)
    print([t async for t in tickets(2)])

asyncio.run(main())
# S1 on
# cook #1
# cook #2
# cook #3
# S1 off
# ['#1', '#2']

contextlib.asynccontextmanager は、yield ひとつで async with 用の資源を作ってくれます。リスト内包表記でも async for が使えます（コルーチンの中だけ）。

7限目の as_completed にもasyncio版があります。3.13からは async for で回せるようになり、このとき最初に入れたタスクそのものを終わった順に返してくれます。

import asyncio

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

async def main():
    tasks = [asyncio.create_task(dish(n, s)) for n, s in [(1, 0.3), (2, 0.1), (3, 0.2)]]
    async for t in asyncio.as_completed(tasks):      # 3.13から
        print(t in tasks, await t)

asyncio.run(main())
# True #2
# True #3
# True #1

以前のように for f in asyncio.as_completed(tasks): await f と書いてもかまいませんが、そのとき受け取る f は元のタスクではなく新しいオブジェクトです。どの注文が終わったかを知りたいなら、async for のほうが便利です。

11-6. ループの外から仕事を投げる、動いているタスクを見る (4分)

予約の電話を受ける係が別のスレッドで働いているとしましょう。そのスレッドがルーの予定に仕事を割り込ませるときは、asyncio.run_coroutine_threadsafe(コルーチン, loop) を使います。返ってくるのは7限目の concurrent.futures.Future なので、スレッド側で result(timeout=) を使って待てます。

import asyncio

async def seat(table):
    await asyncio.sleep(0.05)
    return f"{table} seated"

def phone_line(loop):                  # ループの外のスレッド（予約の電話）
    fut = asyncio.run_coroutine_threadsafe(seat("T7"), loop)
    print(fut.result(timeout=1))       # concurrent.futures.Future

async def main():
    loop = asyncio.get_running_loop()
    await asyncio.to_thread(phone_line, loop)

asyncio.run(main())
# T7 seated

eagerタスク（3.12）：ふつう create_task はタスクを予定に載せるだけで、次の順番が来るまで実行しません。asyncio.eager_task_factory を使うと、作った瞬間に最初の await まで実行します。await なしですぐ終わるコルーチンなら、予定に載ることすらなくその場で終わります。

import asyncio

async def greet(n):
    print(f"greet T{n}")               # 最初のawaitの前まで
    await asyncio.sleep(0)
    return n

async def main(eager):
    if eager:
        asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
    t = asyncio.create_task(greet(1))
    print("created, done =", t.done())
    await t

asyncio.run(main(False))
asyncio.run(main(True))
# created, done = False
# greet T1
# greet T1
# created, done = False

動いているタスクを見る（3.14）：実行中のasyncioプログラムのプロセス番号（PID）がわかれば、別のターミナルから python -m asyncio ps <PID> でタスクの一覧を、python -m asyncio pstree <PID> で誰が誰を待っているかを木の形で見られます。たとえば hall タスクがTaskGroupで T1、T2 を待っている最中なら、こう出ます（一部の列とパスは省略）。

task name   coroutine stack                                   awaiter name
Task-1      main
hall        TaskGroup._aexit -> TaskGroup.__aexit__ -> hall   Task-1
T1          sleep -> table                                    hall
T2          sleep -> table                                    hall

「ホールが止まったのに、ルーがどこに立っているかわからない」という瞬間に使う道具です。タスクに name= を付けておくと、一覧がずっと読みやすくなります。

閉店後 (10分)

3番テーブルのグラタンだけを古いオーブンに入れ、オーブンの前にはスレッドを立たせます。スイッチひとつ（fix）で比べます。

import asyncio, time

def old_oven(sec):                     # 手を入れられない古いライブラリ
    time.sleep(sec)
    return "gratin"

async def table(n, start, fix):
    if n == 3:
        dish = await asyncio.to_thread(old_oven, 0.6) if fix else old_oven(0.6)
    else:
        await asyncio.sleep(0.3)
        dish = "noodle"
    return f"T{n} {dish} {time.perf_counter() - start:.1f}s"

async def service(fix):
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(table(n, start, fix)) for n in range(1, 7)]
    print([t.result() for t in tasks])

asyncio.run(service(fix=False))
asyncio.run(service(fix=True))
# ['T1 noodle 0.6s', 'T2 noodle 0.6s', 'T3 gratin 0.6s', 'T4 noodle 0.9s', 'T5 noodle 0.9s', 'T6 noodle 0.9s']
# ['T1 noodle 0.3s', 'T2 noodle 0.3s', 'T3 gratin 0.6s', 'T4 noodle 0.3s', 'T5 noodle 0.3s', 'T6 noodle 0.3s']

ルーが自分でオーブンの前に立つと、0.3秒の麺がひとつも0.3秒で出せません。T1とT2はオーブンより前に鍋を火にかけていたのに0.6秒になってやっと提供され、T4〜T6はオーブンが終わってから鍋を火にかけるので0.9秒に出ます。タイムラインで見た形そのままです。
to_thread に任せると、麺は0.3秒、グラタンはオーブンの時間である0.6秒に出ます。これが今夜の目標です。

閉店後、シェフが厨房の壁のレシピカードを1枚ずつ指でたどります。スレッド、ロック、キュー、プール、プロセス、ループ、TaskGroup、to_thread。「道具がこれだけたまった。でもお客さんが扉を開けて入ってきたとき、どの道具を先に出すかはどう決める？」

営業日誌 — 十一日目の夜。古いオーブンを待つせいでホールが0.6秒ずつ止まる問題を、デバッグモードで突き止めた。オーブンはto_threadでスレッドに任せ、仕入れ業者への電話はasyncio.Semaphoreで2本までに絞った。

練習問題

タイムラインで、遅いテーブルを1から6まで変えてみてください。遅いテーブルより前の番号と後ろの番号で棒の形がどう違うか、なぜそうなるのかを説明してみてください。
11-1の最初の例を asyncio.run(main(), debug=True) で回し、警告メッセージに、どのタスクが何秒つかんでいたと出るかを確かめてみてください。
11-4のセマフォの例を asyncio.Semaphore(3) に変えると、call #5 が何秒に始まるか、先に計算してから回してみてください。

シェフのレシピカード

ブロッキング呼び出し（time.sleep、同期ライブラリ）は、ループ上のすべてのタスクを止める。
犯人探し：asyncio.run(main(), debug=True)、loop.slow_callback_duration（デフォルト0.1秒）。
待つブロッキングコードは await asyncio.to_thread(f, *args)。重い計算は loop.run_in_executor(ProcessPoolExecutor(), f, x)。
asyncio版の道具：Lock、Semaphore、Event、Queue — async with / await で使う。threading用と混ぜない。
async with、async for、非同期ジェネレーター（async def + yield）。3.13から async for t in asyncio.as_completed(tasks)。
スレッドからループへ：run_coroutine_threadsafe(coro, loop)。のぞき見：python -m asyncio ps/pstree <PID>（3.14）。

← 前へ

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

12限目 · 評論家の夜：総まとめ

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

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