Coroutine در پایتون (2)

در بخش اول از سری مقالات «Coroutine در پایتون» به ماهیت این مفهوم پرداختیم و آن را با سایر روش‌های هم‌زمانی در پایتون مقایسه کردیم. در بخش دوم از این سری‌مقاله به پیاده‌سازی و استفاده از coroutineها در پایتون می‌پردازیم.

Coroutine در پایتون (۱)

براساس نکات اشاره شده در بخش اول، به‌طور خلاصه coroutine‌‌ها توابعی (Task‌های) هستند که مراحل اصلی خود را انجام می‌دهند و فقط زمانی کنترل را به سایر Taskها واگذار می‌کنند که به اتمام رسیده و یا منتظر منابع باشند.

پیاده‌سازی Coroutine به کمک Generator

تولیدکننده‌ها

تولیدکننده‌‌ها یا Generator‌ها نوع خاصی از توابع هستند که یک Lazy iterator برمی‌گردانند؛ به این معنی که شما می‌توانید مانند List بر روی آن‌ها حلقه بزنید. اما برخلاف Lazy Iterator ،Listها محتویات را در درون Memory ذخیره نمی‌کنند.

>>> def fib(count: int):
...     a, b = 1, 0
...     for _ in range(count):
...         a, b = b, a + b
...         yield b

دنباله‌ی Fibonacci یک مثال معروف برای Generatorها است. هربار در حلقه، دو عدد قبلی با یکدیگر جمع می‌شوند و نتیجه را yield می‌کند. اما همانطور که قبلاً اشاره شد زمانی که این تابع را فراخوانی کنیم، مقادیر را مستقیماً دریافت نمی‌کنیم، بلکه generator object دریافت می‌شود. در واقع کد درون تابع هنوز اجرا نشده است.

>>> gen = fib(5)
>>> print(gen)
<generator object fib at 0x7f192e4d2b20>
>>> while True:
...     print(next(gen))
...     
... 
1
1
2
3
5
Traceback (most recent call last):
  File "<input>", line 2, in <module>
    print(next(gen))
StopIteration

هربار که با استفاده از gen ،next را فراخوانی می‌کنیم، از جایی که قبلاً آن را ترک کرده بودیم به آن باز می‌گردیم و با yield‍ بعدی از آن خارج می‌شویم. همچنین زمانی که Generator تمام شود (صریح یا ضمنی return انجام شود)، generator object خطای StopIteration پرتاپ می‌کند. به‌علاوه امکان ارتباط و ارسال داده به generator نیز وجود دارد.

>>> def double_inputs():
...     while True:
...         x = yield
...         yield x * 2
...
>>> gen = double_inputs()
>>> next(gen)       # run up to the first yield
>>> gen.send(10)    # goes into 'x' variable
20
>>> next(gen)       # run up to the next yield
>>> gen.send(6)     # goes into 'x' again
12

برای آشنایی بیشتر با generatorها می‌توانید به دوره‌ی آموزش پایتون پیشرفته کوئرا کالج مراجعه کنید.

نمونه پیاده‌سازی برای Coroutine

قبل از ادامه‌ی مطلب باید با مسئله‌ای که قصد داریم به کمک coroutine حل کنیم آشنا شویم و ارتباط آن با این مفهوم را درک کنیم. به همین دلیل الگوریتم مرتب‌سازی جالبی را معرفی می‌کنیم. الگوریتم sleep sort به این صورت عمل می‌کند که به تعداد اعداد، در آرایه‌ای که قصد داریم مرتب کنیم coroutine ایجاد می‌کند و در هرکدام از این coroutineها، به اندازه‌ی همان عدد sleep می‌کند و سپس عدد را به لیستِ مرتب اضافه می‌کند. به این صورت هرچه عدد بزرگتر باشد، ثانیه‌ی بیشتری منتظر می‌ماند تا به لیست نهایی اضافه شود.

برای مثال اگر آرایه‌ای مانند [3 ,4 ,1 ,2] را به این روش مرتب کنیم، ابتدا باید به‌ازای هر عنصر از این آرایه coroutineای ایجاد کنیم که در آن به اندازه‌ی مقدار آن عنصر، sleep می‌کنیم. سپس باید coroutine‌هایی که ایجاد شده‌اند، اجرا شوند. مراحل اجرا را می‌توان به این صورت بیان کرد:

ابتدا coroutineای که برای ۲ ایجاد شده است، شروع به کار می‌کند و بلافاصله کنترل را واگذار می‌کند. سپس به‌ترتیب برای ۱، ۴‌ و ۳ هم همین اتفاق می‌افتد. در ادامه، برنامه منتظر می‌ماند تا یکی از coroutine‌ها کنترل برنامه را به دست بگیرد. اولین coroutineای که این کار را انجام می‌دهد، coroutineای است که نسبت به سایر coroutine‌ها کمترین زمان را منتظر می‌ماند؛ یعنی ابتدا ۱ کنترل را بدست می‌گیرد و ۱ را به لیستِ مرتب اضافه می‌کند. سپس به‌ترتیب برای ۲، ۳ و ۴ نیز این اتفاق می‌افتد.

پیاده‌سازی

طبق مواردی که تابه‌حال گفتیم، شکل کلی کد می‌تواند به این صورت باشد:

def sleep_sort(numbers: List[int]) -> List[int]:
    result = []

    def sleep_add(number: int): # this generator used for create coroutine for each number 
        # TODO: sleep for number second
        result.append(number)

    # TODO: start coroutine in such a list -> [sleep_add(number) for number in numbers]
    return result

همان طور که احتمالاً حدس می‌زنید، برای صبرکردن در تابع sleep_add نمی‌توانیم از time.sleep استفاده کنیم. برای اینکار باید یک generator ایجاد کنیم که تا زمانی که نیاز است، با yield کردن از آن خارج شویم (کنترل را واگذار کنیم.)

def sleep(duration: float) -> None:
    now = time.time()
    threshold = now + duration
    while now < threshold:
        yield
        now = time.time()

در generator بالا، در ابتدا، مدت‌زمانی که کار generator به اتمام می‌رسد، محاسبه می‌شود. سپس تا هنگامی که زمان فعلی، کمتر از این زمان است با yield از آن خارج می‌شود. همچنین بعد از گذشتن از این زمان، کار generator به اتمام می‌رسد.

قدم بعدی ایجاد چارچوبی برای اجراکردن coroutineهای ایجادشده است.

def wait(tasks: Iterable[Generator]) -> List[Any]:
    pending = list(tasks)
    tasks = {task: None for task in pending}
    before = time.time()

    while pending:
        for gen in pending:
            try:
                tasks[gen] = gen.send(tasks[gen])
            except StopIteration as e:
                tasks[gen] = e.value
                pending.remove(gen)

    print(f"duration: {time.time() - before:.3}")
    return list(tasks.values())

تابع wait که برای اجرا کردن coroutineها ایجاد شده است، به این صورت عمل می‌کند که ابتدا لیستی از coroutineهایی که باید اجرا شوند را می‌سازد (pending)، سپس دیکشنریِ tasks را برای نگهداری خروجی هر coroutine ایجاد می‌کند و تا زمانی که coroutineای برای اجرا وجود داشته باشد(while pending)، آن را اجرا کرده (در واقع gen.send در اینجا کنترل اجرا را به coroutine می‌دهد) و نتیجه‌ی آن را ذخیره می‌کند. همچین درصورت پرتاب خطای StopIteration (پایان کار coroutine) با try except مدیریت و از لیست اجرا pending حذف می‌شود.

در نهایت تابع sleep_sort به این صورت می‌شود:

def sleep_sort(numbers: List[int]) -> List[int]:
    result = []

    def sleep_add(number: int) -> Generator:
        yield from sleep(number)
        result.append(number)

    wait([sleep_add(number) for number in numbers])
    return result

حال اگر این تابع را اجرا کنیم، یک خروجی مشابه زیر دریافت می‌کنیم:

if __name__ == "__main__":
    range_10 = list(range(10))
    random.shuffle(range_10)
    print(f"range_10: {range_10}")
    print(f"sleep_sort(range_10): {sleep_sort(range_10)}")

# OUTPUt
# range_10: [1, 4, 8, 7, 9, 3, 0, 5, 6, 2]
# duration: 9.0
# sleep_sort(range_10): [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

AsyncIO

کاری که تا به اینجا انجام دادیم، اختراع اولیه‌ی asyncIO بود. در واقع این قطعه‌کد صرفاً جهت نشان دادن نحوه‌ی اجرا‌شدن coroutine‌ها گفته شده و به اندازه‌ی کافی انعطاف‌پذیر نیست و استفاده از آن همراه با سایر کتابخانه‌ها به‌سادگی امکان‌پذیر نیست. پس توصیه می‌کنم در کدنویسی‌تان از این روش استفاده نکنید.

از نسخه ۳.۵، پایتون از async def برای تعریف coroutine پشتیبانی می‌کند که صدا‌زدن آن، generator نتیجه را برنمی‌گرداند. همچنین با استفاده از await قادر به رفتاری مشابه با yield from در بالا هستیم.

حال با استفاده از AsyncIO می‌توانیم راه‌حل مسئله‌ی بیان‌شده را به این صورت بازنویسی کنیم:

def sleep_sort(numbers: List[int]) -> List[int]:
    result = []

    async def sleep_add(number: int) -> Coroutine:
        await asyncio.sleep(number)
        result.append(number)

    asyncio.run(asyncio.wait([sleep_add(number) for number in numbers]))
    return result

مسلماً کتابخانه‌ی AsyncIO جزئیات بسیار بیشتری برای ارائه دارد، اما متأسفانه امکان پرداختن به آن در این مقاله وجود ندارد. ولی امیدوارم تصویر مناسبی از coroutineها در این دو مقاله ارائه داده شده باشد تا ادامه‌ی این مسیر را برای شما هموارتر کند.

برای دسترسی به کد‌های این مقاله به این لینک مراجعه کنید.