ข้ามไปยังเนื้อหา
Python Deep Dive

Multiprocessing

ทำไมต้องใช้ processes ไม่ใช่ threads?

หัวข้อที่มีชื่อว่า “ทำไมต้องใช้ processes ไม่ใช่ threads?”

ดังที่กล่าวไว้ในบทเรียน GIL, Global Interpreter Lock ของ CPython ป้องกัน threads หลายตัวจากการรัน Python bytecode แบบ parallel สำหรับงาน CPU-bound เช่น การคำนวณเชิงตัวเลข การประมวลผลภาพ การ parsing ข้อมูล — threads ไม่ให้ความเร็วเพิ่มขึ้น

multiprocessing แก้ปัญหานี้โดยสร้าง separate OS processes แต่ละ process มี Python interpreter, GIL และ memory space ของตัวเอง True parallelism จึงเป็นไปได้: เครื่องที่มี 8 cores สามารถรัน Python processes ได้ 8 ตัวพร้อมกัน

guard if __name__ == "__main__"

หัวข้อที่มีชื่อว่า “guard if __name__ == "__main__"”

บน macOS และ Windows, start method เริ่มต้นคือ spawn — child process import module ของ parent ใหม่ตั้งแต่ต้น หากไม่มี guard, โค้ดระดับ module (รวมถึงการเรียก Pool(...)) จะรันใหม่ใน child ทุกตัว ทำให้เกิด loop การสร้าง process ไม่สิ้นสุด

from multiprocessing import Pool


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


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

ตั้งแต่ Python 3.14 ระบบ Unix ที่ไม่ใช่ macOS จะใช้ forkserver เป็นค่าเริ่มต้นแทน fork (ซึ่งเป็นค่าเริ่มต้นในช่วง 3.3–3.13) — ดูรายละเอียดใน 3.14 What’s New หัวข้อ multiprocessing และ gh-84559 ส่วน macOS และ Windows ยังคงใช้ spawn เป็นค่าเริ่มต้น เนื่องจากทุกแพลตฟอร์มสามารถรัน module ของคุณใหม่ใน process สะอาดตอน pool เริ่มทำงานได้ จึงควรใส่ guard ไว้เสมอทุกแพลตฟอร์ม ไม่ใช่แค่ macOS/Windows

Pool.map — parallel map บน iterable

หัวข้อที่มีชื่อว่า “Pool.map — parallel map บน iterable”

Pool.map(fn, iterable) กระจาย items ไปยัง workers และคืนผลลัพธ์ตามลำดับ input และบล็อกจนกว่า items ทั้งหมดจะถูกประมวลผล

from multiprocessing import Pool
import math


def is_prime(n: int) -> bool:
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True


if __name__ == "__main__":
    candidates = list(range(1, 1_000_001))
    with Pool() as pool:              # Pool() defaults to cpu_count() workers
        flags = pool.map(is_prime, candidates)
    primes = [n for n, flag in zip(candidates, flags) if flag]
    print(f"Found {len(primes)} primes up to 1,000,000")

Pool.starmap — arguments หลายตัวต่อ call

หัวข้อที่มีชื่อว่า “Pool.starmap — arguments หลายตัวต่อ call”

เมื่อ worker function รับ argument มากกว่าหนึ่งตัว ใช้ starmap พร้อม iterable ของ tuples:

from multiprocessing import Pool


def power(base: int, exp: int) -> int:
    return base ** exp


if __name__ == "__main__":
    args = [(2, 10), (3, 5), (5, 4), (7, 3)]
    with Pool(processes=4) as pool:
        results = pool.starmap(power, args)
    print(results)
    # [1024, 243, 625, 343]

ProcessPoolExecutor — modern high-level API

หัวข้อที่มีชื่อว่า “ProcessPoolExecutor — modern high-level API”

concurrent.futures.ProcessPoolExecutor ให้ interface เดียวกับ ThreadPoolExecutor ทำให้ง่ายต่อการสลับระหว่าง thread-based และ process-based parallelism

from concurrent.futures import ProcessPoolExecutor
import math


def cpu_work(n: int) -> float:
    # Simulate CPU-intensive computation
    return sum(math.sin(i) for i in range(n))


if __name__ == "__main__":
    inputs = [3_000_000] * 8   # 8 CPU-bound tasks, large enough that compute dominates pool startup


    # Sequential baseline
    import time
    start = time.perf_counter()
    seq_results = [cpu_work(n) for n in inputs]
    seq_time = time.perf_counter() - start


    # Parallel with ProcessPoolExecutor
    start = time.perf_counter()
    with ProcessPoolExecutor() as executor:
        par_results = list(executor.map(cpu_work, inputs))
    par_time = time.perf_counter() - start


    print(f"Sequential: {seq_time:.2f}s")
    print(f"Parallel:   {par_time:.2f}s")
    print(f"Speedup:    {seq_time / par_time:.1f}x")

วัดจริงสามรอบบนเครื่อง 11-core (python3.14) ได้ 3.0x, 3.3x, 3.6x — ห่างไกลจาก 8 เท่าเต็มๆ และนั่นคือสิ่งที่คาดไว้ ไม่ใช่ความผิดพลาดของการวัด เวอร์ชันเดิมที่ใช้ 500_000 elements ต่องานเสียเวลาไปกับ pool startup มากจนความเร็ว แทบไม่เกิน 1x (ProcessPoolExecutor ต้อง spawn interpreter process ใหม่ทุกตัว) แม้จะเพิ่ม workload ขึ้นหกเท่าแล้ว ต้นทุน process startup, การ pickle arguments/results, และการ schedule ข้าม performance/efficiency core ที่ปนกัน ก็ยังกินส่วนต่างจาก speedup สูงสุดตามทฤษฎี ตัวเลขในที่นี้ขึ้นกับเครื่อง — วัดบนเครื่องของคุณเองก่อนอ้างตัวเลขที่แน่นอน

Picklability: ทำไม lambda ถึงทำให้ ProcessPoolExecutor พัง

หัวข้อที่มีชื่อว่า “Picklability: ทำไม lambda ถึงทำให้ ProcessPoolExecutor พัง”

argument ทุกตัว และ callable เอง ต้องเดินทางจาก parent process ไปยัง worker process เป็น bytes — ทั้ง multiprocessing และ ProcessPoolExecutor ใช้ pickle ทำแบบนี้ lambda หรือฟังก์ชันใดๆ ที่ประกาศในเครื่องภายในฟังก์ชันอื่น pickle ไม่ได้ เพราะการ pickle ฟังก์ชันเก็บ reference ถึงมันด้วยชื่อ (module และ qualified name) ให้ worker import กลับมาใหม่ — และชื่อของ lambda คือ <lambda> ตรงๆ ไม่ใช่สิ่งที่ import ได้:

from concurrent.futures import ProcessPoolExecutor


def main() -> None:
    with ProcessPoolExecutor(max_workers=2) as pool:
        fut = pool.submit(lambda x: x * x, 5)   # lambdas cannot be pickled
        try:
            print(fut.result())
        except Exception as e:
            print(f"{type(e).__name__}: {e}")


if __name__ == "__main__":
    main()
PicklingError: Can't pickle local object <function main.<locals>.<lambda> at 0x...>

ปัญหาเดียวกันเกิดกับ def ธรรมดาที่ประกาศซ้อนในฟังก์ชันอื่นด้วย — “local object” ใน error ครอบคลุมทั้งสองแบบ ทางแก้เหมือนกันเสมอ: ประกาศ callable ที่ระดับ module (หรือเป็น staticmethod บน class ที่ประกาศระดับ module) เพื่อให้ qualified name ของมัน resolve เป็นสิ่งที่ worker process import เองได้ functools.partial ที่ห่อฟังก์ชันระดับ module pickle ได้ จึงเป็นวิธีปกติในการส่ง argument คงที่เพิ่มเติมโดยไม่ต้องเขียน lambda

Inter-process communication

หัวข้อที่มีชื่อว่า “Inter-process communication”

Processes ไม่แชร์ memory เพื่อแลกเปลี่ยนข้อมูลระหว่างกัน ใช้ multiprocessing.Queue หรือ multiprocessing.Pipe

from multiprocessing import Process, Queue


def producer(q: Queue, items: list) -> None:
    for item in items:
        q.put(item)
    q.put(None)    # sentinel


def consumer(q: Queue) -> None:
    while True:
        item = q.get()
        if item is None:
            break
        print(f"consumed: {item}")


if __name__ == "__main__":
    q: Queue = Queue()
    p1 = Process(target=producer, args=(q, [1, 2, 3, 4, 5]))
    p2 = Process(target=consumer, args=(q,))
    p1.start(); p2.start()
    p1.join();  p2.join()

Shared memory ใน Python 3.8+

หัวข้อที่มีชื่อว่า “Shared memory ใน Python 3.8+”

multiprocessing.shared_memory.SharedMemory ช่วยให้ processes เข้าถึง memory block ที่แชร์กันโดยไม่มี serialization overhead — มีประโยชน์สำหรับ NumPy arrays ขนาดใหญ่

from multiprocessing import shared_memory
import numpy as np


# Create a shared memory block
shm = shared_memory.SharedMemory(create=True, size=400)   # 100 float32 values
arr = np.ndarray((100,), dtype=np.float32, buffer=shm.buf)
arr[:] = np.arange(100, dtype=np.float32)
print(arr[:5])    # [0. 1. 2. 3. 4.]
shm.close()
shm.unlink()

Subinterpreters: concurrent.interpreters (Python 3.14)

หัวข้อที่มีชื่อว่า “Subinterpreters: concurrent.interpreters (Python 3.14)”

PEP 734 เปิด subinterpreter ให้ code Python ใช้ได้ แต่ละ interpreter มี GIL ของตัวเอง — พื้นฐานส่วนนี้มาตั้งแต่ 3.12 — จึงรันบน core แยกกันได้เหมือน process แต่อยู่ใน OS process เดียวกัน ทำให้ startup ถูกกว่า fork หรือ spawn มาก

from concurrent import interpreters


interp = interpreters.create()
interp.exec("print(sum(range(1_000_000)))")
interp.close()

และมี executor คู่กันสำหรับงานแบบ pool:

from concurrent.futures import InterpreterPoolExecutor


with InterpreterPoolExecutor(max_workers=4) as pool:
    print(list(pool.map(str, range(4))))

สิ่งที่ต้องแลกคือขอบเขต: interpreter ไม่ได้ใช้ object ร่วมกัน argument และผลลัพธ์จึงต้องข้ามไปในรูป data ที่ serialise ได้ เหมือน multiprocessing เป๊ะ ๆ สิ่งที่ประหยัดได้คือเวลา startup ของ process และหน่วยความจำ

Processes vs threads: คู่มือการตัดสินใจ

หัวข้อที่มีชื่อว่า “Processes vs threads: คู่มือการตัดสินใจ”
เกณฑ์multiprocessingthreadingasyncio
งาน CPU-boundดีที่สุดไม่มีประโยชน์ (GIL)เครื่องมือผิด
I/O-bound, blocking libsเกินความจำเป็นดีrun_in_executor
I/O-bound, async libsเกินความจำเป็นได้ดีที่สุด
Memory overheadสูง (คัดลอกทั้งหมด)ต่ำ (แชร์)น้อยมาก
CommunicationQueue / PipeShared state + LockChannels / queues
Startup timeช้า (fork/spawn)เร็วศูนย์

Mental Model

หัวข้อที่มีชื่อว่า “Mental Model”

Pool เหมือนโรงงานเล็กๆ ที่มีอาคารแยกของตัวเอง (processes) ไม่ใช่โต๊ะทำงานเพิ่มในอาคารเดียวกัน (threads) ไม่มีอะไรเดินข้ามอาคารได้เว้นแต่สิ่งที่คุณตั้งใจส่งทางไปรษณีย์ — pickle คือบริการไปรษณีย์ และมันส่งได้แค่สิ่งที่มีที่อยู่ส่งกลับที่อาคารอื่นค้นหาได้ (ชื่อระดับ module) lambda ไม่มีที่อยู่แบบนั้น; Queue หรือ Pipe คือวิธีที่คุณส่งข้อมูลไปมาจริงๆ เมื่ออาคารเปิดทำงานแล้ว

flowchart LR
P["Parent process"] -->|"pickle(fn, args)"| W1["Worker process 1"]
P -->|"pickle(fn, args)"| W2["Worker process 2"]
W1 -->|"pickle(result)"| P
W2 -->|"pickle(result)"| P
L["lambda / local def"] -.->|"PicklingError — no importable name"| P
argument, callable และผลลัพธ์ทุกตัวข้าม process boundary เป็น pickled bytes

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

หัวข้อที่มีชื่อว่า “ข้อแลกเปลี่ยน”
สิ่งที่ได้ประโยชน์ต้นทุน
Process แยกต่อ workerparallelism จริง, GIL ของตัวเอง, worker ตัวหนึ่งพังไม่กระทบตัวอื่นต้นทุน startup interpreter เต็มรูปแบบต่อ process; ไม่แชร์ memory โดยค่าเริ่มต้น
ส่ง argument ด้วย pickleใช้ได้โปร่งใสกับข้อมูลส่วนใหญ่callable ต้อง import ด้วยชื่อได้ — ไม่มี lambda, ไม่มี local closure
shared_memoryข้าม serialization สำหรับ array ขนาดใหญ่จัดการ lifecycle เอง (close()/unlink()); ไม่มี safety net อัตโนมัติ
Subinterpreters (3.14)startup ถูกกว่า process เต็มรูปแบบยังไม่แชร์ object ข้าม boundary — ต้นทุน serialization เท่ากับข้อมูลที่ข้ามไป
forkserver default (Unix ที่ไม่ใช่ macOS, 3.14+)คาดเดาได้มากกว่า fork รอบ threads/locks ที่ถืออยู่ตอน forkยังต้องมี guard __main__ ต่างจาก fork แบบเก่า

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

หัวข้อที่มีชื่อว่า “ความเข้าใจผิดที่พบบ่อย”
  • “multiprocessing แชร์ memory เหมือน threading” ไม่จริงโดยค่าเริ่มต้น — argument และผลลัพธ์ทุกตัวถูก pickle แล้วส่งผ่าน pipe shared_memory และ Manager มีอยู่ก็เพราะ multiprocessing ธรรมดาไม่แชร์ state
  • “ฟังก์ชันไหนก็ใช้กับ Pool.map หรือ ProcessPoolExecutor.submit ได้” มีแค่ฟังก์ชันที่ worker process import ด้วยชื่อได้ เท่านั้นที่ใช้ได้ — ฟังก์ชันระดับ module และ callable ที่ pickle ได้ lambda หรือฟังก์ชันที่ประกาศในฟังก์ชันอื่น fail ทันทีด้วย PicklingError ไม่มีอะไรเกี่ยวกับตรรกะของ lambda ที่เป็นปัญหา มันแค่ไม่มีชื่อที่ import ได้ การส่ง argument เพิ่มให้ฟังก์ชันระดับ module คือเหตุผลปกติที่คนหันไปใช้ lambda ตรงนี้ — functools.partial ทำงานเดียวกันและ pickle ได้
  • “guard if __name__ สำคัญแค่บน Windows” ตั้งแต่ Python 3.14 Unix ที่ไม่ใช่ macOS ก็ default เป็น forkserver ซึ่ง import module ใหม่เช่นกัน — guard จำเป็นบนทุกแพลตฟอร์มแล้ว ไม่ใช่แค่ Windows/macOS
  • “process ยิ่งเยอะยิ่งเร็ว” เกินจำนวน CPU core, process เพิ่มเติมเพิ่ม overhead ของ scheduling และ memory โดยไม่เพิ่มความสามารถคำนวณแบบ parallel; สำหรับ workload เล็กหรือ I/O หนัก ต้นทุน process startup อย่างเดียวก็ลบล้างประโยชน์ทั้งหมดได้ — ดู trade-off ของ pool startup กับ workload ที่วัดไว้ด้านบน

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

หัวข้อที่มีชื่อว่า “อ่านโค้ดที่ AI เขียน”

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

  • การสร้าง Pool(...), Process(...), หรือ ProcessPoolExecutor(...) เกิดที่ระดับ module นอก guard if __name__ == "__main__": หรือไม่? บนแพลตฟอร์ม spawn/forkserver สิ่งนี้ import module ใหม่ในทุก worker
  • มี lambda หรือฟังก์ชันที่ประกาศในฟังก์ชันอื่น ถูกส่งเข้า pool.map, pool.submit, หรือเป็น Process target หรือไม่? มันจะ fail ด้วย PicklingError ทันทีที่รันจริงใน worker — ไม่ใช่ตอนที่เขียน
  • workload เป็น CPU-bound จริงหรือไม่? การใช้ multiprocessing กับงาน I/O-bound เพิ่ม overhead ของ serialization และ process-startup โดยไม่ได้ประโยชน์ parallelism ที่ asyncio/threading ให้ได้ถูกกว่าอยู่แล้ว
  • โค้ดสมมติว่า child process แชร์ mutable state ระดับ module กับ parent (cache ที่แชร์, counter) โดยไม่มี Manager, shared_memory, หรือ Queue หรือไม่? แต่ละ process มีสำเนาของตัวเอง

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

from concurrent.futures import ProcessPoolExecutor


def main() -> None:
  multiplier = 3
  with ProcessPoolExecutor() as pool:
      results = list(pool.map(lambda x: x * multiplier, range(5)))
  print(results)


if __name__ == "__main__":
  main()
เฉลย

โค้ดนี้มี guard if __name__ == "__main__": ถูกต้อง ซึ่งแก้ bug multiprocessing ที่สอนกันบ่อยที่สุด — เลยง่ายที่จะคิดว่าที่เหลือถูกต้องแล้ว แต่ lambda ที่ closure ทับ multiplier จะ fail ทันทีด้วย PicklingError: Can't pickle local object <function main.<locals>.<lambda> at 0x...> ตอนที่ pool.map พยายามส่งมันไปยัง worker process ไม่ว่าจะมี guard หรือไม่ ทางแก้คือฟังก์ชันระดับ module บวก functools.partial เพื่อ bind multiplier: from functools import partial แล้ว pool.map(partial(multiply, multiplier), range(5)) โดยมี def multiply(m: int, x: int) -> int: return x * m ประกาศที่ module scope

เหตุใด multiprocessing จึงได้ true CPU parallelism แต่ threading ไม่ได้ใน CPython?
โค้ดนี้ไม่มี guard if __name__ == '__main__': เกิดอะไรขึ้นเมื่อรันบน macOS หรือ Windows?
from multiprocessing import Pool

def square(x):
    return x * x

with Pool(processes=4) as pool:
    results = pool.map(square, range(10))
print(results)
method ใดกระจาย iterable ไปยัง process pool และคืนผลลัพธ์ตามลำดับ input?
ควรใช้กลไกใดเพื่อแชร์ NumPy arrays ขนาดใหญ่ระหว่าง processes โดยไม่มี serialization overhead?
เกิดอะไรขึ้นเมื่อรันโค้ดนี้?
from concurrent.futures import ProcessPoolExecutor

def main():
    with ProcessPoolExecutor() as pool:
        fut = pool.submit(lambda x: x * x, 5)
        print(fut.result())

if __name__ == "__main__":
    main()
benchmark CPU-bound ที่ใช้ 8 ProcessPoolExecutor workers บนเครื่อง 11-core วัด speedup ได้ 3.3x ไม่ใช่ 8x คำอธิบายที่ซื่อสัตย์ที่สุดคืออะไร?