asyncio พื้นฐาน

Event loop

asyncio ทำงานบน event loop: ตัว scheduler ที่เก็บ queue ของ coroutines และ callbacks, รันทีละตัว และสลับระหว่างกันเมื่อตัวใดหยุดที่ expression await

ไม่มีการสร้าง OS threads ใดๆ concurrency เป็นแบบ cooperative — coroutine ต้องยกการควบคุมโดยสมัครใจด้วย await เพื่อให้ coroutines อื่นทำงานได้

import asyncio

async def say(message: str, delay: float) -> None:
    await asyncio.sleep(delay)
    print(message)

async def main() -> None:
    await say("Hello", 0.1)
    await say("World", 0.1)

asyncio.run(main())
# Hello
# World

ในตัวอย่างแบบ sequential นี้ say("World", …) ยังไม่เริ่มจนกว่า say("Hello", …) จะเสร็จสิ้น

async def และ coroutines

ฟังก์ชันที่กำหนดด้วย async def คือ coroutine function การเรียกฟังก์ชันนี้ไม่ได้รันเนื้อหา — แต่คืน coroutine object เนื้อหาจะรันเฉพาะเมื่อ coroutine ถูก await หรือถูก schedule บน event loop เท่านั้น

import asyncio

async def greet(name: str) -> str:
    return f"Hello, {name}!"

# Calling greet() returns a coroutine object — nothing runs yet
coro = greet("Python")
print(type(coro))          # <class 'coroutine'>

# Never awaiting it is a real bug. Under `python3.14 -X dev`, deleting the
# never-awaited coroutine triggers:
#   RuntimeWarning: coroutine 'greet' was never awaited
del coro

# Fix: await it to actually run the body
result = asyncio.run(greet("Python"))
print(result)              # Hello, Python!

# If you decide not to run a coroutine after all, close it explicitly —
# closing (instead of just dropping the reference) silences the warning
unused = greet("Python")
unused.close()

Warning เรื่อง never-awaited ไม่ใช่แค่ความสวยงาม

RuntimeWarning ด้านบนคือ CPython กำลังบอกว่าเนื้อหาของ coroutine ไม่เคยรันเลย — ไม่ใช่แค่ lint จุกจิก ในโค้ด production เรื่องนี้มักหมายความว่าลืมใส่ await หน้าการเรียก ซึ่งทำให้ network request, database write หรือ asyncio.sleep ที่ตั้งใจไว้กลายเป็น no-op แบบเงียบๆ:

async def send_email(to: str) -> None:
    print(f"sending email to {to}")

async def process_signup(email: str) -> None:
    send_email(email)   # BUG: missing `await` — the coroutine object is created and discarded
    print("signup processed")

การเรียก process_signup("[email protected]") จะพิมพ์ signup processed และ return ตามปกติ — ไม่มี exception ไม่มีอาการชัดเจน สัญญาณเดียวที่มีแบบ real-time คือ RuntimeWarning เมื่อ coroutine object ถูก garbage-collect นี่คือเหตุผลที่ -X dev (หรือ -W error::RuntimeWarning ใน CI) ควรอยู่ใน test suite ของ asyncio ทุกชุด: มันเปลี่ยน no-op เงียบๆ ให้กลายเป็น failure ทันทีที่จุดเรียกที่ลืม await

await หยุดและกลับมาทำงาน

keyword await ใช้ได้เฉพาะภายในฟังก์ชัน async def เท่านั้น await หยุด coroutine ปัจจุบันและยกการควบคุมกลับไปยัง event loop event loop รัน coroutines อื่นที่พร้อมแล้วกลับมาที่ coroutine นี้เมื่อ awaited object เสร็จสิ้น

import asyncio

async def fetch_data(label: str) -> str:
    print(f"  {label}: starting fetch")
    await asyncio.sleep(0.05)          # suspends here
    print(f"  {label}: fetch done")
    return f"data-{label}"

async def main() -> None:
    result = await fetch_data("A")
    print(result)

asyncio.run(main())
#   A: starting fetch
#   A: fetch done
# data-A

await แบบ sequential ใน loop เทียบกับ asyncio.gather

การเขียน for item in items: await do(item) คือรูปแบบที่พบบ่อยที่สุดที่ทำให้โค้ด async ที่ AI สร้างดูเหมือน concurrent แต่จริงๆ ไม่ใช่ แต่ละ iteration รอทั้งหมดจนกว่า iteration ก่อนหน้าจะเสร็จ — loop นี้เป็นแบบ sequential ทีละ await:

import asyncio
import time

async def fetch(label: str) -> str:
    await asyncio.sleep(0.2)
    return f"{label} done"

async def sequential() -> None:
    start = time.perf_counter()
    results = []
    for label in ["A", "B", "C"]:
        results.append(await fetch(label))   # each await blocks the loop until this one finishes
    print(f"sequential: {round(time.perf_counter() - start, 1)}s")   # ~0.6s (3 × 0.2s)

asyncio.run(sequential())

asyncio.gather schedule coroutines ทั้งสามตัวพร้อมกันแทน ดังนั้นเวลารวมคือของตัวที่ ช้าที่สุด ไม่ใช่ผลรวม:

import asyncio
import time

async def fetch(label: str) -> str:
    await asyncio.sleep(0.2)
    return f"{label} done"

async def concurrent() -> None:
    start = time.perf_counter()
    await asyncio.gather(fetch("A"), fetch("B"), fetch("C"))
    print(f"gather: {round(time.perf_counter() - start, 1)}s")   # ~0.2s (all three overlap)

asyncio.run(concurrent())

กฎง่ายๆ คือ: for loop ที่ await งานที่เป็นอิสระต่อกันทีละตัวคือ bug ที่รอให้เจอตอน code review เว้นแต่แต่ละ iteration ต้องพึ่งผลลัพธ์ของ iteration ก่อนหน้าจริงๆ

asyncio.run — จุดเข้าหลัก

asyncio.run(coro) สร้าง event loop ใหม่, รัน coroutine จนเสร็จสมบูรณ์, ปิด loop และคืนค่าผลลัพธ์ เป็น entry point มาตรฐานสำหรับโปรแกรม async

import asyncio

async def compute(x: int) -> int:
    await asyncio.sleep(0)     # yield to event loop once
    return x ** 2

result = asyncio.run(compute(7))
print(result)    # 49

การเรียก asyncio.run ครั้งที่สอง แบบ sequential หลังจากครั้งแรก return และปิด loop ไปแล้ว ทำได้ปกติ — แต่ละครั้งได้ loop ใหม่เป็นของตัวเอง:

import asyncio

async def compute(x: int) -> int:
    await asyncio.sleep(0)
    return x ** 2

print(asyncio.run(compute(3)))   # 9
print(asyncio.run(compute(3)))   # 9 — a second, independent event loop, same process

สิ่งที่ ไม่ ปกติคือการเรียก asyncio.run จากภายใน coroutine ที่กำลังรันบน loop อยู่แล้ว — นั่นคือ nesting ไม่ใช่ sequencing และมันจะ raise RuntimeError ดูบท Async / Await สำหรับ failure mode แบบนี้

asyncio.gather — รัน coroutines แบบ concurrent

asyncio.gather(*coroutines) กำหนด schedule coroutines ทั้งหมดพร้อมกันและรอทั้งหมด ผลลัพธ์คืนในลำดับเดียวกับ input coroutines

import asyncio

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

async def main() -> None:
    results = await asyncio.gather(
        step("A", 0.3),
        step("B", 0.1),
        step("C", 0.2),
    )
    print(results)
    # ['A done', 'B done', 'C done']   — always input order
    # Total time ≈ 0.3 s, not 0.6 s

asyncio.run(main())

ผลลัพธ์อยู่ใน ลำดับ input เสมอ ไม่ว่าจะเสร็จในลำดับใด ซึ่งเหมือนกับ Promise.all ของ JavaScript

การจัดการ failure ด้วย return_exceptions=True

โดยค่าเริ่มต้น หาก coroutine ใด raise, gather จะส่งต่อ exception แรกและยกเลิกที่เหลือ ส่ง return_exceptions=True เพื่อรวบรวม exceptions เป็น values แทน (เหมือน Promise.allSettled)

import asyncio

async def risky(n: int) -> int:
    if n == 2:
        raise ValueError(f"n={n} is bad")
    return n * 10

async def main() -> None:
    results = await asyncio.gather(
        risky(1),
        risky(2),
        risky(3),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"OK: {r}")

asyncio.run(main())
# OK: 10
# Error: n=2 is bad
# OK: 30

Python 3.11+: asyncio.TaskGroup

asyncio.TaskGroup (3.11+) เป็น structured-concurrency ทางเลือกแทน gather หาก task ใด raise, group จะยกเลิก tasks ที่เหลือทั้งหมดโดยอัตโนมัติ

import asyncio

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

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(work("A", 0.2))
        task_b = tg.create_task(work("B", 0.1))
    # Both tasks are done here; exceptions are re-raised automatically
    print(task_a.result())
    print(task_b.result())

asyncio.run(main())

async with และ async for

Asynchronous context manager และ iterator มีเหตุผลเดียวกับ synchronous — setup/teardown และ iteration — เพียงแต่ __aenter__/__aexit__/__anext__ เป็น coroutines ดังนั้นแต่ละขั้นตอนสามารถ await ได้เอง (เช่น async DB driver ที่ flush connection, async generator ที่รอ chunk ถัดไป):

import asyncio

class AsyncResource:
    async def __aenter__(self) -> "AsyncResource":
        await asyncio.sleep(0.01)          # e.g. an async connection handshake
        print("acquired resource")
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        await asyncio.sleep(0.01)          # e.g. an async flush/close
        print("released resource")

async def numbers(n: int):
    for i in range(n):
        await asyncio.sleep(0.01)
        yield i                             # an async generator — defines __anext__ automatically

async def main() -> None:
    async with AsyncResource() as res:
        print("using the resource")

    async for i in numbers(3):
        print(f"got {i}")

asyncio.run(main())

การ break ออกจาก async for loop ก่อนกำหนดขณะวนบน async generator สามารถทำให้ generator ค้างอยู่กลาง yield โดยไม่ได้รัน cleanup (finally, try/except GeneratorExit) contextlib.aclosing ปิดมันให้แน่นอนไม่ว่ากรณีใด เหมือนกับที่ generator ธรรมดาใน for loop ได้ฟรีจาก reference counting:

from contextlib import aclosing

async def numbers(n: int):
    for i in range(n):
        yield i

async def main() -> None:
    async with aclosing(numbers(5)) as gen:
        async for i in gen:
            if i == 1:
                break
    # gen.aclose() already ran here, even though the loop exited early

asyncio.run(main())

CancelledError เป็น BaseException — ต้อง re-raise เสมอ

asyncio.CancelledError inherit จาก BaseException ไม่ใช่ Exception (ตั้งแต่ Python 3.8) เป็นการตั้งใจ: handler except Exception: แบบกว้างๆ — ที่มีไว้จับ error ระดับ application — ต้องไม่จับและทิ้ง cancellation request โดยไม่ตั้งใจ:

import asyncio

async def leaky_cancel_handler() -> None:
    try:
        await asyncio.sleep(10)
    except Exception as e:
        # This does NOT match CancelledError — it inherits from BaseException.
        print(f"caught: {e!r}")

async def main() -> None:
    task = asyncio.create_task(leaky_cancel_handler())
    await asyncio.sleep(0.01)
    task.cancel()
    await task   # CancelledError propagates straight through — except Exception never ran

asyncio.run(main())

ถ้าต้องการดัก cancellation เพื่อทำ cleanup — ปิดไฟล์, ปล่อย lock — ให้จับ asyncio.CancelledError โดยเฉพาะ รัน cleanup แล้วจบด้วย raise เพื่อให้ cancellation ทำงานต่อเสมอ การกลืนมัน (except asyncio.CancelledError: pass) จะทำให้ task ดู “ถูกยกเลิก” ในสายตาผู้เรียก ทั้งที่โค้ดของคุณยังรันต่อผ่านจุด cancellation ไป ซึ่งทำลาย guarantee ที่ TaskGroup/wait_for/timeout() ทั้งหมดพึ่งพา

Debug mode และการส่องโปรแกรมที่กำลังรันอยู่ (Python 3.14)

asyncio.run(main(), debug=True) (หรือ environment variable PYTHONASYNCIODEBUG=1) เปิด check เพิ่มเติม: log coroutine ที่ไม่เคยถูก await ตั้งแต่ตอนที่เกิดขึ้น (ไม่ใช่แค่ตอน GC), เตือนเมื่อ callback block loop นานเกินไป และรายงาน “Task was destroyed but it is pending”

เวลา async service ค้าง คำถามที่ยากคือ coroutine ตัวไหนที่ติดอยู่ Python 3.14 เข้าไปเกาะ process ที่ กำลังรัน แล้วพิมพ์ task tree ออกมาได้เลย ไม่ต้องใส่ instrument ไม่ต้อง restart:

python -m asyncio ps 12345      # flat table of tasks in PID 12345
python -m asyncio pstree 12345  # the await tree

ความสามารถนี้ต่อยอดจาก external debugger interface ของ PEP 768 ซึ่งทำให้ python -m pdb -p 12345 เกาะ process ที่รันอยู่ได้ด้วย ส่วนภายใน code ของคุณเอง asyncio.capture_call_graph() และ asyncio.print_call_graph() ให้ภาพเดียวกันสำหรับ task ปัจจุบัน

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

Run ▸ Open in an online REPL

import asyncio
import time
from contextlib import aclosing

# --- sequential await in a loop vs asyncio.gather ---
async def fetch(label: str) -> str:
    await asyncio.sleep(0.2)
    return f"{label} done"

async def sequential() -> float:
    start = time.perf_counter()
    results = []
    for label in ["A", "B", "C"]:
        results.append(await fetch(label))  # each iteration waits for the previous one
    elapsed = time.perf_counter() - start
    print(f"  sequential: {results}")
    return elapsed

async def concurrent() -> float:
    start = time.perf_counter()
    results = await asyncio.gather(fetch("A"), fetch("B"), fetch("C"))
    elapsed = time.perf_counter() - start
    print(f"  gather:     {results}")
    return elapsed

# --- async with / async for / async generators ---
class AsyncResource:
    async def __aenter__(self) -> "AsyncResource":
        await asyncio.sleep(0.01)
        print("  acquired resource")
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        await asyncio.sleep(0.01)
        print("  released resource")

    async def read(self) -> str:
        return "resource data"

async def numbers(n: int):
    for i in range(n):
        await asyncio.sleep(0.01)
        yield i

async def demo_async_with_for() -> None:
    async with AsyncResource() as res:
        print(f"  got: {await res.read()}")

    async for i in numbers(3):
        print(f"  async for got: {i}")

    gen = numbers(5)
    async with aclosing(gen):
        async for i in gen:
            print(f"  early-exit item: {i}")
            if i == 1:
                break
    print("  aclosing closed the generator cleanly despite the early break")

# --- CancelledError is a BaseException, not an Exception ---
async def leaky_cancel_handler() -> None:
    try:
        await asyncio.sleep(10)
    except Exception as e:
        print(f"  except Exception caught: {e!r}")  # never runs

async def correct_cancel_handler() -> None:
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("  cleaning up before re-raising")
        raise

async def demo_cancellation() -> None:
    task = asyncio.create_task(leaky_cancel_handler())
    await asyncio.sleep(0.01)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("  CancelledError propagated out untouched -- except Exception inside never matched it")

    task2 = asyncio.create_task(correct_cancel_handler())
    await asyncio.sleep(0.01)
    task2.cancel()
    try:
        await task2
    except asyncio.CancelledError:
        print("  task cancelled cleanly after cleanup ran")

async def main() -> None:
    print("=== sequential vs gather ===")
    seq_time = await sequential()
    con_time = await concurrent()
    print(f"  sequential took {round(seq_time, 1)}s, gather took {round(con_time, 1)}s")

    print("=== async with / async for / async generators ===")
    await demo_async_with_for()

    print("=== CancelledError must be re-raised ===")
    await demo_cancellation()

asyncio.run(main())

🐍 รันตัวอย่าง walkthrough ในเครื่องของคุณ

ตัวอย่างแบบ prose ด้านบนใช้ \“pythonfence ธรรมดา และตั้งใจให้วางลงไฟล์.pyแล้วรันด้วยpython3.14` Depth demo Playground ด้านบนรัน idea เดียวกันแบบ live ในเบราว์เซอร์ผ่าน Pyodide

Mental Model

ลองนึกภาพ event loop เป็นพนักงานคนเดียวที่โต๊ะเดียวมี in-tray เดียว ทุก coroutine คือ task ที่นั่งอยู่ใน tray นั้น พนักงานหยิบ task มาทำจนกว่าจะเจอ await — นั่นคือ coroutine กำลังบอกว่า “ฉันกำลังรอสิ่งภายนอกอยู่ ช่วยเอาฉันกลับไปวางใน tray แล้วไปทำงานของคนอื่นระหว่างนี้” พนักงานไม่เคยทำสอง task พร้อมกันในเสี้ยววินาทีเดียวกัน เพียงแต่ไม่เคยนั่งว่างรอ I/O ตัวหนึ่งในขณะที่ task อื่นพร้อมทำงานอยู่

sequenceDiagram
participant EventLoop as Event loop
participant A as Coroutine A
participant B as Coroutine B
A->>EventLoop: await asyncio.sleep(0.2)
EventLoop->>B: run B until its next await
B->>EventLoop: await asyncio.sleep(0.1)
EventLoop-->>B: B's sleep is done first — resume B
B->>EventLoop: return
EventLoop-->>A: A's sleep is done — resume A
A->>EventLoop: return

await หนึ่งครั้งคืนการควบคุมให้ loop เสมอ loop จะรัน coroutine ที่พร้อมก่อนตัวไหนก่อน

นี่คือเหตุผลที่ coroutine เดียวที่ไม่เคย await ใน loop ยาวๆ ทำให้ทุกตัวอื่นอดตาย: พนักงานไม่สามารถวาง task ลงแล้วสลับออกไปได้ — มีแค่ await เท่านั้นที่สร้างโอกาสนั้น

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

สิ่งที่ได้	ประโยชน์	ต้นทุน
Event loop เธรดเดียว	ไม่ต้องใช้ lock สำหรับโค้ดระหว่างจุด await	coroutine ช้าตัวเดียวที่ไม่ await block ทุกตัวอื่น
asyncio.gather	fan-out แบบ concurrent ง่ายๆ โค้ดน้อย	exception แรกยกเลิก siblings เว้นแต่ใช้ return_exceptions=True
TaskGroup (3.11+)	lifetime มีโครงสร้าง ยกเลิก sibling อัตโนมัติเมื่อ fail	API ใหม่; ต้องใช้ except* เพื่อแกะ ExceptionGroup
async with / async for	setup/teardown และ iteration await ได้เอง	async for ที่ break ตรงๆ อาจข้าม cleanup ของ generator — ต้องใช้ aclosing
-X dev / debug mode	เห็น coroutine ที่ไม่เคย await และ callback ช้าทันที	overhead เพิ่ม — ใช้เฉพาะ dev/CI ไม่ใช่ production

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

• “asyncio ให้ parallelism” จริงๆ มันให้ concurrency บนเธรดเดียว coroutine สองตัวไม่เคยรัน bytecode ในเสี้ยววินาทีเดียวกัน ตัวหนึ่งต้อง yield ผ่าน await ก่อนตัวอื่นจะทำงานต่อ
• “for loop ของ await calls concurrent อยู่แล้วเพราะเป็น async code” ไม่จริง — แต่ละ await ใน loop body ยัง block iteration นั้นจนกว่าจะเสร็จ มีแค่การ schedule ด้วย gather/TaskGroup/create_task เท่านั้นที่ทำให้มันซ้อนกันได้
• “except Exception ปลอดภัยสำหรับ catch-all ใน async code” มันพลาด asyncio.CancelledError แบบเงียบๆ — ซึ่งถูกต้องตาม contract ของ Exception แต่มักทำให้คนที่คาดว่ามันจะจับ “อะไรก็ตามที่ผิดพลาด” ประหลาดใจ
• “เรียก asyncio.run สองครั้งคือ nesting event loops” เรียกสองครั้งติดกันที่ระดับบนสุดทำได้ปกติ — แต่ละครั้งได้ loop อิสระ Nesting หมายถึงเรียก จากภายใน coroutine ที่กำลังรันอยู่แล้ว ซึ่งเป็นคนละเรื่อง (และพังเสมอ)
• “coroutine ที่ไม่เคย await แค่เปลืองหน่วยความจำ” มันยังหมายความว่า เนื้อหาของ coroutine ไม่เคยรันเลย — side effect ใดๆ ข้างใน (การเขียน, request, การเปลี่ยนสถานะ) ไม่เคยเกิดขึ้นแบบเงียบๆ

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

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

• ทุก async def call ถูก await, create_task, หรือส่งเข้า gather/TaskGroup จริงหรือไม่? การเรียก coro_fn(...) เปล่าๆ โดยไม่มี await สร้าง coroutine object แล้วไม่ทำอะไรต่อ
• มี except Exception: แบบกว้างๆ ใน coroutine ที่ task.cancel() เข้าถึงได้หรือไม่? ถ้ามี มันจะ re-raise เมื่อเจอ asyncio.CancelledError หรือไม่ หรือโค้ดที่สร้างใช้ except: เปล่าๆ ที่กลืน cancellation ไปพร้อมกับ bug จริง?
• มี for loop ที่ await งานอิสระทีละตัวในจุดที่ควรใช้ gather/TaskGroup อย่างชัดเจนหรือไม่?
• async for บน generator มี break ก่อนกำหนดโดยไม่ใช้ aclosing หรือไม่?
• โค้ดรันภายใต้ -X dev หรือ -W error::RuntimeWarning ใน CI หรือไม่ เพื่อให้ await ที่ลืมใส่ fail แบบดังชัดแทนที่จะไม่ทำอะไรแบบเงียบๆ?

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

async def close_connection(conn) -> None:
  await conn.aclose()

async def handle_request(conn) -> None:
  try:
      await conn.serve()
  except asyncio.CancelledError:
      pass  # "handle" cancellation quietly
  finally:
      await close_connection(conn)

เฉลย

โค้ดนี้ดูรอบคอบ — มี finally ที่ปิด connection ด้วยซ้ำ แต่ except asyncio.CancelledError: pass กลืน cancellation แทนที่จะ re-raise: จากมุมมองผู้เรียก (TaskGroup, wait_for, หรือ timeout() ที่ยกเลิก task นี้) task ไม่ได้หยุดจริงๆ — มันยังรันต่อผ่านจุด cancellation ไปแล้ว return ตามปกติแทน โค้ดใดๆ ที่รอให้ task นี้ถูกยกเลิกจริง (เพื่อดำเนินการ shutdown ของตัวเอง หรือรู้ว่างานถูกยกเลิก) จะค้างหรือได้ผลลัพธ์ผิด วิธีแก้คือทำ cleanup ใน except block แล้ว raise (หรือลบ except asyncio.CancelledError ออกไปเลยแล้วพึ่ง finally ซึ่งรันเสมอไม่ว่า exception จะถูก re-raise หรือไม่)

🐍 await คือจุดเดียวที่การควบคุมเปลี่ยนมือ

ระหว่าง await expression สองตัว coroutine รันเหมือน synchronous code ธรรมดาทุกประการ — ไม่มี coroutine อื่นแทรกได้ Concurrency ใน asyncio เกิดขึ้นแค่ ที่ await เท่านั้น และเกิดเมื่อคุณ schedule มากกว่าหนึ่ง coroutine อย่างชัดเจน (gather, TaskGroup, create_task) for loop ของ await แบบ sequential ไม่ concurrent เพียงเพราะฟังก์ชันเป็น async def

การเรียกฟังก์ชัน async def คืนอะไรก่อนที่คุณจะ await?

ค่า return ของ function body coroutine object Future None

entry point มาตรฐานสำหรับรัน top-level async coroutine คืออะไร?

loop.run_until_complete(coro) asyncio.run(coro) await coro ที่ระดับ module asyncio.start(coro)

ใช้เวลาประมาณเท่าไหร่ และเพราะอะไร?

async def fetch(label, delay):
    await asyncio.sleep(delay)
    return label

async def main():
    results = []
    for label, delay in [("A", 0.2), ("B", 0.2), ("C", 0.2)]:
        results.append(await fetch(label, delay))
    print(results)

asyncio.run(main())

~0.2s -- asyncio รันทุกอย่างแบบ concurrent โดยค่าเริ่มต้น ~0.6s -- แต่ละ await ใน loop block จนกว่า call นั้นจะเสร็จก่อนตัวถัดไปจะเริ่ม ~0.0s -- await ไม่ block ค้างตลอดไป

ทำไม asyncio.CancelledError ถึงไม่ถูก except Exception: จับ?

การจับ CancelledError เป็น syntax error CancelledError inherit จาก BaseException ไม่ใช่ Exception โดยการออกแบบ asyncio ปิดการใช้ except Exception ภายใน coroutine มันเกิดขึ้นนอก try block เท่านั้น

เกิดอะไรขึ้นเมื่อรันโค้ดนี้?

async def main():
    print(9)

asyncio.run(main())
asyncio.run(main())

RuntimeError: event loop is already running พิมพ์ 9 สองครั้ง -- แต่ละ asyncio.run call ได้ event loop อิสระของตัวเอง asyncio.run ครั้งที่สองถูกข้ามแบบเงียบๆ deadlock

aclosing รับประกันอะไรที่ loop ธรรมดาไม่มี async with จะไม่รับประกัน?

async def numbers(n):
    for i in range(n):
        yield i

async def main():
    async with aclosing(numbers(5)) as gen:
        async for i in gen:
            if i == 1:
                break

loop จะรันครบทั้ง 5 item ไม่ว่าจะมี break cleanup ของ async generator (aclose) จะรันแม้ loop จะออกก่อนกำหนดผ่าน break generator จะเริ่มนับใหม่จาก 0 ในการเรียกครั้งถัดไป ไม่มีผลอะไร -- aclosing ไม่มีผลกับ generator

ภายใต้ python3.14 -X dev อะไรทำให้เกิด RuntimeWarning: coroutine '...' was never awaited?

การเรียกฟังก์ชัน async def ด้วยจำนวน argument ผิด coroutine object ถูก garbage-collect โดยไม่เคยถูก await หรือปิดเลย การใช้ async def แทน def การลืมเรียก asyncio.run เลย

Submit

Mark as complete