How do Async IO Coroutines get executed?

Question

I'm looking at this piece of code from the example from here

And i want to know at what exact moment does the consumers() coroutine get called?

import asyncio
import itertools as it
import os
import random
import time

async def makeitem(size: int = 5) -> str:
    return os.urandom(size).hex()

async def randsleep(a: int = 1, b: int = 5, caller=None) -> None:
    i = random.randint(0, 10)
    if caller:
        print(f"{caller} sleeping for {i} seconds.")
    await asyncio.sleep(i)

async def produce(name: int, q: asyncio.Queue) -> None:
    n = random.randint(0, 10)
    for _ in it.repeat(None, n):  # Synchronous loop for each single producer
        await randsleep(caller=f"Producer {name}")
        i = await makeitem()
        t = time.perf_counter()
        await q.put((i, t))
        print(f"Producer {name} added <{i}> to queue.")

async def consume(name: int, q: asyncio.Queue) -> None:
    while True:
        await randsleep(caller=f"Consumer {name}")
        i, t = await q.get()
        now = time.perf_counter()
        print(f"Consumer {name} got element <{i}>"
              f" in {now-t:0.5f} seconds.")
        q.task_done()

async def main(nprod: int, ncon: int):
    q = asyncio.Queue()
    producers = [asyncio.create_task(produce(n, q)) for n in range(nprod)]
    consumers = [asyncio.create_task(consume(n, q)) for n in range(ncon)]
    await asyncio.gather(*producers)
    await q.join()  # Implicitly awaits consumers, too
    for c in consumers:
        c.cancel()

if __name__ == "__main__":
    import argparse
    random.seed(444)
    parser = argparse.ArgumentParser()
    parser.add_argument("-p", "--nprod", type=int, default=5)
    parser.add_argument("-c", "--ncon", type=int, default=10)
    ns = parser.parse_args()
    start = time.perf_counter()
    asyncio.run(main(**ns.__dict__))
    elapsed = time.perf_counter() - start
    print(f"Program completed in {elapsed:0.5f} seconds.")

I only see this line triggering the execution for both producer and consumer coroutines.

await asyncio.gather(*producers)

I don't understand how until the await line mentioned above there's no execution in the background when tasks are defined and created in these lines( because none of the print statements inside the producers and consumers are displayed).:

producers = [asyncio.create_task(produce(n, q)) for n in range(nprod)]


consumers = [asyncio.create_task(consume(n, q)) for n in range(ncon)]

user_1600898 · Answer

While create_task() doesn't start executing the coroutine immediately, it schedules execution in the background at the first possible opportunity, i.e. at the first await that suspends to the event loop.

gather() is just a helper function that waits for the given awaitables to complete. It doesn't prevent previously scheduled coroutines (such as those started with create_task, but also start_server etc.) from executing.

i want to know at what exact moment does the consumers() coroutine get called?

Since consumers is a coroutine, while it's called once, it can suspend and resume many times, each await serving as a point of suspension/resumption. When you call create_task() it is placed in a queue of runnable coroutines. In each iteration of the event loop asyncio goes through runnable coroutines and executes a "step" of each, where the step executes it until the first await that chooses to suspend. In your code the step happens when your main coroutine suspends in order to wait for gather() to complete.

How do Async IO Coroutines get executed?

1 Answers