تفاوت عمیق بین Pool و ProcessPoolExecutor؛ راهنمای انتخاب برای پردازش موازی

در پایتون، دو ابزار اصلی برای اجرای موازی کدهای CPU-bound وجود دارد: multiprocessing.Pool و concurrent.futures.ProcessPoolExecutor. هر دو کار یکسانی انجام می‌دهند: یک مجموعه پردازش (Pool of Processes) ایجاد می‌کنند و وظایف را بین آنها توزیع می‌کنند تا از هسته‌های چندگانه‌ی پردازنده استفاده کنند و محدودیت GIL را دور بزنند. اما با وجود شباهت‌های زیاد، تفاوت‌های عمیقی در طراحی، API و کاربردهای خاص دارند. در این مقاله، این دو ابزار را موشکافانه بررسی می‌کنیم و به شما کمک می‌کنیم تصمیم بگیرید که در چه شرایطی از هرکدام استفاده کنید.

شباهت‌های بنیادین

قبل از پرداختن به تفاوت‌ها، بهتر است بدانیم که این دو ابزار در چه مواردی کاملاً شبیه هستند. اولاً، هر دو از زیرساخت یکسانی استفاده می‌کنند. ProcessPoolExecutor در واقع یک لایه‌ی انتزاعی (Wrapper) روی multiprocessing.Pool است و از همان مکانیزم‌های پایه‌ای برای ایجاد و مدیریت پردازش‌ها استفاده می‌کند. در نتیجه، هر دو سربار (Overhead) راه‌اندازی یکسانی دارند و تفاوت عملکردی قابل‌توجهی بین آنها وجود ندارد. انتخاب بین این دو، بیشتر یک تصمیم معماری و سلیقه‌ای است تا عملکردی. هر دو ابزار نیز از مفهوم chunksize برای بهینه‌سازی ارتباطات بین‌پردازه‌ای (IPC) پشتیبانی می‌کنند، که در ادامه به آن خواهیم پرداخت.

تفاوت کلیدی: Future در برابر نتیجه‌ی مستقیم

اصلی‌ترین تفاوت این دو، در نحوه‌ی بازگرداندن نتایج است. این تفاوت، بر تمام جنبه‌های دیگر کار با آنها تأثیر می‌گذارد.

multiprocessing.Pool نتایج را به‌صورت مستقیم و هم‌زمان (Blocking) برمی‌گرداند. یعنی وقتی تابع pool.map را صدا می‌زنید، برنامه تا زمانی که همه‌ی نتایج آماده نشوند، متوقف می‌شود و سپس یک لیست از نتایج را دریافت می‌کند.

from multiprocessing import Pool

def square(x):
    return x ** 2

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        results = pool.map(square, [1, 2, 3, 4, 5])
        print(results)  # [1, 4, 9, 16, 25]

from multiprocessing import Pool

def square(x):
    return x ** 2

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        results = pool.map(square, [1, 2, 3, 4, 5])
        print(results)  # [1, 4, 9, 16, 25]

اما concurrent.futures.ProcessPoolExecutor یک Future برمی‌گرداند. Future شبیه به یک قول (Promise) است که در آینده نتیجه را در اختیار شما قرار می‌دهد. این تفاوت باعث می‌شود کنترل بسیار بیشتری روی وظایف داشته باشید.

from concurrent.futures import ProcessPoolExecutor

def square(x):
    return x ** 2

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        future = executor.submit(square, 10)
        result = future.result()  # اینجا منتظر می‌شود تا نتیجه برسد
        print(result)

from concurrent.futures import ProcessPoolExecutor

def square(x):
    return x ** 2

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        future = executor.submit(square, 10)
        result = future.result()  # اینجا منتظر می‌شود تا نتیجه برسد
        print(result)

قابلیت‌های پیشرفته با Future

همین تفاوت بنیادین، مجموعه‌ای از قابلیت‌ها را در ProcessPoolExecutor فعال می‌کند که در Pool وجود ندارند.

لغو وظایف (Cancellation): با Future می‌توانید یک وظیفه را قبل از شروع، لغو کنید. این قابلیت در سناریوهایی که ممکن است یک وظیفه دیگر سریع‌تر پاسخ دهد یا کاربر آن را لغو کند، بسیار ارزشمند است.

from concurrent.futures import ProcessPoolExecutor
import time

def slow_task(n):
    time.sleep(n)
    return n ** 2

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        future = executor.submit(slow_task, 10)
        time.sleep(0.1)
        if future.cancel():
            print("Task cancelled successfully")
        else:
            print("Task already running; cannot cancel")

from concurrent.futures import ProcessPoolExecutor
import time

def slow_task(n):
    time.sleep(n)
    return n ** 2

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        future = executor.submit(slow_task, 10)
        time.sleep(0.1)
        if future.cancel():
            print("Task cancelled successfully")
        else:
            print("Task already running; cannot cancel")

مدیریت زمان انتظار (Timeout): ProcessPoolExecutor امکان تعیین زمان انتظار برای هر Future را فراهم می‌کند. اگر وظیفه در آن زمان به پایان نرسد، یک استثنای TimeoutError پرتاب می‌شود.

from concurrent.futures import ProcessPoolExecutor, TimeoutError
import time

def task(duration):
    time.sleep(duration)
    return "Done"

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        future = executor.submit(task, 10)
        try:
            result = future.result(timeout=2)
        except TimeoutError:
            print("Task exceeded 2-second timeout")
            future.cancel()

from concurrent.futures import ProcessPoolExecutor, TimeoutError
import time

def task(duration):
    time.sleep(duration)
    return "Done"

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        future = executor.submit(task, 10)
        try:
            result = future.result(timeout=2)
        except TimeoutError:
            print("Task exceeded 2-second timeout")
            future.cancel()

دریافت نتایج به محض آماده شدن (as_completed): یکی از قدرتمندترین قابلیت‌های ProcessPoolExecutor، تابع as_completed است. این تابع به شما اجازه می‌دهد نتایج را به محض آماده شدن هر وظیفه، بدون توجه به ترتیب ارسال، دریافت کنید. این ویژگی برای پردازش‌های لوله‌ای (Pipeline) بسیار مفید است.

from concurrent.futures import ProcessPoolExecutor, as_completed
import time

def task(x):
    time.sleep(x)
    return x ** 2

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(task, x): x for x in [3, 1, 4, 1, 5]}
        for future in as_completed(futures):
            x = futures[future]
            result = future.result()
            print(f"Task {x} completed with result: {result}")

from concurrent.futures import ProcessPoolExecutor, as_completed
import time

def task(x):
    time.sleep(x)
    return x ** 2

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(task, x): x for x in [3, 1, 4, 1, 5]}
        for future in as_completed(futures):
            x = futures[future]
            result = future.result()
            print(f"Task {x} completed with result: {result}")

با Pool، برای دریافت نتایج به‌محض آماده شدن باید از روش‌های پیچیده‌تری مثل apply_async با callback استفاده کنید.

تفاوت‌های API و سهولت استفاده

ارسال وظایف با چند آرگومان: ارسال وظایف با چندین آرگومان در ProcessPoolExecutor ساده‌تر است. تابع map این ابزار، امکان پذیرفتن چندین iterable را دارد.

# ProcessPoolExecutor: ارسال چند آرگومان به صورت مستقیم
def add(a, b):
    return a + b

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(add, [1, 2, 3], [4, 5, 6]))
        print(results)  # [5, 7, 9]

# ProcessPoolExecutor: ارسال چند آرگومان به صورت مستقیم
def add(a, b):
    return a + b

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(add, [1, 2, 3], [4, 5, 6]))
        print(results)  # [5, 7, 9]

اما multiprocessing.Pool برای این کار نیاز به استفاده از starmap دارد و آرگومان‌ها باید به‌صورت زیپ‌شده (Zipped) به آن پاس داده شوند.

بهینه‌سازی chunksize: هر دو ابزار از پارامتر chunksize برای کاهش سربار ارتباط بین پردازش‌ها استفاده می‌کنند. اما multiprocessing.Pool در روش map، اگر chunksize مشخص نشود، به‌طور خودکار یک مقدار مناسب بر اساس اندازه‌ی ورودی و تعداد پردازش‌ها محاسبه می‌کند. در مقابل، ProcessPoolExecutor به‌طور پیش‌فرض از chunksize=1 استفاده می‌کند، یعنی هر وظیفه را به‌صورت جداگانه به پردازش‌ها می‌فرستد. برای تعداد زیادی وظیفه‌ی کوچک، تنظیم دستی chunksize در ProcessPoolExecutor ضروری است تا عملکرد خوبی داشته باشد، در غیر این صورت، Pool ممکن است سریع‌تر عمل کند.

یکپارچگی با asyncio: ProcessPoolExecutor به‌صورت طبیعی با asyncio ادغام می‌شود. می‌توانید از loop.run_in_executor برای اجرای وظایف سنگین CPU-bound در یک پردازش جداگانه، بدون مسدود کردن event loop استفاده کنید. این کار با Pool به‌سختی امکان‌پذیر است و نیاز به نوشتن کدهای اضافی دارد.

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

از concurrent.futures.ProcessPoolExecutor استفاده کنید اگر:

• نیاز به لغو وظایف قبل از اجرا دارید
• می‌خواهید به‌ازای هر وظیفه، یک Timeout مجزا تعیین کنید
• کد شما با asyncio ادغام شده است و نیاز به اجرای وظایف سنگین در پس‌زمینه دارید
• می‌خواهید نتایج را به محض آماده شدن، با as_completed پردازش کنید
• در حال شروع یک پروژه‌ی جدید هستید و API مدرن‌تر و Future-based را ترجیح می‌دهید

از multiprocessing.Pool استفاده کنید اگر:

• کد قدیمی (Legacy) دارید که از قبل از Pool استفاده می‌کند و تغییر آن هزینه‌بر است
• به یک initializer یا maxtasksperchild برای تنظیم اولیه‌ی پردازش‌ها یا محدود کردن تعداد وظایف هر پردازش نیاز دارید (این قابلیت‌ها در ProcessPoolExecutor وجود ندارند)
• پردازش دسته‌ای (Batch Processing) ساده‌ای دارید و نیازی به قابلیت‌های پیشرفته‌ی Future ندارید
• با تعداد بسیار زیادی وظیفه‌ی کوچک کار می‌کنید و می‌خواهید از محاسبه‌ی خودکار chunksize توسط Pool بهره‌مند شوید

توجه: در بیشتر موارد، اگر ProcessPoolExecutor نیازهای شما را برآورده می‌کند، انتخاب بهتری است. این ابزار در آینده احتمالاً توسعه و بهبود بیشتری خواهد یافت، زیرا هدف توسعه‌دهندگان هسته‌ی پایتون، تثبیت APIهای concurrent.futures است.

جمع‌بندی

هر دو ابزار multiprocessing.Pool و concurrent.futures.ProcessPoolExecutor برای پردازش موازی در پایتون قدرتمند هستند. انتخاب بین آنها به نیازهای خاص شما بستگی دارد. اگر به کنترل دقیق روی وظایف، قابلیت لغو، Timeout و یکپارچگی با asyncio نیاز دارید، ProcessPoolExecutor گزینه‌ی برتر است. اما اگر با یک پایگاه کد قدیمی کار می‌کنید یا به سادگی و سرعت در پردازش دسته‌ای نیاز دارید، multiprocessing.Pool همچنان یک ابزار معتبر و کارآمد است.