Async & Concurrency

เครื่องมือสามตัว สำหรับสามปัญหา

Python มีโมเดล concurrency ที่แตกต่างกันถึงสามแบบ แต่ละแบบเหมาะกับประเภทของงานที่ต่างกัน:

โมเดล	เหมาะกับ	กลไก
asyncio	งาน I/O-bound (เครือข่าย, ไฟล์, DB)	Single thread, cooperative multitasking
threading	งาน I/O-bound ที่ใช้ไลบรารีแบบ blocking	OS threads, ถูกจำกัดด้วย GIL
multiprocessing	งาน CPU-bound (การคำนวณ, การแยกวิเคราะห์)	Separate processes, parallelism จริง

การเลือกโมเดลผิดเป็นสาเหตุของบั๊กด้านประสิทธิภาพที่พบบ่อยที่สุดใน Python การเรียก database แบบ blocking ใน threading.Thread อาจดูปกติ แต่การเรียกแบบเดียวกันนั้นใน asyncio coroutine จะทำให้ event loop ทั้งหมดหยุดทำงาน

เหตุใด asyncio จึงครองตลาด Python ยุคใหม่

งานฝั่งเซิร์ฟเวอร์ส่วนใหญ่เป็น I/O-bound: รอ database, HTTP API, หรือการอ่านไฟล์ asyncio จัดการการรอพร้อมกันหลายพันรายการใน OS thread เดียว โดยรัน event loop ที่สลับการควบคุมระหว่าง coroutines เมื่อตัวใดตัวหนึ่งรอ I/O

import asyncio

async def fetch(url: str) -> str:
    await asyncio.sleep(0.1)   # simulates a network delay
    return f"Response from {url}"

async def main():
    results = await asyncio.gather(
        fetch("https://api.example.com/a"),
        fetch("https://api.example.com/b"),
    )
    for r in results:
        print(r)

asyncio.run(main())

การเรียกทั้งสองทำงานพร้อมกัน — ไม่มีตัวไหนบล็อกตัวอื่น เวลารวมประมาณ 0.1 วินาที ไม่ใช่ 0.2 วินาที

GIL และ Threads

CPython มี mutex ที่เรียกว่า Global Interpreter Lock (GIL) ซึ่งอนุญาตให้ thread เดียวเรียกใช้ Python bytecode ได้ในช่วงเวลาใดเวลาหนึ่ง สำหรับงาน I/O-bound นี้ไม่ใช่ปัญหา — threads จะปล่อย GIL ขณะรอ OS — แต่สำหรับงาน CPU-bound การเพิ่ม threads ไม่ได้ให้ความเร็วเพิ่มขึ้นเลย

import threading

def count_down(n: int) -> None:
    while n > 0:
        n -= 1

# Two threads — still runs on one core due to the GIL
t1 = threading.Thread(target=count_down, args=(5_000_000,))
t2 = threading.Thread(target=count_down, args=(5_000_000,))
t1.start(); t2.start()
t1.join();  t2.join()

Python 3.14 เพิ่มทางออกจากมุมนี้อีกสองทาง: free-threaded build ที่รองรับอย่างเป็นทางการแล้ว (python3.14t) ซึ่ง thread รันขนานกันได้จริง และ concurrent.interpreters ที่ให้ worker แต่ละตัวมี interpreter และ GIL ของตัวเอง ภายใน process เดียว ทั้งคู่อธิบายไว้ในบทเรียน GIL และ Multiprocessing

Multiprocessing สำหรับ parallelism จริง

multiprocessing สร้าง OS processes แยกกัน แต่ละ process มี Python interpreter และ GIL ของตัวเอง งาน CPU-bound จึงสามารถใช้ทุก core ที่มีได้

from multiprocessing import Pool

def square(x: int) -> int:
    return x * x

if __name__ == "__main__":        # required on every platform
    with Pool(processes=4) as pool:
        results = pool.map(square, range(10))
    print(results)
    # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

รันเฉพาะในเครื่องเท่านั้น — ไม่มี process จริงใน browser sandbox ต้องใส่ guard if __name__ == "__main__": เสมอทุกแพลตฟอร์ม: หากไม่มี, macOS/Windows (spawn) จะ import module นี้ใหม่ในทุก child process และสร้าง process ไม่สิ้นสุด ดูรายละเอียด guard/start-method ทั้งหมดได้ในบทเรียน Multiprocessing

ตารางตัดสินใจ: เครื่องมือไหนเหมาะกับงานไหน

Workload	ตัวเลือกแรก	ตัวเลือกรอง	เครื่องมือผิด
I/O-bound, มี async-native library (httpx, asyncpg)	asyncio	—	threading (ใช้ได้ แต่ overhead ต่อ connection มากกว่า)
I/O-bound, มีแค่ library แบบ blocking/synchronous (requests, boto3, psycopg2)	threading หรือ asyncio.to_thread	—	เรียก blocking library ตรงๆ ภายใน coroutine
CPU-bound, dependency รองรับ free-threading	Free-threaded python3.14t กับ threading	multiprocessing	threading บน default build (ไม่เร็วขึ้น)
CPU-bound, dependency ผูกกับ GIL เท่านั้น (ส่วนใหญ่ของ ecosystem วันนี้)	multiprocessing / ProcessPoolExecutor	concurrent.interpreters (3.14, startup ถูกกว่า)	threading (ไม่เร็วขึ้น)
งาน CPU สั้นๆ จำนวนมากที่ startup ของ process ครองเวลา	concurrent.interpreters (3.14)	ProcessPoolExecutor ที่มี pool อายุยืน	สร้าง Pool() ใหม่ทุกงาน
ผสม: มีทั้ง I/O และ CPU	asyncio + run_in_executor/to_thread สำหรับส่วน blocking, ProcessPoolExecutor สำหรับส่วน CPU	—	เลือกเครื่องมือเดียวแล้วยัดทุกอย่างผ่านมัน

คำถามเดียวที่ตัดสินสามแถวแรก: workload นี้ใช้เวลาส่วนใหญ่ไปกับการรอ หรือการคำนวณ? การรอ (network, disk, database round trip) หมายความว่า GIL ไม่เกี่ยวข้อง — เลือก asyncio ถ้า library รองรับ ไม่งั้นใช้ threading การคำนวณ (parsing, คำนวณตัวเลข, ประมวลผลภาพ) หมายความว่า GIL คือคอขวด — เลือก multiprocessing, concurrent.interpreters, หรือ free-threaded build

-X dev เผยให้เห็น bug ที่การรันปกติซ่อนไว้

Bug pattern ของ async/threading/multiprocessing ในโมดูลนี้ทุกตัว — coroutine ที่ไม่เคย await, ResourceWarning จาก resource ที่ไม่ปิด, exception ในเธรดที่หายไปเฉยๆ — มีคุณสมบัติร่วมกันหนึ่งอย่าง: มัน fail แบบเงียบๆภายใต้การรัน python3.14 script.py ปกติ โหมด -X dev เปลี่ยนหลายๆ อย่างนี้ให้เป็น warning ที่มองเห็นได้โดยไม่ต้องแก้โค้ดเลย:

python3.14 -X dev script.py

นี่คือเหตุผลที่ playground ทุกอันในโมดูลนี้ถูกตรวจสอบด้วย python3.14 -X dev -W error::RuntimeWarning แทนที่จะเป็น python3.14 เปล่าๆ — -W error::RuntimeWarning ไปไกลกว่า -X dev อีกขั้นโดยเปลี่ยน warning “coroutine was never awaited” ให้เป็น exception จริง ซึ่งเป็นพฤติกรรมที่ต้องการพอดีใน test suite หรือ CI pipeline: await ที่ลืมใส่ควรทำให้ build fail ไม่ใช่แค่พิมพ์ warning ที่ไหลผ่าน log ที่ไม่มีใครอ่าน

สิ่งที่โมดูลนี้ครอบคลุม

บทเรียน	แนวคิดหลัก
asyncio Basics	Event loop, async def, await, asyncio.run, asyncio.gather, async with/async for
Async / Await	TaskGroup เทียบ gather, Semaphore, Queue, Lock races, pattern retry/rate-limit/shutdown
GIL & Threads	GIL ปกป้องอะไร (และไม่ปกป้องอะไร), ThreadPoolExecutor, daemon threads, free-threaded 3.14t
Multiprocessing	Processes vs threads, ProcessPoolExecutor, picklability, subinterpreters

แต่ละบทเรียนสามารถเรียนได้อิสระ เริ่มด้วย asyncio หากสร้าง network services หรือข้ามไปที่ multiprocessing หากต้องการ parallelize งาน CPU-bound

🐍 Python 3.11+ task groups

Python 3.11 นำเสนอ asyncio.TaskGroup เป็นวิธีแนะนำในการรัน coroutines หลายตัวพร้อมกัน มี structured concurrency พร้อม automatic cancellation เมื่อเกิด failure — ควรใช้แทน gather แบบเปลือยในโค้ดใหม่

async with asyncio.TaskGroup() as tg:
    task_a = tg.create_task(fetch("https://a.example.com"))
    task_b = tg.create_task(fetch("https://b.example.com"))
# Both tasks are guaranteed to be done here

Mental Model

ลองนึกภาพสามโมเดลนี้เป็นสามวิธีจัดคนทำงาน: asyncio คือพนักงานคนเดียวที่จัดระเบียบเก่งมาก ไม่เคยนั่งว่างเลย — ขณะรออะไรบางอย่าง เขาหยิบงานถัดไปที่พร้อมทันที แล้วสลับกลับทันทีที่งานแรกพร้อม threading คือพนักงานหลายคนแชร์โต๊ะเดียวกันที่มีเครื่องมือใช้ครั้งละคนเดียว (GIL) — มีแค่คนเดียวแตะเครื่องมือได้ในเวลาหนึ่ง ดังนั้นพวกเขาจะนำหน้าพนักงานคนเดียวได้ก็ต่อเมื่อใช้เวลาส่วนใหญ่ทำอะไรที่ไม่ต้องใช้เครื่องมือ (รอ I/O) multiprocessing คือพนักงานหลายคนที่มีโต๊ะแยกกันสนิทและไม่มีเครื่องมือร่วมเลย — ขนานกันจริง แต่ทุกอย่างที่ส่งต่อกันต้องผ่านไปรษณีย์ (pickle) ซึ่งเสียเวลาและห้ามส่งอะไรที่ระบุที่อยู่ไม่ได้

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

สิ่งที่ได้	ประโยชน์	ต้นทุน
asyncio	การรอ I/O พร้อมกันนับพันบนเธรดเดียว หน่วยความจำน้อยมาก	ทุก library ใน call chain ต้องเป็น async-native ไม่งั้นกลับไป block loop
threading	ใช้แทนที่ library แบบ synchronous/blocking เดิมได้ทันที	ไม่เร็วขึ้นเชิง CPU บน default build; ต้องใช้ Lock สำหรับ shared state
multiprocessing	parallelism ของ CPU จริง แยก crash ระหว่าง worker	ต้นทุน process startup เต็ม; ข้ามได้แค่ data และ callable ที่ pickle ได้
Free-threaded 3.14t	parallelism ของ CPU จริงพร้อมความง่ายของ shared-memory แบบ thread	binary แยกต่างหาก; ระบบนิเวศ C extension ยังตามไม่ทัน
เลือกเครื่องมือเดียวสำหรับทุกอย่าง	codebase ง่ายกว่า mental model เดียว	บังคับงาน I/O-bound และ CPU-bound ผ่านเครื่องมือที่ไม่เหมาะกับทั้งคู่

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

• “asyncio เร็วที่สุดเสมอ” มันเป็นตัวเลือกที่ดีที่สุดสำหรับงาน I/O-bound ที่มี library แบบ async-native สำหรับงาน CPU-bound มันไม่ช่วยอะไรเลย — tight loop ภายใน coroutine ยัง block ทั้ง event loop เหมือนที่มัน block เธรดหนึ่ง
• “Threads กับ processes เป็นวิธีขนานกันแบบใช้แทนกันได้” Threads แชร์ memory และถูกจำกัดด้วย GIL สำหรับงาน CPU-bound; processes ข้าม GIL แต่ไม่แชร์ memory ดังนั้น argument และผลลัพธ์ทุกตัวเดินทางเป็น pickled data Free-threaded Python ลดช่องว่างนี้สำหรับ workload แบบ pure-Python แต่แค่บน python3.14t เท่านั้น
• “ถ้าไม่แน่ใจ ใช้ asyncio เป็นค่า default ที่ปลอดภัย” มันปลอดภัยก็ต่อเมื่อทุก library ใน call path เป็น async-native หรือถูก offload อย่างชัดเจน (to_thread, run_in_executor) การเพิ่ม synchronous blocking call แค่ตัวเดียวใน coroutine ทำให้ coroutine อื่นทั้งหมดบน loop แข็งแบบเงียบๆ
• “bug เงียบๆ ในโปรแกรม threading/asyncio หมายความว่าเครื่องมือใช้ไม่ได้” เกือบจะตรงข้ามเลย: -X dev และ -W error::RuntimeWarning มีอยู่ก็เพราะ bug เหล่านี้ (coroutine ที่ไม่เคย await, resource ที่ไม่ปิด, exception ที่ถูกกลืนในเธรด) เป็นเรื่องจริงและพบบ่อย — เปิดใช้ในทุก test suite แทนที่จะ debug ด้วยการลองผิดลองถูกใน production

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

ก่อนเชื่อโค้ด concurrency ที่ AI สร้างในโมดูลนี้ ให้ถาม:

• เครื่องมือที่เลือก (asyncio/threading/multiprocessing) ตรงกับว่า workload เป็น I/O-bound หรือ CPU-bound จริงหรือไม่? โค้ดที่สร้างมักเลือก asyncio เป็นค่า default โดยไม่สนใจ
• มี coroutine ที่เรียกฟังก์ชัน synchronous แบบ blocking (requests.get, time.sleep, การคำนวณหนัก) ตรงๆ แทนที่จะผ่าน to_thread/run_in_executor/process pool หรือไม่?
• โค้ด — หรือ test suite ของมัน — รันภายใต้ -X dev / -W error::RuntimeWarning หรือไม่? bug pattern ส่วนใหญ่ในโมดูลนี้ (coroutine ที่ไม่เคย await, CancelledError ที่ถูกกลืน, update ที่หายไปบน shared state) ไม่มีอาการที่มองเห็นได้ภายใต้การรันปกติที่ไม่มี flag
• มี lambda หรือ local closure ถูกส่งเข้า multiprocessing/ProcessPoolExecutor หรือไม่? มัน fail ด้วย PicklingError จริงตอนที่รันจริงบน worker ครั้งแรก ไม่ใช่ตอนที่เขียน

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

import asyncio
import time

async def fetch_user(user_id: int) -> dict:
  time.sleep(0.2)  # "quick" call to a synchronous ORM
  return {"id": user_id, "name": f"user-{user_id}"}

async def fetch_all_users(ids: list[int]) -> list[dict]:
  return await asyncio.gather(*[fetch_user(i) for i in ids])

เฉลย

โค้ดนี้ดูเหมือน asyncio แบบ idiomatic — async def, await, gather เพื่อความ concurrent — และมันก็คืน list ของ user dict ที่ถูกต้องจริง แต่ fetch_user เรียก time.sleep ซึ่ง block จริงๆ ตรงๆ ภายใน coroutine แทนที่จะเป็น async-native ORM call หรือ asyncio.to_thread เพราะ time.sleep ไม่คืนการควบคุมให้ event loop gather จึงไม่ให้ concurrency จริงเลยตรงนี้: ทุก “concurrent” fetch ยังรันทีละตัวบน event-loop thread เดียว ใช้เวลารวม 0.2 * len(ids) วินาทีแทนที่จะเป็นประมาณ 0.2 วินาที bug นี้ไม่มี exception และไม่มี warning — มีแค่ runtime ที่ไม่ scale แบบเงียบๆ ซึ่งเป็นเหตุผลพอดีที่การ profile ว่า “coroutine ตัวไหนคืนการควบคุมจริง” สำคัญกว่าการอ่าน keyword async/await แล้วสมมติว่า concurrency กำลังเกิดขึ้น

🐍 เลือกเครื่องมือให้ตรงกับที่เวลาไปจริงๆ

ก่อนหยิบ asyncio, threading, หรือ multiprocessing ให้ profile หรือคิดให้ออกว่าเวลาของ workload ไปอยู่ที่ไหน เวลาที่ใช้รอ (I/O) ต้องการ asyncio หรือ threading; เวลาที่ใช้คำนวณ (CPU) ต้องการ multiprocessing หรือ free-threaded build เดาผิดจะได้โค้ดที่ดู concurrent แต่ไม่ใช่ โดยไม่มี error ชี้ไปที่ความผิดพลาดเลย

โมเดล concurrency ใดของ Python เหมาะที่สุดสำหรับงาน CPU-bound เช่น การคำนวณเชิงตัวเลข?

asyncio — ใช้ event loop threading — OS threads รันแบบ parallel multiprocessing — separate processes ข้าม GIL ได้ ทั้งสามแบบเทียบเท่ากันสำหรับงาน CPU-bound

เหตุใด asyncio จึงจัดการการรอ I/O พร้อมกันหลายพันรายการได้อย่างมีประสิทธิภาพใน single thread?

คอมไพล์ coroutines เป็น native machine code Event loop สลับระหว่าง coroutines เมื่อตัวใดตัวหนึ่ง await I/O ใช้ GIL ในการ time-slice ระหว่าง coroutines แต่ละ coroutine รันใน subprocess ของตัวเอง

Global Interpreter Lock (GIL) คืออะไร?

Mutex ที่ป้องกันไม่ให้สอง threads รัน Python bytecode พร้อมกัน Lock ที่ป้องกัน asyncio จากการรันในโหมด multi-core Global variable ที่ติดตามจำนวน threads ที่กำลังรัน C extension ที่เร่งความเร็วโค้ด Python แบบ I/O-bound

ข้อความใดเกี่ยวกับ threading และงาน CPU-bound ใน Python ที่ถูกต้อง?

Threads ใช้ CPU cores ทั้งหมดได้อย่างเต็มที่สำหรับ compute-intensive loops Threads ปล่อย GIL ระหว่าง pure Python computation การเพิ่ม threads ใน CPU-bound loop แทบไม่ได้ความเร็วเพิ่มขึ้นเพราะ GIL GIL ใช้กับ I/O operations เท่านั้น ไม่ใช่การคำนวณ

สำหรับ 5 ID ใช้เวลาประมาณเท่าไหร่ และเพราะอะไร?

async def fetch_user(uid):
    time.sleep(0.2)
    return {"id": uid}

async def fetch_all(ids):
    return await asyncio.gather(*[fetch_user(i) for i in ids])

~0.2s -- gather รันทุกอย่างแบบ concurrent ตามที่ตั้งใจ ~1.0s (5 x 0.2s) -- time.sleep block event loop ดังนั้น gather ไม่ให้ concurrency จริงตรงนี้ ~0.04s -- asyncio ขนานกันข้าม CPU core raise RuntimeError เพราะ time.sleep ไม่ถูก await

รันตรงๆ ด้วย python3.14 script.py ธรรมดา ไม่มีอะไรถูกพิมพ์ อะไรเปลี่ยนไปถ้ารันด้วย python3.14 -X dev script.py แทน?

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

coro = greet("Python")
del coro

สคริปต์รันเร็วขึ้นโดยข้าม bytecode verification เผยให้เห็น RuntimeWarning: coroutine 'greet' was never awaited ตอนที่ coroutine ถูก garbage-collect ซึ่งการรันปกติจะเงียบไป raise SyntaxError เพราะ coroutine ต้องมี await เรียก asyncio.run ให้อัตโนมัติกับ coroutine ที่ไม่ได้ใช้ทุกตัวที่เจอ

workload หนึ่งใช้เวลา wall-clock 95% รอ network response จาก async-native HTTP client เครื่องมือใดจากตารางตัดสินใจเหมาะที่สุด?

multiprocessing เพราะ parallelism ช่วยเสมอ asyncio เพราะงานเป็น I/O-bound และมี async-native library ให้ใช้ Free-threaded python3.14t เพราะมันเอา GIL ออก threading เพราะเป็นตัวเลือกที่ง่ายที่สุด

Submit

Mark as complete