การซิงโครไนซ์และ Pooling ของกระบวนการใน Python

การซิงโครไนซ์ระหว่างกระบวนการ (Synchronization)

Multiprocessing รองรับการสร้างกระบวนการด้วย API ใช้ได้ทั้ง local และ remote concurrencies ใช้หลาย CPU บนเครื่องเดียว รันบน Windows และ UNIX

ทุก synchronization primitives เทียบเท่ากันอยู่ในแพ็กเกจนี้

ตัวอย่าง: ใช้ Lock ป้องกันการพิมพ์ทับกัน

from multiprocessing import Process, Lock

def my_function(lock, num):
    lock.acquire()
    print('hello world', num)
    lock.release()

if __name__ == '__main__':
    lock = Lock()
    for num in range(10):
        Process(target=my_function, args=(lock, num)).start()

Lock ใช้ให้แน่ใจว่ามีเพียงกระบวนการเดียวที่เข้าถึง standard output ได้ในเวลาเดียวกัน

Pooling (กลุ่มกระบวนการ)

ใช้คลาส Pool สร้าง pool ของ worker processes ที่ทำงานที่ส่งมาทั้งหมด

class multiprocessing.Pool([processes[, initializer[, initargs[, maxtasksperchild]]]])

Pool object ควบคุม pool ของ worker รองรับ asynchronous results พร้อม timeouts, callbacks และ parallel map implementation

เมธอดสำคัญของ Pool

เมธอด	คำอธิบาย
`apply(func, args, kwds)`	เหมือน built-in apply() บล็อกจนได้ผลลัพธ์
`apply_async(func, args, kwds, callback)`	รันแบบ asynchronous คืนค่า result object
`map(func, iterable, chunksize)`	เหมือน built-in map() แต่แบบขนาน บล็อกจนเสร็จ
`map_async(func, iterable, chunksize, callback)`	map แบบ asynchronous
`imap(func, iterable, chunksize)`	เหมือน itertools.imap() lazy evaluation
`imap_unordered(func, iterable, chunksize)`	เหมือน imap แต่ผลลัพธ์ไม่เรียงลำดับ
`close()`	ไม่รับ task ใหม่ worker จะออกเมื่อทำ task หมด
`terminate()`	หยุด worker ทันที ไม่รอจบ task
`join()`	รอ worker จบ ต้องเรียก close() หรือ terminate() ก่อน

คลาส AsyncResult

คืนค่าจาก apply_async() และ map_async()

get([timeout]) - คืนผลลัพธ์เมื่อพร้อม
wait([timeout]) - รอจนกว่าจะพร้อมหรือ timeout
ready() - ตรวจสอบว่าเสร็จแล้วหรือยัง
successful() - ตรวจสอบว่าเสร็จโดยไม่มี error

โค้ดตัวอย่าง Pool

from multiprocessing import Pool
import time

def myfunction(m):
    return m * m

if __name__ == '__main__':
    my_pool = Pool(processes=4)
    
    # Asynchronous single call
    result = my_pool.apply_async(myfunction, (10,))
    print(result.get(timeout=1))
    
    # Parallel map
    print(my_pool.map(myfunction, range(10)))
    
    # Iterator style
    my_it = my_pool.imap(myfunction, range(10))
    print(next(my_it))  # 0
    print(next(my_it))  # 1
    print(next(my_it))  # 4
    
    # Timeout example
    result = my_pool.apply_async(time.sleep, (10,))
    print(result.get(timeout=1))  # Raises TimeoutError
    
    my_pool.close()
    my_pool.join()

ข้อดีของ Pool: จัดการกระบวนการอัตโนมัติ เหมาะกับงานที่แบ่งได้เป็น chunks เล็กๆ นำไปประมวลผลขนาน