Async / Await และ Tasks

Sequential เทียบกับ concurrent await

จุดสำคัญในการเขียนโปรแกรม async คือความแตกต่างระหว่างการ await coroutines แบบ sequential กับการ schedule ให้รันแบบ concurrent

การ await ทีละตัวคือ sequential — แต่ละ coroutine ต้องเสร็จก่อนตัวถัดไปจะเริ่ม:

import asyncio

async def fetch(label: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{label} result"

async def sequential() -> None:
    a = await fetch("A", 0.3)   # waits 0.3 s
    b = await fetch("B", 0.2)   # waits 0.2 s after A finishes
    print(a, b)
    # Total: ~0.5 s

asyncio.run(sequential())

การใช้ asyncio.gather รันแบบ concurrent — เวลารวมเท่ากับตัวที่ช้าที่สุด ไม่ใช่ผลรวม:

import asyncio

async def fetch(label: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{label} result"

async def concurrent() -> None:
    a, b = await asyncio.gather(
        fetch("A", 0.3),
        fetch("B", 0.2),
    )
    print(a, b)
    # Total: ~0.3 s

asyncio.run(concurrent())

`asyncio.create_task` — schedule task อย่างชัดเจน

asyncio.create_task(coro) ห่อ coroutine ไว้ใน Task และ schedule บน event loop ที่กำลังรันทันที ต่างจากการเรียก coroutine ตรงๆ task จะเริ่มรันทันทีที่ coroutine ปัจจุบันยกการควบคุม

import asyncio

async def worker(name: str, delay: float) -> str:
    print(f"{name}: starting")
    await asyncio.sleep(delay)
    print(f"{name}: done")
    return f"{name}-result"

async def main() -> None:
    # Schedule both tasks immediately — they run concurrently
    task_a = asyncio.create_task(worker("A", 0.3))
    task_b = asyncio.create_task(worker("B", 0.1))

    # Await them to collect results
    result_a = await task_a
    result_b = await task_b
    print(result_a, result_b)

asyncio.run(main())
# A: starting
# B: starting
# B: done
# A: done
# A-result B-result

สังเกตว่า “B: done” ปรากฏก่อน “A: done” แม้เราจะ await task_a ก่อนก็ตาม — tasks รันแบบ concurrent และ B เสร็จเร็วกว่า

กับดัก fire-and-forget: เก็บ reference ของ task ไว้เสมอ

การเรียก asyncio.create_task(coro) โดยไม่เก็บผลลัพธ์ไว้มัก compile และรันได้ในตัวอย่างสั้นๆ — ซึ่งเป็นสิ่งที่ทำให้มันอันตราย เอกสารทางการพูดถึงความเสี่ยงนี้ตรงๆ: “the event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done.” ไม่มีอะไรในภาษาบังคับให้คุณเก็บ reference ดังนั้นไม่มีอะไรเตือนตอนที่คุณลืม:

import asyncio

async def background_job(label: str) -> None:
    await asyncio.sleep(0.05)
    print(f"{label}: finished")

async def fire_and_forget() -> None:
    asyncio.create_task(background_job("leaked"))   # no variable holds the Task
    await asyncio.sleep(0.1)   # only works because the loop's own scheduling keeps it alive here

การแก้ไม่มีต้นทุนและไม่มีข้อเสีย — เก็บ task ไว้ใน set แล้วให้ done-callback ทำความสะอาด set เมื่อเสร็จ:

import asyncio

background_tasks: set[asyncio.Task] = set()

async def background_job(label: str) -> None:
    await asyncio.sleep(0.05)
    print(f"{label}: finished")

def spawn(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)
    return task

หรือถ้า task แบบ fire-and-forget มี lifetime ร่วมกับ block ของโค้ด ให้ใช้ TaskGroup แทนการจดบันทึกเอง — มันทำให้ “รอทุกอย่างที่ฉันเริ่ม” เป็นค่า default แทนที่จะเป็นสิ่งที่ต้องจำเอง

`TaskGroup` cancellation เทียบกับ `gather(return_exceptions=True)`

ทั้ง TaskGroup และ gather(return_exceptions=True) ให้คุณรัน coroutine หลายตัวและเห็นทุกผลลัพธ์แม้บางตัว fail — แต่ทั้งสองไม่เห็นตรงกันว่าเกิดอะไรกับ ผู้รอดชีวิต:

import asyncio

async def worker(label: str, delay: float, fail: bool) -> str:
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"{label} failed")
    return f"{label} ok"

async def main() -> None:
    results = await asyncio.gather(
        worker("A", 0.05, False),
        worker("B", 0.02, True),
        worker("C", 0.08, False),
        return_exceptions=True,
    )
    print(results)
    # ['A ok', ValueError('B failed'), 'C ok'] — C still ran to completion

asyncio.run(main())

import asyncio

async def worker(label: str, delay: float, fail: bool) -> str:
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"{label} failed")
    return f"{label} ok"

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(worker("A", 0.05, False))
            tg.create_task(worker("B", 0.02, True))
            tg.create_task(worker("C", 0.08, False))
    except* ValueError as eg:
        print([str(e) for e in eg.exceptions])
        # ['B failed'] — B's failure cancelled C before its 0.08s sleep finished

asyncio.run(main())

gather(return_exceptions=True) ไม่ยกเลิกอะไรเลย — ทุก coroutine ที่ส่งเข้าไปรันจนจบไม่ว่าตัวอื่นจะเป็นอย่างไร แล้วคุณแยก success กับ exception เอง TaskGroup มองว่าความล้มเหลวหนึ่งครั้งเป็นเหตุให้ยกเลิกทั้งกลุ่ม: เป็นค่า default ที่ถูกต้องเมื่อ task ทำงานไปสู่เป้าหมายเดียวกัน (request fan-out ที่ผลลัพธ์บางส่วนไม่มีประโยชน์) และผิดเมื่อคุณต้องการทุกผลลัพธ์แบบอิสระจากความล้มเหลวของตัวอื่นจริงๆ (batch job ที่รายงานสถานะทีละรายการ)

Timeouts: `asyncio.timeout()` และ `wait_for`

asyncio.timeout() (3.11+) เป็น context manager: ทุกอย่างข้างในถูกยกเลิกถ้า block ไม่เสร็จภายในเวลาที่กำหนด แล้ว raise TimeoutError

import asyncio

async def slow_op() -> str:
    await asyncio.sleep(5)
    return "done"

async def main() -> None:
    try:
        async with asyncio.timeout(0.1):
            await slow_op()
    except TimeoutError:
        print("operation timed out")

asyncio.run(main())
# operation timed out

asyncio.wait_for(coro, timeout) คือรูปแบบเก่ากว่าที่ครอบ awaitable ตัวเดียว — ต่างจาก timeout() ที่ครอบ block ได้ ซึ่งสำคัญถ้าต้องการจำกัดเวลาหลาย statement พร้อมกัน (มีแค่ timeout() เท่านั้นที่ครอบมากกว่าหนึ่งบรรทัดได้) ทั้งสองยกเลิกแบบเดียวกันและ raise TimeoutError เหมือนกัน (ตั้งแต่ 3.11 asyncio.TimeoutError เป็น alias ของ builtin) asyncio.shield(coro) คือเครื่องมือตรงข้าม: มันปกป้อง awaitable จากการถูกยกเลิก โดยผู้เรียก — มีประโยชน์เมื่อขั้นตอน cleanup ต้องเสร็จแม้ operation รอบนอกจะถูกยกเลิก — แต่มันไม่ป้องกัน timeout()/wait_for ที่ครอบตัว shield เองจาก timeout; coroutine ข้างใน shield ยังถูกยกเลิกเองได้อิสระ

Blocking call ทำให้ loop แข็ง — `time.sleep` เทียบกับ `asyncio.sleep`

time.sleep() block ทั้ง OS thread ที่ event loop รันอยู่ เมื่อเรียกจากภายใน coroutine มันไม่ได้ “หยุด task นั้น” แบบที่ asyncio.sleep() ทำ — มันหยุด ทุกอย่าง เพราะมีแค่เธรดเดียวที่รัน loop:

import asyncio

async def heartbeat(n: int) -> None:
    for i in range(n):
        await asyncio.sleep(0.03)
        print(f"heartbeat {i}")

def blocking_sleep(seconds: float) -> str:
    import time
    time.sleep(seconds)      # blocks the WHOLE event loop thread
    return "worker done"

async def main() -> None:
    hb = asyncio.create_task(heartbeat(2))
    result = await asyncio.to_thread(blocking_sleep, 0.08)  # runs off the event loop's thread
    print(result)
    await hb

asyncio.run(main())
# heartbeat 0
# heartbeat 1
# worker done

asyncio.to_thread(fn, *args) (3.9+) คือทางแก้: มันรัน call ที่ block จริงๆ — time.sleep, synchronous DB driver, file I/O เบาๆ — ใน thread pool ทำให้ event loop ยังให้บริการ coroutine อื่นได้ระหว่างรอ มันเป็น wrapper บาง ๆ ของ loop.run_in_executor(None, functools.partial(fn, *args)) ซึ่งอยู่ในบท GIL & Threads

`Semaphore` — จำกัด concurrency

asyncio.Semaphore(n) ให้ coroutine ถือมันพร้อมกันได้สูงสุด n ตัว — วิธีมาตรฐานในการจำกัด request ที่ concurrent ไปยัง API ที่มี rate limit หรือ connection pool:

import asyncio

async def limited_call(sem: asyncio.Semaphore, label: str, active: list) -> None:
    async with sem:
        active.append(label)
        print(f"{label}: running, concurrent={len(active)}")
        await asyncio.sleep(0.05)
        active.remove(label)

async def main() -> None:
    sem = asyncio.Semaphore(2)
    active: list = []
    await asyncio.gather(*[limited_call(sem, f"job-{i}", active) for i in range(4)])

asyncio.run(main())
# never more than 2 "running" lines are concurrent

`Queue` — producer/consumer

asyncio.Queue คือเทียบเท่าแบบ async-native ของ queue.Queue: put/get เป็น coroutines ดังนั้น queue ที่เต็มจะหยุด producer (แทนที่จะ block thread) จนกว่า consumer จะดึงออกไป

import asyncio

async def producer(q: asyncio.Queue, n: int) -> None:
    for i in range(n):
        await q.put(i)
    await q.put(None)   # sentinel signals "no more items"

async def consumer(q: asyncio.Queue, out: list) -> None:
    while True:
        item = await q.get()
        if item is None:
            break
        out.append(item)

async def main() -> None:
    q: asyncio.Queue = asyncio.Queue(maxsize=2)   # bounded — producer waits if it fills up
    out: list = []
    await asyncio.gather(producer(q, 4), consumer(q, out))
    print(out)   # [0, 1, 2, 3]

asyncio.run(main())

`Lock` — check-then-act ยังเป็น race ข้าม `await`

Event loop เธรดเดียวหมายความว่าไม่มี coroutine สองตัวรัน bytecode ในเสี้ยววินาทีเดียวกัน — แต่นั่น ไม่ได้ แปลว่า shared state ปลอดภัย await ใดๆ ระหว่าง check กับ action ที่มัน guard คือช่องว่างที่ coroutine อื่นวิ่งเข้ามาเปลี่ยน state ก่อนได้:

import asyncio

class Account:
    def __init__(self, balance: int) -> None:
        self.balance = balance

async def withdraw_unsafe(acct: Account, amount: int) -> None:
    if acct.balance >= amount:   # check
        await asyncio.sleep(0)   # <-- another coroutine can run right here
        acct.balance -= amount   # act -- based on a now-stale read

async def main() -> None:
    acct = Account(balance=100)
    await asyncio.gather(withdraw_unsafe(acct, 80), withdraw_unsafe(acct, 80))
    print(acct.balance)   # -60 -- both checks passed before either write happened

asyncio.run(main())

asyncio.Lock ปิดช่องว่างนี้โดยทำให้ check-and-act เป็นหน่วยเดียวที่ทำต่อเนื่อง:

import asyncio

class Account:
    def __init__(self, balance: int) -> None:
        self.balance = balance
        self.lock = asyncio.Lock()

async def withdraw_safe(acct: Account, amount: int) -> None:
    async with acct.lock:
        if acct.balance >= amount:
            await asyncio.sleep(0)
            acct.balance -= amount

async def main() -> None:
    acct = Account(balance=100)
    await asyncio.gather(withdraw_safe(acct, 80), withdraw_safe(acct, 80))
    print(acct.balance)   # 20 -- the second withdrawal correctly sees the updated balance

asyncio.run(main())

นี่คือความจริงในโลก async ของสิ่งที่จริงกับ thread เช่นกัน: ไม่มี data race (undefined behavior) ไม่ได้แปลว่าไม่มี race condition Event loop เธรดเดียวรับประกันว่าคุณจะไม่ทำ memory เสียหายจาก coroutine สองตัวเขียนพร้อมกัน — แต่มันไม่รับประกันความถูกต้องเชิงตรรกะของลำดับ check-then-act ที่คร่อม await เลย asyncio.Event คือ primitive คู่กันในทิศตรงข้าม — ไม่ใช่ “ทีละตัว” แต่เป็น “ทุกคนรอจนกว่าฉันจะบอกให้ไป”: await event.wait() หยุดจนกว่า coroutine อื่นจะเรียก event.set()

Pattern ที่ใช้จริง: retry, rate limiting และ graceful shutdown

Retry แบบ exponential backoff และ jitter — การ retry ทันทีหลัง transient failure มักแค่ทำให้ failure เกิดซ้ำ (หรือไปโดน server ที่ overload อยู่แล้ว); backoff เว้นระยะ retry ออกจากกัน และ jitter (สุ่มระยะเวลา) ป้องกันไม่ให้ client หลายตัว retry พร้อมกันเป็นจังหวะเดียวกัน:

import asyncio
import random

class FlakyError(Exception):
    pass

async def retry_with_backoff(coro_fn, *, max_attempts: int = 5, base: float = 0.01) -> object:
    for attempt in range(1, max_attempts + 1):
        try:
            return await coro_fn()
        except FlakyError:
            if attempt == max_attempts:
                raise
            delay = base * (2 ** (attempt - 1)) + random.uniform(0, base)
            await asyncio.sleep(delay)

Rate limiting — token bucket จำกัดว่า action หนึ่งเกิดขึ้นได้บ่อยแค่ไหน ไม่ขึ้นกับจำนวนผู้เรียกที่รออยู่:

import asyncio

class RateLimiter:
    def __init__(self, rate: int, period: float) -> None:
        self.rate = rate
        self.period = period
        self.tokens = float(rate)

    async def acquire(self) -> None:
        while self.tokens < 1:
            await asyncio.sleep(self.period / self.rate)
            self.tokens += 1
        self.tokens -= 1

Graceful shutdown — service ที่รันยาวควรหยุดรับงานใหม่และปล่อยให้งานที่กำลังทำอยู่เสร็จ (หรือยกเลิกอย่างตั้งใจ) เมื่อถูกขอให้หยุด แทนที่จะถูก kill กลาง request รูปแบบทั่วไปคือ shutdown asyncio.Event ที่ set จาก signal handler โดย worker loop คอยเช็ค:

import asyncio
import signal

async def worker(shutdown: asyncio.Event) -> None:
    while not shutdown.is_set():
        await asyncio.sleep(0.1)   # do one unit of work per iteration in a real service
    print("worker exiting cleanly")

async def main() -> None:
    shutdown = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGTERM, shutdown.set)   # Unix only; not testable on Pyodide
    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker(shutdown))
        # ... other tasks ...

loop.add_signal_handler ใช้ได้เฉพาะ Unix และต้องมี OS signal จริงเพื่อสาธิต จึงไม่รวมอยู่ใน demo ที่รันได้ด้านล่าง — pattern (Event, worker คอยเช็ค, TaskGroup เพื่อให้ shutdown รอทุก worker) เหมือนกันไม่ว่าตัวกระตุ้นจะเป็น signal หรืออย่างใน demo คือ coroutine อื่นเรียก shutdown.set()

Structured concurrency

แนวคิดหลักของ TaskGroup — ทุก task ที่คุณ spawn มีเจ้าของชัดเจน และเจ้าของนั้นออกไม่ได้จนกว่า child ทั้งหมดจะเสร็จหรือถูกยกเลิก — เรียกว่า structured concurrency มันตัด bug ทั้งกลุ่มที่โค้ดแบบ create_task-แล้วลืมชักชวนให้เกิด: task ที่มีอายุยืนกว่า scope ที่ควรเป็นเจ้าของมัน, exception ใน child ที่ไม่มีใครสังเกตเห็น, โปรแกรมออกไปทั้งที่งาน background ยังทำอยู่ gather และ create_task เปล่าๆ ไม่บังคับสิ่งนี้ แต่ TaskGroup บังคับ

นอกเหนือจาก standard library

asyncio เป็น async framework ในตัวเดียวของ CPython แต่ third-party library สองตัวควรรู้จักไว้แม้จะรันใน Pyodide sandbox ของคอร์สนี้ไม่ได้: anyio ให้ API เดียวที่รันบน event loop ของ asyncio หรือ trio ก็ได้ มีประโยชน์สำหรับโค้ด library ที่ไม่ควรบังคับตัวเลือกให้ผู้เรียก; trio เป็นผู้บุกเบิกโมเดล structured-concurrency ที่ TaskGroup เอามาใช้ใน asyncio เองในภายหลัง สำหรับ HTTP โดยเฉพาะ httpx.AsyncClient คือคู่แบบ async-native ของ requests — การเรียก requests.get() แบบ synchronous ภายใน coroutine จะ block loop เหมือน time.sleep ด้านบนทุกประการ

Nested `asyncio.run` raise `RuntimeError`

การเรียก asyncio.run สองครั้งติดกันที่ระดับบนสุดทำได้ปกติ (ดูบท asyncio พื้นฐาน) แต่การเรียก จากภายใน coroutine ที่กำลังรันบน loop อยู่แล้วไม่ได้ — มี loop รันอยู่ในเธรดนี้แล้ว และ asyncio.run ปฏิเสธที่จะเริ่มตัวที่สอง:

import asyncio

async def inner() -> None:
    print("inner")

async def main() -> None:
    asyncio.run(inner())   # calling asyncio.run() from inside a running event loop

try:
    asyncio.run(main())
except RuntimeError as e:
    print(f"RuntimeError: {e}")
    # RuntimeError: asyncio.run() cannot be called from a running event loop

นี่คือความประหลาดใจแบบ Jupyter-notebook คลาสสิกด้วย: cell ของ notebook รันอยู่ใน event loop อยู่แล้ว ดังนั้น top-level asyncio.run(...) ใน cell จะ raise error เดียวกัน — ให้ await coroutine ตรงๆ แทน

🐍 ตัวอย่างนี้ตรวจสอบแล้วบน CPython เท่านั้น

เมื่อรันในเครื่อง asyncio.run(inner()) ด้านบน raise RuntimeError: asyncio.run() cannot be called from a running event loop ได้แน่นอนบน python3.14 แต่การทดสอบโค้ดเดียวกันใน Pyodide sandbox ของคอร์สนี้ไม่พบ error แบบนั้นเลย — การผสาน event loop ของ Pyodide ไม่ได้ปฏิเสธการเรียกซ้อนแบบเดียวกับที่ CPython ทำ เพราะสอง runtime ไม่ตรงกันตรงนี้ ตัวอย่างนี้จึงถูกตั้งใจไม่ใส่ใน Depth demo Playground ด้านล่าง ถือว่าเป็นข้อเท็จจริงที่ตรวจสอบแล้วบน CPython ไม่ใช่บนเบราว์เซอร์

Depth demo: ทุกอย่างข้างบน รันจริง

Open in an online REPL

import asyncio
import random

# --- create_task fire-and-forget: keep a reference, or use TaskGroup ---
async def background_job(label: str) -> None:
    await asyncio.sleep(0.02)
    print(f"  {label}: finished")

async def demo_create_task_safety() -> None:
    background_tasks: set[asyncio.Task] = set()
    task = asyncio.create_task(background_job("tracked"))
    background_tasks.add(task)               # keep a strong reference
    task.add_done_callback(background_tasks.discard)
    await asyncio.sleep(0.05)
    print(f"  background_tasks now empty: {len(background_tasks) == 0}")

# --- TaskGroup cancellation vs gather(return_exceptions=True) ---
async def worker(label: str, delay: float, fail: bool) -> str:
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"{label} failed")
    return f"{label} ok"

async def demo_gather_vs_taskgroup() -> None:
    results = await asyncio.gather(
        worker("A", 0.05, False), worker("B", 0.02, True), worker("C", 0.08, False),
        return_exceptions=True,
    )
    print(f"  gather(return_exceptions=True): {results} -- all three ran")

    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(worker("A", 0.05, False))
            tg.create_task(worker("B", 0.02, True))
            tg.create_task(worker("C", 0.08, False))
    except* ValueError as eg:
        print(f"  TaskGroup: {[str(e) for e in eg.exceptions]} -- B's failure cancelled C early")

# --- asyncio.timeout() ---
async def demo_timeout() -> None:
    try:
        async with asyncio.timeout(0.05):
            await asyncio.sleep(5)
    except TimeoutError:
        print("  asyncio.timeout() cancelled the block")

# --- time.sleep would block the loop; asyncio.to_thread does not ---
def blocking_sleep(seconds: float) -> str:
    import time
    time.sleep(seconds)
    return "worker done"

async def heartbeat(n: int) -> None:
    for i in range(n):
        await asyncio.sleep(0.03)
        print(f"  heartbeat {i}")

async def demo_to_thread() -> None:
    hb = asyncio.create_task(heartbeat(2))
    result = await asyncio.to_thread(blocking_sleep, 0.08)
    print(f"  {result}")
    await hb
    print("  heartbeat kept ticking through the blocking call")

# --- Semaphore bounds concurrency ---
async def limited(sem: asyncio.Semaphore, label: str, active: list, peak: list) -> None:
    async with sem:
        active.append(label)
        peak[0] = max(peak[0], len(active))
        await asyncio.sleep(0.03)
        active.remove(label)

async def demo_semaphore() -> None:
    sem = asyncio.Semaphore(2)
    active: list = []
    peak = [0]
    await asyncio.gather(*[limited(sem, f"job-{i}", active, peak) for i in range(4)])
    print(f"  peak concurrent workers: {peak[0]} (bounded by Semaphore(2))")

# --- Queue producer/consumer ---
async def demo_queue() -> None:
    q: asyncio.Queue = asyncio.Queue(maxsize=2)

    async def producer() -> None:
        for i in range(4):
            await q.put(i)
        await q.put(None)

    async def consumer(out: list) -> None:
        while True:
            item = await q.get()
            if item is None:
                break
            out.append(item)

    out: list = []
    await asyncio.gather(producer(), consumer(out))
    print(f"  consumed: {out}")

# --- Lock: check-then-act across an await is still a race ---
class Account:
    def __init__(self, balance: int) -> None:
        self.balance = balance
        self.lock = asyncio.Lock()

async def withdraw_unsafe(acct: Account, amount: int) -> None:
    if acct.balance >= amount:
        await asyncio.sleep(0)
        acct.balance -= amount

async def withdraw_safe(acct: Account, amount: int) -> None:
    async with acct.lock:
        if acct.balance >= amount:
            await asyncio.sleep(0)
            acct.balance -= amount

async def demo_lock() -> None:
    unsafe_acct = Account(100)
    await asyncio.gather(withdraw_unsafe(unsafe_acct, 80), withdraw_unsafe(unsafe_acct, 80))
    print(f"  unsafe balance: {unsafe_acct.balance} (went negative -- lost update)")

    safe_acct = Account(100)
    await asyncio.gather(withdraw_safe(safe_acct, 80), withdraw_safe(safe_acct, 80))
    print(f"  safe balance:   {safe_acct.balance} (Lock serialized the two withdrawals)")

# --- retry with backoff + jitter ---
class FlakyError(Exception):
    pass

async def flaky_call(attempts: list) -> str:
    attempts[0] += 1
    if attempts[0] < 3:
        raise FlakyError(f"attempt {attempts[0]} failed")
    return "success"

async def retry_with_backoff(coro_fn, *, max_attempts: int = 5, base: float = 0.01) -> str:
    rng = random.Random(0)   # seeded for a reproducible demo
    for attempt in range(1, max_attempts + 1):
        try:
            return await coro_fn()
        except FlakyError:
            if attempt == max_attempts:
                raise
            delay = base * (2 ** (attempt - 1)) + rng.uniform(0, base)
            await asyncio.sleep(delay)
    raise AssertionError("unreachable")

async def demo_retry() -> None:
    attempts = [0]
    result = await retry_with_backoff(lambda: flaky_call(attempts))
    print(f"  {result} after {attempts[0]} attempts")

# --- rate limiting (token bucket) ---
class RateLimiter:
    def __init__(self, rate: int, period: float) -> None:
        self.rate = rate
        self.period = period
        self.tokens = float(rate)

    async def acquire(self) -> None:
        while self.tokens < 1:
            await asyncio.sleep(self.period / self.rate)
            self.tokens += 1
        self.tokens -= 1

async def demo_rate_limit() -> None:
    limiter = RateLimiter(rate=2, period=0.06)
    log: list = []

    async def call(label: str) -> None:
        await limiter.acquire()
        log.append(label)

    await asyncio.gather(*[call(f"req-{i}") for i in range(4)])
    print(f"  completed {len(log)} calls, throttled to 2 per 0.06s")

# --- graceful shutdown: cancel + let tasks observe a shutdown signal ---
async def demo_graceful_shutdown() -> None:
    shutdown = asyncio.Event()

    async def worker_loop() -> None:
        ticks = 0
        while not shutdown.is_set():
            ticks += 1
            await asyncio.sleep(0.02)
        print(f"  worker saw shutdown and exited cleanly (ticks: {ticks}, timing-dependent)")

    async def trigger() -> None:
        await asyncio.sleep(0.05)
        shutdown.set()

    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker_loop())
        tg.create_task(trigger())

async def main() -> None:
    print("=== create_task: keep a reference, don't fire-and-forget ===")
    await demo_create_task_safety()
    print("=== TaskGroup vs gather(return_exceptions=True) ===")
    await demo_gather_vs_taskgroup()
    print("=== asyncio.timeout() ===")
    await demo_timeout()
    print("=== to_thread keeps the loop responsive ===")
    await demo_to_thread()
    print("=== Semaphore bounds concurrency ===")
    await demo_semaphore()
    print("=== Queue producer/consumer ===")
    await demo_queue()
    print("=== Lock across an await point ===")
    await demo_lock()
    print("=== retry with backoff + jitter ===")
    await demo_retry()
    print("=== rate limiting ===")
    await demo_rate_limit()
    print("=== graceful shutdown ===")
    await demo_graceful_shutdown()

asyncio.run(main())

Mental Model

ลองนึกภาพ TaskGroup เป็น project manager ที่จะไม่กลับบ้านจนกว่าทุก task ที่มอบหมายจะเสร็จหรือถูกยกเลิกอย่างชัดเจน — และยกเลิกทีมที่เหลือทันทีที่ task หนึ่งล้มเหลว เพราะการเดินหน้าต่อไปสู่เป้าหมายที่พังไปแล้วเสียเวลาทุกคน gather คือ manager ที่ไม่มีสัญชาตญาณแบบนั้น — เก็บผลลัพธ์จากทุกคนไม่ว่า success หรือ failure แล้วปล่อยให้คุณแยกเอง ไม่มี manager คนไหนทำให้ worker ปลอดภัยจากกันเอง — นั่นยังเป็นหน้าที่ของ Lock, Semaphore, หรือ Queue ที่คุณเพิ่มเอง

sequenceDiagram
participant TG as TaskGroup
participant A as Task A
participant B as Task B (fails)
participant C as Task C
TG->>A: create_task
TG->>B: create_task
TG->>C: create_task
B-->>TG: raises ValueError
TG->>C: cancel()
TG-->>TG: raise ExceptionGroup([ValueError])

Task หนึ่งล้มเหลวใน TaskGroup ยกเลิก sibling ก่อนที่จะเสร็จ

ข้อแลกเปลี่ยน

สิ่งที่ได้	ประโยชน์	ต้นทุน
`create_task` (เปล่าๆ)	เริ่มงานทันที พิธีกรรมน้อย	ไม่มีอะไรบังคับให้เก็บ reference — ถูกเก็บกวาดแบบเงียบๆ ได้
`TaskGroup`	lifetime มีโครงสร้าง failure หนึ่งยกเลิกที่เหลือ	all-or-nothing; batch ที่ต้องการ partial success ต้องใช้ `gather` แทน
`gather(return_exceptions=True)`	ทุก task รันจนจบไม่ว่าตัวอื่นจะ fail	ต้องเช็ค `isinstance(r, Exception)` เองทุกผลลัพธ์
`Semaphore` / rate limiting	ปกป้อง service หรือ pool ปลายทางจาก overload	เพิ่ม latency ให้ผู้เรียกเมื่อถึง limit
`asyncio.to_thread`	blocking call ไม่ทำให้ loop แข็งอีกต่อไป	overhead ของ thread pool; งาน pure-Python ยังโดน GIL
Retry แบบ backoff + jitter	รอด transient failure โดยไม่ยิง server ที่กำลังแย่ซ้ำๆ	บัง dependency ที่พังถาวรไว้จนครบ `max_attempts` ครั้งก่อน

ความเข้าใจผิดที่พบบ่อย

“create_task กับ await coro ใช้แทนกันได้” create_task schedule ทันทีแล้วคืนการควบคุมทันที; await coro เปล่าๆ รัน (และ block บน) coroutine ตัวนั้นก่อนไปต่อ สลับสองอย่างนี้คือวิธีที่ fan-out แบบ “concurrent” กลายเป็น sequential แบบเงียบๆ
“task ที่ไม่มีตัวแปรชี้ถึงจะถูก garbage collect แน่นอน” ปกติ loop เองมักเก็บมันไว้ให้เสร็จได้ในตัวอย่างสั้นๆ — ซึ่งนี่แหละที่ทำให้ bug นี้พลาดง่ายตอน code review และไม่แน่นอนเมื่อพยายาม reproduce คำแนะนำอย่างเป็นทางการให้เก็บ reference มีอยู่ไม่ว่า run ไหนจะแสดงปัญหาหรือไม่
“gather(return_exceptions=True) กับ TaskGroup เป็นแค่การเขียนสองแบบของสิ่งเดียวกัน” ทั้งสองต่างกันโดยตั้งใจ: gather ไม่เคยยกเลิก sibling; TaskGroup ยกเลิกเสมอเมื่อ fail ครั้งแรก
“event loop เธรดเดียวหมายความว่า shared state ไม่ต้องใช้ lock เลย” ไม่มี coroutine สองตัวรัน bytecode ในเสี้ยววินาทีเดียวกันจริง แต่ลำดับ check-then-act ที่คร่อม await ยังเป็น race — check อาจเก่าไปแล้วตอนที่ action รัน
“asyncio.shield ทำให้ coroutine ยกเลิกไม่ได้” มันปกป้องแค่จาก cancellation ที่แพร่มาจากการยกเลิกของผู้เรียก; coroutine ที่ shield ยังถูกยกเลิกเองได้ และ timeout()/wait_for ที่ครอบ shield ยังมีผล

อ่านโค้ดที่ AI เขียน

ก่อนเชื่อโค้ด task/concurrency ที่ AI สร้าง ตรวจสิ่งเหล่านี้:

ทุก asyncio.create_task(...) เก็บผลลัพธ์ไว้ในสิ่งที่มีอายุยืนกว่า statement ปัจจุบันหรือไม่ (ตัวแปร, set, TaskGroup)? asyncio.create_task(coro()) เปล่าๆ บนบรรทัดของตัวเองคือ fire-and-forget task ที่ไม่มีอะไรติดตาม
ใช้ gather ในจุดที่ task ทำงานไปสู่เป้าหมายร่วมกันจริงๆ หรือไม่ (ควรเป็น TaskGroup ให้ failure หนึ่งยกเลิกที่เหลือหรือเปล่า)?
มี shared object ที่ถูกอ่านแล้วแก้ไข โดยมี await คั่นกลาง โดยไม่มี Lock ครอบทั้งลำดับหรือไม่?
มี blocking call จริงๆ (time.sleep, requests.get แบบ synchronous, งาน CPU หนัก) อยู่ตรงๆ ใน async def แทนที่จะอยู่หลัง asyncio.to_thread หรือ run_in_executor หรือไม่?
retry loop retry ไม่มีที่สิ้นสุดโดยไม่มี max_attempts หรือ retry ด้วย delay คงที่ที่อาจทำให้ client หลายตัว sync กันยิง server ที่กำลังแย่พร้อมกัน (ไม่มี jitter) หรือไม่?

🐛 หา bug ในโค้ดนี้

async def fetch(session, url: str) -> str:
  async with session.get(url) as resp:
      return await resp.text()

async def fetch_all(session, urls: list[str]) -> list[str]:
  results = []
  for url in urls:
      results.append(await fetch(session, url))
  return results

เฉลย

โค้ดนี้อ่านแล้วดูเหมือน “fetch ทุก URL” และมันก็ให้ผลลัพธ์ที่ถูกต้องจริง — เลยง่ายที่จะผ่าน review แต่ for loop await ทีละ URL: การ fetch 10 URL ที่ 200ms ต่อครั้งใช้เวลารวมประมาณ 2 วินาที แทนที่จะเป็นประมาณ 200ms ฟังก์ชันเป็น async def และมีการใช้ await จริง ซึ่งทำให้คอขวดแบบ sequential พลาดง่ายถ้าไม่ได้วัดเวลา วิธีแก้คือ schedule ทุก fetch พร้อมกันแล้วรอทั้งหมดพร้อมกัน: results = await asyncio.gather(*[fetch(session, u) for u in urls]) หรือใช้ TaskGroup ถ้าอยากให้ fetch ที่ fail ยกเลิกที่เหลือ

เวลารวมโดยประมาณเมื่อ await fetch('A', 0.3) และ await fetch('B', 0.2) ถูกเรียกแบบ sequential (ทีละตัว) คือเท่าไหร่? ประมาณ 0.3 s — นับแค่ตัวที่ช้าที่สุด ประมาณ 0.2 s — นับแค่ตัวที่เร็วที่สุด ประมาณ 0.5 s — เกิด delay ทั้งสองตัว ประมาณ 0.0 s — asyncio รันทันที

asyncio.create_task(coro) ทำอะไรที่ await coro เปล่าๆ ไม่ทำ? รัน coroutine แบบ synchronous schedule coroutine ทันทีบน event loop ให้รันแบบ concurrent ได้ สร้าง OS thread ใหม่สำหรับ coroutine ปิดการยกเลิกสำหรับ coroutine นั้น

ทำไม asyncio.create_task(coro()) บนบรรทัดของตัวเอง โดยไม่มีตัวแปรเก็บผลลัพธ์ ถึงเป็นความเสี่ยง bug จริง แม้ demo สั้นๆ จะทำงานได้พอดี? เป็น SyntaxError ใน Python 3.11+ event loop เก็บแค่ weak reference ของ task ดังนั้นไม่มีอะไรที่คุณควบคุมได้รับประกันว่ามันจะรอดจนเสร็จ create_task raise เสมอถ้าผลลัพธ์ถูกทิ้ง มันแปลง task เป็น synchronous call แบบเงียบๆ

เกิดอะไรขึ้นกับ task C เมื่อ B raise?

async def worker(label, delay, fail):
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(label)
    return label

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker("A", 0.05, False))
        tg.create_task(worker("B", 0.01, True))
        tg.create_task(worker("C", 0.2, False))

C รันจนจบไม่ว่า B จะ fail หรือไม่ TaskGroup ยกเลิก C ทันทีที่เห็น exception ของ B ก่อนที่ delay เต็มของ C จะผ่านไป exception ของ B ถูกเพิกเฉยแบบเงียบๆ โปรแกรมทั้งหมด crash ทันทีโดยไม่มีทาง catch error

อะไรผิดปกติกับ time.sleep(2) ภายใน async def coroutine ที่มี coroutine อื่นรันแบบ concurrent อยู่ด้วย? ไม่มีอะไรผิด -- asyncio แปลงเป็น asyncio.sleep ให้อัตโนมัติ raise SyntaxError เพราะ sleep ไม่ถูก await มัน block ทั้ง event loop thread เป็นเวลา 2 วินาที ทำให้ coroutine อื่นแข็งไปด้วย มันมีผลแค่กับ coroutine ที่เรียกมันเท่านั้น

โค้ดนี้พิมพ์อะไร และเพราะอะไร?

class Account:
    def __init__(self, balance):
        self.balance = balance

async def withdraw(acct, amount):
    if acct.balance >= amount:
        await asyncio.sleep(0)
        acct.balance -= amount

async def main():
    acct = Account(100)
    await asyncio.gather(withdraw(acct, 80), withdraw(acct, 80))
    print(acct.balance)

20 -- การถอนครั้งที่สองเห็น balance ที่ลดแล้วถูกต้อง -60 -- ทั้งสอง coroutine อ่าน balance=100 ก่อนที่ตัวใดตัวหนึ่งจะเขียนค่าใหม่ raise exception เพราะ balance จะติดลบ 100 -- ไม่มีการถอนไหนถูกใช้จริง

สคริปต์ CLI เรียก asyncio.run(main()) ภายใน main() ฟังก์ชัน helper เรียก asyncio.run(fetch_config()) เกิดอะไรขึ้น? ทำงานได้ปกติ -- การเรียกซ้อนกันแค่สร้าง event loop ซ้อน RuntimeError: asyncio.run() cannot be called from a running event loop fetch_config() ไม่รันเลยแบบเงียบๆ raise DeprecationWarning แต่ยังทำงานได้