معرفی Trio و AnyIO؛ کتابخانه‌های مدرن جایگزین asyncio

اگر با asyncio کار کرده باشید، حتماً با پیچیدگی‌های آن آشنا هستید. مدیریت Cancellation، Task Group‌ها و شیلد کردن تسک‌ها، گاهی آنقدر دشوار می‌شود که حتی توسعه‌دهندگان حرفه‌ای را هم به چالش می‌کشد. اما خبر خوب این است که asyncio تنها گزینه نیست. در این مقاله، دو کتابخانه‌ی مدرن و قدرتمند را معرفی می‌کنیم که می‌توانند تجربه‌ی برنامه‌نویسی ناهمگام شما را متحول کنند: Trio و AnyIO.

چرا به دنبال جایگزین برای asyncio باشیم؟

asyncio کتابخانه‌ی استاندارد پایتون است و قدرت زیادی دارد. اما برخی طراحی‌های آن، مشکلاتی ایجاد کرده که در کتابخانه‌های جدیدتر برطرف شده‌اند. بیایید نگاهی به مهم‌ترین این مشکلات بیندازیم.

مشکل اول: Cancellation به سبک لبه‌ای (Edge Cancellation)

asyncio از مکانیزمی به نام edge cancellation استفاده می‌کند. یعنی وقتی یک تسک را کنسل می‌کنید، یک استثنای CancelledError فقط یک بار در آن تسک پرتاب می‌شود. اگر تسک این استثنا را بگیرد و دوباره پرتاب نکند، تسک به کار خود ادامه می‌دهد و دیگر کنسل نمی‌شود.

بیایید با یک مثال ساده این مشکل را ببینیم:

import asyncio

async def sleeper():
    try:
        await asyncio.sleep(1)
    except BaseException:
        pass  # اولین Cancellation اینجا گرفته می‌شود

    # این await هرگز برنمی‌گردد مگر اینکه تسک دوباره کنسل شود
    await asyncio.sleep(float("inf"))

async def main():
    async with asyncio.TaskGroup() as tg:
        task = tg.create_task(sleeper())
        await asyncio.sleep(0)
        task.cancel()

    print("done")

# برنامه در اینجا متوقف می‌شود
asyncio.run(main())

import asyncio

async def sleeper():
    try:
        await asyncio.sleep(1)
    except BaseException:
        pass  # اولین Cancellation اینجا گرفته می‌شود

    # این await هرگز برنمی‌گردد مگر اینکه تسک دوباره کنسل شود
    await asyncio.sleep(float("inf"))

async def main():
    async with asyncio.TaskGroup() as tg:
        task = tg.create_task(sleeper())
        await asyncio.sleep(0)
        task.cancel()

    print("done")

# برنامه در اینجا متوقف می‌شود
asyncio.run(main())

این رفتار غیرمنتظره، یکی از رایج‌ترین مشکلاتی است که برنامه‌نویسان asyncio با آن دست و پنجه نرم می‌کنند.

مشکل دوم: از دست رفتن نتیجه‌ی Future هنگام Cancellation

مشکل دیگری که در asyncio وجود دارد، مربوط به زمانی است که یک تسک در آستانه‌ی دریافت نتیجه از یک Future است و همزمان کنسل می‌شود. در این حالت، CancelledError جایگزین مقدار Future می‌شود و نتیجه از دست می‌رود:

import asyncio

async def receive(f):
    print(await f)  # این خط هرگز اجرا نمی‌شود
    await asyncio.sleep(1)

async def main():
    f = asyncio.get_running_loop().create_future()
    task = asyncio.create_task(receive(f))
    await asyncio.sleep(0)
    f.set_result("hello")  # نتیجه تنظیم می‌شود
    task.cancel()  # اما کنسل شدن، نتیجه را از بین می‌برد

    try:
        await task
    except asyncio.CancelledError:
        pass

# هیچ خروجی چاپ نمی‌شود
asyncio.run(main())

import asyncio

async def receive(f):
    print(await f)  # این خط هرگز اجرا نمی‌شود
    await asyncio.sleep(1)

async def main():
    f = asyncio.get_running_loop().create_future()
    task = asyncio.create_task(receive(f))
    await asyncio.sleep(0)
    f.set_result("hello")  # نتیجه تنظیم می‌شود
    task.cancel()  # اما کنسل شدن، نتیجه را از بین می‌برد

    try:
        await task
    except asyncio.CancelledError:
        pass

# هیچ خروجی چاپ نمی‌شود
asyncio.run(main())

مشکل سوم: asyncio.shield یک تله‌ی خطرناک

تابع asyncio.shield برای محافظت از یک تسک در برابر کنسل شدن طراحی شده است. اما استفاده از آن به‌شدت خطرناک است. وقتی تسک میزبان کنسل می‌شود، تسک شیلد شده یتیم می‌شود و اگر در آن خطایی رخ دهد، فقط یک اخطار در کنسول چاپ می‌شود و استثنا به جایی نمی‌رسد.

جالب است بدانید که همین مشکل، یکی از عوامل اصلی یک قطعی بزرگ در ChatGPT بوده است. تیم OpenAI در گزارش خود اعلام کرد که استفاده‌ی نادرست از asyncio.shield در کتابخانه‌ی Redis Python، به این قطعی منجر شده است.

import asyncio
import gc

async def shielded_task():
    fut = asyncio.get_running_loop().create_future()
    await fut  # اینجا برای همیشه منتظر می‌ماند

async def host_task():
    await asyncio.shield(shielded_task())

async def main():
    async with asyncio.TaskGroup() as tg:
        task = tg.create_task(host_task())
        await asyncio.sleep(0)
        task.cancel()
        await asyncio.sleep(0)
        gc.collect()  # تسک یتیم شده جمع‌آوری می‌شود

# اخطار: Task was destroyed but it is pending!
asyncio.run(main())

import asyncio
import gc

async def shielded_task():
    fut = asyncio.get_running_loop().create_future()
    await fut  # اینجا برای همیشه منتظر می‌ماند

async def host_task():
    await asyncio.shield(shielded_task())

async def main():
    async with asyncio.TaskGroup() as tg:
        task = tg.create_task(host_task())
        await asyncio.sleep(0)
        task.cancel()
        await asyncio.sleep(0)
        gc.collect()  # تسک یتیم شده جمع‌آوری می‌شود

# اخطار: Task was destroyed but it is pending!
asyncio.run(main())

Trio؛ فلسفه‌ی Structured Concurrency

Trio یک کتابخانه‌ی ناهمگام است که از ابتدا با رویکرد ساختارمند (Structured Concurrency) طراحی شده است. در این رویکرد، همه‌ی تسک‌ها دارای چرخه‌ی عمر مشخصی هستند و هیچ تسکی نمی‌تواند خارج از یک محدوده‌ی مشخص اجرا شود.

مفهوم Nursery

در Trio، به‌جای ایجاد مستقیم تسک، از Nursery استفاده می‌کنید. Nursery مانند یک ظرف برای تسک‌های فرزند عمل می‌کند. تمام تسک‌هایی که در یک Nursery ایجاد می‌شوند، قبل از خروج از آن Nursery، باید به پایان برسند.

این رویکرد، مدیریت تسک‌ها را بسیار ساده‌تر و قابل‌پیش‌بینی‌تر می‌کند. دیگر نگران تسک‌های سرگردان و یتیم نخواهید بود.

Cancel Scope

Trio به‌جای مکانیزم edge cancellation asyncio، از Cancel Scope استفاده می‌کند. Cancel Scope یک محدوده‌ی کدی را مشخص می‌کند که در صورت نیاز، همه‌ی عملیات‌های داخل آن کنسل می‌شوند. این مکانیزم که به level cancellation معروف است، بسیار قابل‌پیش‌بینی‌تر و ایمن‌تر از رویکرد asyncio عمل می‌کند.

AnyIO؛ پلی بین دو دنیا

AnyIO یک کتابخانه‌ی ناهمگام است که APIی یکپارچه را برای هر دو backend asyncio و Trio فراهم می‌کند. شما با AnyIO کد می‌نویسید و می‌توانید آن را بدون تغییر، روی هر دو backend اجرا کنید.

اما AnyIO فقط یک لایه‌ی سازگاری نیست. این کتابخانه APIهای مدرن و الهام‌گرفته از Trio را ارائه می‌دهد که بسیاری از مشکلات asyncio را برطرف می‌کنند.

ویژگی‌های کلیدی AnyIO

Task Group معادل Nursery Trio:

AnyIO تابع create_task_group را ارائه می‌دهد که معادل Nursery در Trio است. این Task Group شامل Cancel Scope مخصوص خود است و شما می‌توانید همه‌ی تسک‌های داخل آن را یکجا کنسل کنید.

Cancel Scope هوشمند:

در AnyIO، هر تلاش برای yielding control به event loop از یک تسک کنسل‌شده، منجر به پرتاب یک CancelledError جدید می‌شود. این یعنی cancellation در AnyIO همیشه قابل‌اعتماد است:

import asyncio
import anyio

async def sleeper():
    try:
        await asyncio.sleep(1)
    except BaseException:
        pass  # اولین Cancellation اینجا گرفته می‌شود

    # اینجا دوباره CancelledError پرتاب می‌شود
    await asyncio.sleep(float("inf"))

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(sleeper)
        await asyncio.sleep(0)
        tg.cancel_scope.cancel()

    print("done")  # این چاپ می‌شود

asyncio.run(main())

import asyncio
import anyio

async def sleeper():
    try:
        await asyncio.sleep(1)
    except BaseException:
        pass  # اولین Cancellation اینجا گرفته می‌شود

    # اینجا دوباره CancelledError پرتاب می‌شود
    await asyncio.sleep(float("inf"))

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(sleeper)
        await asyncio.sleep(0)
        tg.cancel_scope.cancel()

    print("done")  # این چاپ می‌شود

asyncio.run(main())

شیلد کردن واقعی:

در AnyIO، شیلد کردن با Cancel Scope انجام می‌شود و نه با ایجاد یک تسک جدید. به این ترتیب، دیگر خبری از تسک‌های یتیم و از دست رفتن استثناها نیست.

سناریوی کاربردی: یک سرویس پردازش سفارش

فرض کنید می‌خواهید یک سرویس پردازش سفارش بنویسید که باید اطلاعات کاربر، موجودی کالا و وضعیت پرداخت را همزمان از سه سرویس مختلف دریافت کند. هر سرویس ممکن است کند کار کند یا با خطا مواجه شود.

پیاده‌سازی با asyncio خالص

import asyncio
import random

async def get_user_info(user_id):
    await asyncio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("user service unavailable")
    return {"id": user_id, "name": f"user_{user_id}"}

async def get_inventory(product_id):
    await asyncio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("inventory service unavailable")
    return {"product": product_id, "stock": random.randint(0, 100)}

async def get_payment_status(order_id):
    await asyncio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("payment service unavailable")
    return {"order": order_id, "status": "approved"}

async def process_order_asyncio(user_id, product_id, order_id):
    try:
        async with asyncio.timeout(2.0):
            user_task = asyncio.create_task(get_user_info(user_id))
            inventory_task = asyncio.create_task(get_inventory(product_id))
            payment_task = asyncio.create_task(get_payment_status(order_id))

            user, inventory, payment = await asyncio.gather(
                user_task, inventory_task, payment_task
            )

            return {"user": user, "inventory": inventory, "payment": payment}

    except TimeoutError:
        return {"status": "timeout", "order_id": order_id}
    except Exception as e:
        return {"status": "error", "order_id": order_id, "error": str(e)}

import asyncio
import random

async def get_user_info(user_id):
    await asyncio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("user service unavailable")
    return {"id": user_id, "name": f"user_{user_id}"}

async def get_inventory(product_id):
    await asyncio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("inventory service unavailable")
    return {"product": product_id, "stock": random.randint(0, 100)}

async def get_payment_status(order_id):
    await asyncio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("payment service unavailable")
    return {"order": order_id, "status": "approved"}

async def process_order_asyncio(user_id, product_id, order_id):
    try:
        async with asyncio.timeout(2.0):
            user_task = asyncio.create_task(get_user_info(user_id))
            inventory_task = asyncio.create_task(get_inventory(product_id))
            payment_task = asyncio.create_task(get_payment_status(order_id))

            user, inventory, payment = await asyncio.gather(
                user_task, inventory_task, payment_task
            )

            return {"user": user, "inventory": inventory, "payment": payment}

    except TimeoutError:
        return {"status": "timeout", "order_id": order_id}
    except Exception as e:
        return {"status": "error", "order_id": order_id, "error": str(e)}

پیاده‌سازی با AnyIO

import anyio
import random

async def get_user_info(user_id):
    await anyio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("user service unavailable")
    return {"id": user_id, "name": f"user_{user_id}"}

async def get_inventory(product_id):
    await anyio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("inventory service unavailable")
    return {"product": product_id, "stock": random.randint(0, 100)}

async def get_payment_status(order_id):
    await anyio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("payment service unavailable")
    return {"order": order_id, "status": "approved"}

async def process_order_anyio(user_id, product_id, order_id):
    try:
        # AnyIO نیز از timeout پشتیبانی می‌کند
        with anyio.fail_after(2.0):
            async with anyio.create_task_group() as tg:
                # شروع تسک‌ها با start_soon
                tg.start_soon(get_user_info, user_id)
                tg.start_soon(get_inventory, product_id)
                tg.start_soon(get_payment_status, order_id)

                # جمع‌آوری نتایج از Task Group
                # توجه: در AnyIO باید نتایج را با مکانیزم‌های دیگری جمع‌آوری کنید
                return {"status": "success"}

    except TimeoutError:
        return {"status": "timeout", "order_id": order_id}
    except Exception as e:
        return {"status": "error", "order_id": order_id, "error": str(e)}

import anyio
import random

async def get_user_info(user_id):
    await anyio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("user service unavailable")
    return {"id": user_id, "name": f"user_{user_id}"}

async def get_inventory(product_id):
    await anyio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("inventory service unavailable")
    return {"product": product_id, "stock": random.randint(0, 100)}

async def get_payment_status(order_id):
    await anyio.sleep(random.uniform(0.2, 2.0))
    if random.random() < 0.1:
        raise ValueError("payment service unavailable")
    return {"order": order_id, "status": "approved"}

async def process_order_anyio(user_id, product_id, order_id):
    try:
        # AnyIO نیز از timeout پشتیبانی می‌کند
        with anyio.fail_after(2.0):
            async with anyio.create_task_group() as tg:
                # شروع تسک‌ها با start_soon
                tg.start_soon(get_user_info, user_id)
                tg.start_soon(get_inventory, product_id)
                tg.start_soon(get_payment_status, order_id)

                # جمع‌آوری نتایج از Task Group
                # توجه: در AnyIO باید نتایج را با مکانیزم‌های دیگری جمع‌آوری کنید
                return {"status": "success"}

    except TimeoutError:
        return {"status": "timeout", "order_id": order_id}
    except Exception as e:
        return {"status": "error", "order_id": order_id, "error": str(e)}

تفاوت اصلی در اینجا، استفاده از anyio.create_task_group و anyio.fail_after است که رفتار مشابهی با asyncio دارند اما بر اساس مکانیزم level cancellation کار می‌کنند.

مزایای AnyIO نسبت به asyncio خالص

قابلیت حمل (Portability): کد شما روی هر دو backend asyncio و Trio کار می‌کند. فقط کافی است backend مورد نظر را به anyio.run پاس دهید.

مدیریت Cancellation قابل‌اعتمادتر: مکانیزم level cancellation در AnyIO بسیار قابل‌پیش‌بینی‌تر از edge cancellation asyncio است.

Task Group با Cancel Scope یکپارچه: Task Group در AnyIO شامل Cancel Scope مخصوص خود است و شما می‌توانید همه‌ی تسک‌ها را یکجا کنسل کنید.

شیلد کردن ایمن‌تر: شیلد کردن در AnyIO با Cancel Scope انجام می‌شود و تسک‌های یتیم ایجاد نمی‌کند.

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

اگر یک کتابخانه می‌نویسید که قرار است توسط دیگران استفاده شود، AnyIO بهترین انتخاب است. کتابخانه‌ی شما با هر دو backend asyncio و Trio کار خواهد کرد و کاربران بیشتری می‌توانند از آن استفاده کنند.

اگر برنامه‌ی نهایی (Application) می‌نویسید و به دنبال ساده‌ترین و قابل‌اعتمادترین راه هستید، Trio انتخاب عالی‌ای است. APIهای Trio بسیار تمیز و مستند هستند و رویکرد Structured Concurrency آن، مدیریت تسک‌ها را بسیار آسان می‌کند.

اگر مجبور به استفاده از asyncio هستید به دلیل وابستگی‌های موجود، اما از مشکلات آن خسته شده‌اید، AnyIO را به‌عنوان یک لایه‌ی روی asyncio استفاده کنید. بسیاری از مشکلات asyncio با AnyIO برطرف می‌شوند، حتی اگر backend شما asyncio باشد.

جمع‌بندی

Trio و AnyIO دو گزینه‌ی قدرتمند برای برنامه‌نویسی ناهمگام در پایتون هستند. Trio با رویکرد Structured Concurrency، تجربه‌ای تمیز و قابل‌اعتماد ارائه می‌دهد. AnyIO با ارائه‌ی API یکپارچه برای هر دو backend asyncio و Trio، هم قابلیت حمل کد را فراهم می‌کند و هم بسیاری از مشکلات asyncio را برطرف می‌سازد.

اگر از پیچیدگی‌های asyncio خسته شده‌اید، حتماً Trio و AnyIO را امتحان کنید. این کتابخانه‌ها نشان می‌دهند که برنامه‌نویسی ناهمگام در پایتون می‌تواند ساده، لذت‌بخش و قابل‌اعتماد باشد.