Asyncio

Python has a native asyncio package that supports async and await (similar to JS). We will be using the lastest Python version 3.11, some features may not be available in older Python versions.

Task Group

In a task group context, tasks are executed concurrently. In the example below, 2 tasks should finish at around the same time.

"""asyncio task group (python 3.11 only)"""
 
import asyncio
 
import time
 
  
 
async def say_after(delay, what):
	await asyncio.sleep(delay)
	print(what)
 
 
async def main():
	async with asyncio.TaskGroup() as tg:
		task1 = tg.create_task(say_after(2, 'hello'))
		task2 = tg.create_task(say_after(2, 'world'))
 
		print(f"started at {time.strftime('%X')}")
	# The await is implicit when the context manager exits.
	print(f"finished at {time.strftime('%X')}")
asyncio.run(main())
print("finished all")

asyncio.gather()

The gather method also blocks the code from keep running

import asyncio
 
async def say(msg: str):
    await asyncio.sleep(1)
    print(f"say: {msg}")
 
 
async def main():
    await say("a")
 
    # this is also valid:
    await asyncio.gather(
        say("b"),
        say("c")
    )
    print("finished")
 
asyncio.run(main())

Not thread safe. Queues — Python 3.11.2 documentation asyncio.Queue is similar to Golang: WaitGroups Tasks are created, added to queue and executed concurrently. task_done() is called after each job is finished. queue.join() will wait for all tasks in queue to finish.

import asyncio
import random
import time
 
 
async def worker(name, queue):
    while True:
        # Get a "work item" out of the queue.
        sleep_for = await queue.get()
 
        # Sleep for the "sleep_for" seconds.
        await asyncio.sleep(sleep_for)
 
        # Notify the queue that the "work item" has been processed.
        queue.task_done()
 
        print(f'{name} has slept for {sleep_for:.2f} seconds')
 
 
async def main():
    # Create a queue that we will use to store our "workload".
    queue = asyncio.Queue()
 
    # Generate random timings and put them into the queue.
    total_sleep_time = 0
    for _ in range(20):
        sleep_for = random.uniform(0.05, 1.0)
        total_sleep_time += sleep_for
        queue.put_nowait(sleep_for)
 
    # Create three worker tasks to process the queue concurrently.
    tasks = []
    for i in range(3):
        task = asyncio.create_task(worker(f'worker-{i}', queue))
        tasks.append(task)
 
    # Wait until the queue is fully processed.
    started_at = time.monotonic()
    await queue.join()
    total_slept_for = time.monotonic() - started_at
 
    # Cancel our worker tasks.
    for task in tasks:
        task.cancel()
    # Wait until all worker tasks are cancelled.
    await asyncio.gather(*tasks, return_exceptions=True)
 
    print('====')
    print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds')
    print(f'total expected sleep time: {total_sleep_time:.2f} seconds')
 
 
asyncio.run(main())

Asyncio

Task Group

asyncio.gather()

Queue

On this page