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

این کد را ببینید و حدس بزنید چقدر طول می‌کشد. دو کوروتین داریم که هر کدام باید ۲ ثانیه «منتظر» بمانند، پس انتظار داریم برنامه در مجموع حدود ۲ ثانیه تمام شود؛ اما در عمل حدود ۴ ثانیه طول می‌کشد. چرا؟ چون یکی از آن‌ها به جای await، از یک تابع Blocking استفاده کرده و همین موضوع ما را مستقیم به سراغ متد run_in_executor می‌برد.

import asyncio
import time

def blocking_io(name, delay):
    print(f'{name} started')
    time.sleep(delay)  # blocks the whole thread, not just this coroutine
    print(f'{name} finished')

async def async_task(name, delay):
    print(f'{name} started')
    await asyncio.sleep(delay)  # yields control back to the event loop
    print(f'{name} finished')

async def bad_wrapper(name, delay):
    blocking_io(name, delay)  # no await -> event loop is frozen here

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

asyncio.run(main())

چرا حلقهٔ رویداد یخ می‌زند؟

در درس Task و create_task دیدیم که همزمانی در asyncio روی یک تک‌ترد و با «همکاری» کوروتین‌ها اتفاق می‌افتد: هر کوروتین در نقطهٔ await کنترل را داوطلبانه پس می‌دهد تا بقیه پیش بروند. حالا مشکل واضح می‌شود؛ time.sleep یک تابع سینکرون است و هیچ await ندارد. وقتی اجرا می‌شود، تنها ترد برنامه را کاملاً در اختیار می‌گیرد و حلقهٔ رویداد حتی نمی‌تواند async_task را جلو ببرد. نتیجه؟ همزمانی از بین می‌رود و دو کار پشت سر هم اجرا می‌شوند. متد run_in_executor دقیقاً پلی است بین دنیای asyncio و دنیای Threadها تا همین کد Blocking را از روی ترد اصلی برداریم.

راه‌حل: انتقال کار مسدودکننده با run_in_executor

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

async def good_wrapper(name, delay):
    loop = asyncio.get_running_loop()
    # runs blocking_io in a worker thread; the loop stays free meanwhile
    await loop.run_in_executor(None, blocking_io, name, delay)

حالا اگر همان main را با good_wrapper اجرا کنید، کل برنامه حدود ۲ ثانیه طول می‌کشد، نه ۴ ثانیه. چون blocking_io روی یک ترد کارگر می‌رود و ترد اصلی آزاد می‌ماند تا async_task را همزمان پیش ببرد. توجه کنید که خود تابع Blocking را دست‌نخورده گذاشتیم؛ فقط محل اجرای آن را عوض کردیم.

مثال واقعی: یک وب‌سرور async که تصویر پردازش می‌کند

تصور کنید یک سرویس async دارید که هزاران درخواست شبکه را همزمان مدیریت می‌کند، اما گاهی باید یک تصویر را با کتابخانه‌ای مثل Pillow تغییر اندازه دهد؛ کتابخانه‌ای که کاملاً سینکرون است. اگر آن را مستقیم صدا بزنید، در همان لحظه پاسخ‌گویی به همهٔ کاربران دیگر متوقف می‌شود. راه درست، سپردن این کار به یک executor است:

async def handle_upload(image_bytes):
    loop = asyncio.get_running_loop()
    # heavy, synchronous work goes to a thread; other requests keep flowing
    thumbnail = await loop.run_in_executor(None, resize_image, image_bytes)
    return thumbnail

تلهٔ رایج: آرگومان کلیدی و راه‌حل functools.partial

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

from functools import partial

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

Thread یا Process؟ نقش GIL

انتخاب executor به نوع کار بستگی دارد. برای کارهای I/O-bound مثل خواندن فایل یا درخواست شبکه، همان ThreadPoolExecutor پیش‌فرض عالی است، چون این عملیات‌ها هنگام انتظار GIL را آزاد می‌کنند. اما برای کارهای CPU-bound مثل محاسبات سنگین، تردها به‌خاطر GIL نمی‌توانند واقعاً موازی شوند؛ اینجا باید سراغ ProcessPoolExecutor بروید تا کار در یک پردازش کاملاً مجزا انجام شود:

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy_task(n):
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:  # separate process, real parallelism
        result = await loop.run_in_executor(pool, cpu_heavy_task, 10_000_000)
    print(result)

asyncio.run(main())

یک هشدار مهم: تابعی که به ProcessPoolExecutor می‌سپارید باید قابل pickle باشد (مثلاً یک تابع سطح ماژول، نه یک lambda یا تابع تودرتو)، وگرنه با خطا مواجه می‌شوید؛ چون آرگومان‌ها و تابع باید برای انتقال به پردازش دیگر سریالایز شوند.

جمع‌بندی

• هر تابع Blocking که مستقیم درون کوروتین صدا زده شود، کل حلقهٔ رویداد را یخ می‌زند و همزمانی asyncio را نابود می‌کند.
• متد run_in_executor آن کار را به یک ترد یا پردازش جدا منتقل می‌کند و یک awaitable می‌دهد تا حلقه آزاد بماند.
• برای I/O-bound از ThreadPoolExecutor (یا None) و برای CPU-bound از ProcessPoolExecutor استفاده کنید؛ GIL تعیین‌کننده است.
• برای آرگومان کلیدی از functools.partial کمک بگیرید و مراقب pickle‌پذیری در حالت پردازش باشید.