آموزش ماژول asyncio – متد run_in_executor

مقدمه

در درس‌های قبلی دیدیم که asyncio برای کارهای «I/O-bound» مثل درخواست‌های شبکه یا خواندن فایل بسیار کارآمد است، چون در نقاط await می‌تواند کنترل را به کوروتین‌های دیگر بدهد. اما همیشه با کتابخانه‌هایی سروکار نداریم که نسخهٔ async داشته باشند. خیلی از توابع پایتون – مثل عملیات سنگین محاسباتی یا توابع کتابخانه‌های قدیمی‌تر – به صورت «Blocking» (مسدودکننده) نوشته شده‌اند و هیچ نقطهٔ await ندارند. اگر چنین تابعی را مستقیماً درون یک کوروتین صدا بزنیم، کل حلقهٔ رویداد (Event Loop) را مسدود می‌کند. متد run_in_executor دقیقاً برای حل همین مشکل طراحی شده: پلی بین دنیای asyncio و دنیای Thread/Process.

مشکل: کد Blocking در دل asyncio

فرض کنید تابعی داریم که به صورت سینکرون (synchronous) کار می‌کند و از time.sleep برای شبیه‌سازی یک عملیات طولانی استفاده می‌کند:

import asyncio
import time

def blocking_io(name, delay):
    print(f'{name} شروع شد')
    time.sleep(delay)
    print(f'{name} تمام شد')

async def async_task(name, delay):
    print(f'{name} شروع شد')
    await asyncio.sleep(delay)
    print(f'{name} تمام شد')

async def bad_wrapper(name, delay):
    # این تابع مستقیماً صدا زده می‌شود و هیچ await ندارد
    blocking_io(name, delay)

async def main():
    delay = 2
    await asyncio.gather(
        async_task('Task A', delay),
        bad_wrapper('Task B', delay),
    )

asyncio.run(main())

چون blocking_io هیچ نقطهٔ await ندارد، وقتی اجرا می‌شود حلقهٔ رویداد نمی‌تواند بین آن و async_task سوییچ کند. در نتیجه کل برنامه عملاً سینکرون می‌شود و مزیت اصلی asyncio – یعنی همزمانی – از بین می‌رود.

راه‌حل: متد run_in_executor

متد loop.run_in_executor() به ما اجازه می‌دهد یک تابع Blocking را در یک Thread یا Process جدا اجرا کنیم، بدون اینکه حلقهٔ رویداد اصلی مسدود شود. امضای آن به این شکل است: loop.run_in_executor(executor, func, *args). اگر برای پارامتر executor مقدار None بدهیم، پایتون از یک ThreadPoolExecutor پیش‌فرض استفاده می‌کند.

import asyncio
import time

def blocking_io(name, delay):
    print(f'{name} شروع شد')
    time.sleep(delay)
    print(f'{name} تمام شد')

async def good_wrapper(name, delay):
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, blocking_io, name, delay)

async def async_task(name, delay):
    print(f'{name} شروع شد')
    await asyncio.sleep(delay)
    print(f'{name} تمام شد')

async def main():
    delay = 2
    await asyncio.gather(
        async_task('Task A', delay),
        good_wrapper('Task B', delay),
    )

asyncio.run(main())

اینجا blocking_io در یک Thread جدا اجرا می‌شود. خود await loop.run_in_executor(…) یک Future برمی‌گرداند که asyncio می‌تواند منتظرش بماند، در حالی که حلقهٔ رویداد آزاد است تا کوروتین‌های دیگر مثل async_task را هم پیش ببرد. به همین دلیل Task A و Task B این بار واقعاً به صورت همزمان اجرا می‌شوند.

پاس دادن آرگومان کلیدی با functools.partial

یک نکتهٔ مهم: run_in_executor فقط آرگومان‌های موضعی (positional) را می‌پذیرد و از آرگومان کلیدی (keyword) پشتیبانی نمی‌کند. اگر تابع ما پارامتر کلیدی دارد، باید از functools.partial استفاده کنیم:

import asyncio
import time
from functools import partial

def blocking_io(name, delay=1):
    print(f'{name} شروع شد')
    time.sleep(delay)
    print(f'{name} تمام شد')

async def main():
    loop = asyncio.get_running_loop()
    delay = 2
    func = partial(blocking_io, 'Task A', delay=delay)
    await loop.run_in_executor(None, func)

asyncio.run(main())

کارهای CPU-bound: استفاده از ProcessPoolExecutor

ThreadPoolExecutor پیش‌فرض برای کارهای I/O-bound (مثل time.sleep، خواندن فایل یا درخواست شبکه) مناسب است، چون این عملیات‌ها در زمان انتظار، GIL (Global Interpreter Lock) را آزاد می‌کنند. اما برای کارهای CPU-bound – مثل محاسبات سنگین ریاضی – Thread‌ها به دلیل GIL نمی‌توانند واقعاً به صورت موازی اجرا شوند. در این حالت باید از ProcessPoolExecutor استفاده کنیم تا کار در یک پردازش (Process) کاملاً مجزا انجام شود:

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy_task(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

async def main():
    loop = asyncio.get_running_loop()
    n = 10_000_000
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_task, n)
    print(f'نتیجه: {result}')

asyncio.run(main())

در این حالت محاسبهٔ سنگین در یک Process جدا اجرا می‌شود و در طول اجرای آن، حلقهٔ رویداد اصلی همچنان آزاد است تا کارهای دیگر (مثلاً پاسخ‌گویی به درخواست‌های شبکه) را پیش ببرد.

جمع‌بندی

با run_in_executor یاد گرفتیم چگونه می‌توان کد Blocking یا CPU-bound را با asyncio ترکیب کرد، بدون اینکه حلقهٔ رویداد مسدود شود: برای کارهای I/O-bound از ThreadPoolExecutor (یا مقدار پیش‌فرض None) و برای کارهای CPU-bound از ProcessPoolExecutor استفاده می‌کنیم. تا اینجا با ابزارهای threading، multiprocessing و asyncio به صورت جداگانه آشنا شدیم. در درس بعدی یک راهنمای جامع برای انتخاب درست بین این سه روش همزمانی ارائه می‌دهیم.