評論家の夜：総まとめ

import asyncio, time

SLOW = {3, 7, 11}                      # 最初の1回は3倍かかる皿

async def plate(n, attempt):
    await asyncio.sleep(0.5 * (3 if n in SLOW and attempt == 0 else 1))

async def order(n, stoves, limit_s, retries):
    for attempt in range(retries + 1):
        async with stoves:             # コンロを確保しているあいだだけ料理する
            try:
                async with asyncio.timeout(limit_s):
                    await plate(n, attempt)
                return True
            except TimeoutError:
                pass                   # コンロを手放し、列の最後に並び直してもう一度
    return False

async def service(stoves_n, limit_s, retries):
    stoves = asyncio.Semaphore(stoves_n)
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(order(n, stoves, limit_s, retries)) for n in range(1, 13)]
    served = sum(t.result() for t in tasks)
    print(f"stoves={stoves_n} limit={limit_s} retries={retries}: "
          f"served {served}/12, {time.perf_counter() - start:.1f}s")

asyncio.run(service(3, None, 0))       # 制限時間なし
asyncio.run(service(3, 0.6, 0))        # 制限だけ
asyncio.run(service(3, 0.6, 1))        # 制限 + 再試行1回
# stoves=3 limit=None retries=0: served 12/12, 3.5s
# stoves=3 limit=0.6 retries=0: served 9/12, 2.2s
# stoves=3 limit=0.6 retries=1: served 12/12, 2.7s
# (かかる時間は実行のたびに0.1秒ほど変わる。1行目は3.5sか3.6s)

import asyncio, threading, time
from concurrent.futures import ProcessPoolExecutor

def knead(n):                          # 生地：手がふさがる仕事 → 支店の厨房（プロセス）
    return sum(i * i for i in range(n)) % 97

tally_lock = threading.Lock()          # オーブンのスレッドとループが一緒に書き換える帳簿
tally = {"plates": 0}

def count_plate():
    with tally_lock:                   # ロックの中にはawaitも長い仕事もない
        tally["plates"] += 1

def old_oven(item):                    # 古いオーブン：ブロッキング → スレッド
    time.sleep(0.4)
    count_plate()
    return f"{item} baked"

async def stove_dish(n, stoves):       # コンロ料理：セマフォ
    async with stoves:
        await asyncio.sleep(0.2)
    count_plate()
    return f"#{n}"

async def main():
    loop = asyncio.get_running_loop()
    stoves = asyncio.Semaphore(3)
    with ProcessPoolExecutor(max_workers=2) as dough_room:
        async with asyncio.timeout(5):         # コース全体の締め切り
            results = await asyncio.gather(
                loop.run_in_executor(dough_room, knead, 3_000_000),
                asyncio.to_thread(old_oven, "bread"),
                *(stove_dish(n, stoves) for n in range(1, 7)),
            )
    print(results)
    print(tally)

if __name__ == "__main__":
    asyncio.run(main())
# [56, 'bread baked', '#1', '#2', '#3', '#4', '#5', '#6']
# {'plates': 7}

import asyncio

running = set()                        # タスクをつかんでおく場所

async def wipe(table):
    await asyncio.sleep(0.05)
    print("wiped", table)

async def main():
    for t in ("T1", "T2"):
        task = asyncio.create_task(wipe(t))
        running.add(task)
        task.add_done_callback(running.discard)   # 終わったら自分で抜ける
    await asyncio.sleep(0.1)
    print(len(running))

asyncio.run(main())
# wiped T1
# wiped T2
# 0

import asyncio

async def cook(q):
    while True:
        await q.get()
        await asyncio.sleep(0.01)
        # q.task_done()  ← 忘れた

async def main():
    q = asyncio.Queue()
    for n in range(3):
        q.put_nowait(n)
    worker = asyncio.create_task(cook(q))
    try:
        async with asyncio.timeout(0.3):
            await q.join()
    except TimeoutError:
        print("join never finished:", q.qsize(), "in queue")
    worker.cancel()

asyncio.run(main())
# join never finished: 0 in queue

# 12-2のservice関数で
asyncio.run(service(3, 0.6, 1))
# stoves=3 limit=0.6 retries=1: served 12/12, 2.7s