معرفی کتابخانه multiprocessing در پایتون

کتابخانه multiprocessing در پایتون ابزاری قدرتمند برای اجرای همزمان چندین فرایند (process) روی CPU است. این کتابخانه راهی برای عبور از محدودیت GIL در پایتون و بهره‌گیری از چند هسته پردازنده فراهم می‌کند. در این مقاله مفاهیم پایه، اجزا، مثال‌های عملی و نکات بهینه‌سازی را به زبان ساده و کاربردی توضیح می‌دهیم.

چرا از multiprocessing استفاده کنیم؟

• پردازش‌های CPU-bound (محاسبات سنگین) با فرآیندها سرعت می‌گیرند چون GIL در فرآیندهای جداگانه وجود ندارد.
• جدا کردن کارها به فرایندهای مستقل باعث افزایش پایداری می‌شود (crash ایزوله می‌شود).
• امکان استفاده از منابع جداگانه حافظه برای هر فرایند.

مفاهیم کلیدی

• Process: معادل threading.Thread اما به‌صورت فرآیند سیستم‌عامل.
• Pool: مدیریت مجموعه‌ای از فرآیندها و تسهیم کارها بین آن‌ها.
• Queue / Pipe: مکانیزم‌های ارتباطی بین فرایندها.
• Manager: فراهم‌کننده اشیاء مشترک مانند dict و list بین فرایندها (پشتیبانی از proxy).
• shared_memory: از Python 3.8 امکان به‌اشتراک‌گذاری حافظه خام (مثلاً آرایه‌های numpy) بدون سریال‌سازی فراهم شد.

مثال ساده: Process

from multiprocessing import Process
import time

def worker(n):
    print(f"Worker {n} started")
    time.sleep(1)
    print(f"Worker {n} finished")

if __name__ == '__main__':
    procs = []
    for i in range(4):
        p = Process(target=worker, args=(i,))
        p.start()
        procs.append(p)
    for p in procs:
        p.join()

در این مثال چهار فرایند جداگانه ساخته و اجرا می‌شوند. استفاده از شرط if __name__ == '__main__' در ویندوز ضروری است تا از ایجاد بازگشتی (recursive spawn) جلوگیری شود. متد join() منتظر اتمام فرایند می‌ماند.

مثال کاربردی: Pool برای پردازش موازی داده‌ها

from multiprocessing import Pool
import math

def is_prime(n):
    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__':
    numbers = list(range(10_000, 10_500))
    with Pool(processes=4) as pool:
        results = pool.map(is_prime, numbers)
    primes = [n for n, prime in zip(numbers, results) if prime]
    print(len(primes))

در این مثال با Pool چهار فرایند ساخته شده و تابع is_prime به‌طور موازی روی مجموعه‌ای از اعداد اجرا می‌شود. pool.map جریان ساده‌ای برای تقسیم کار فراهم می‌کند.

بهینه‌سازی Pool: chunksize و apply_async

with Pool(processes=4) as pool:
    # تعیین chunksize برای کاهش سربار ارسالی بین پروسس‌ها
    results = pool.map(is_prime, numbers, chunksize=50)

افزایش مقدار chunksize می‌تواند برای توابع سریع و مجموعه‌های بزرگ بهبود کارایی ایجاد کند چون کاهش دفعات سریال‌سازی و ارسال داده بین پردازش‌ها رخ می‌دهد. برای کنترل دقیق‌تر نتایج غیرهمزمان می‌توان از apply_async استفاده کرد.

ارتباط بین فرایندها: Queue و Pipe

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(i)
    q.put(None)  # sentinel

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print("Consumed", item)

if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

در این الگو از یک sentinel (مانند None) برای خاتمه مصرف‌کننده استفاده شده است. Queue به‌صورت ایمن بین فرآیندها کار می‌کند اما هزینه pickling برای داده‌های بزرگ ممکن است قابل توجه باشد.

اشتراک‌گذاری داده‌های بزرگ: shared_memory و numpy

from multiprocessing import shared_memory, Process
import numpy as np

def worker(name, shape):
    existing = shared_memory.SharedMemory(name=name)
    arr = np.ndarray(shape, dtype=np.int64, buffer=existing.buf)
    arr += 1
    existing.close()

if __name__ == '__main__':
    a = np.zeros((1000,), dtype=np.int64)
    shm = shared_memory.SharedMemory(create=True, size=a.nbytes)
    b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
    b[:] = a[:]
    p = Process(target=worker, args=(shm.name, a.shape))
    p.start()
    p.join()
    print(b[0:5])
    shm.close()
    shm.unlink()

با shared_memory می‌توان از هزینه سریال‌سازی جلوگیری کرد و آرایه‌های numpy را به‌صورت مستقیم بین فرایندها به اشتراک گذاشت. پس از تمام شدن کار باید shared memory را آزاد کرد (unlink()).

مقایسه کوتاه اجزا

نکات عملی و تجربه‌ای

• همیشه اندازه داده‌های ارسالی بین فرایندها را در نظر بگیرید؛ pickling می‌تواند هزینه‌بر باشد.
• در ویندوز از spawn استفاده می‌شود؛ بنابراین کد باید محافظت شده با if __name__ == '__main__' باشد.
• برای مدل‌های بزرگ یا بارگذاری‌های گران‌قیمت، از initializer در Pool استفاده کنید تا هر فرایند یکبار کاری مثل بارگذاری مدل را انجام دهد.
• در حالتی که نیاز به latency پایین و اشتراک داده زیاد دارید، shared_memory یا mmap را مدنظر قرار دهید.
• برای پروفایل و اندازه‌گیری از ابزارهایی مثل time، perf و memory_profiler استفاده کنید تا گلوگاه‌ها مشخص شوند.

خلاصه و توصیه‌ها

کتابخانه multiprocessing یک راه‌حل بالغ و انعطاف‌پذیر برای اجرای موازی در پایتون است. برای کارهای محاسباتی سنگین از Process/Pool استفاده کنید، برای ارتباط ساده از Queue و برای اشتراک داده‌های بزرگ از shared_memory بهره ببرید. مراقب سربار سریال‌سازی باشید و با تست و پروفایل، تنظیمات مانند chunksize و تعداد فرایندها را بهینه کنید.

لطفاً از کمبود ها و مشکلات این محتوا برای ما بنویسید

ارسال