python_syntax_asyncio_blocking_task

gistfile1.txt

import asyncio
import threading
from datetime import datetime

'''
https://stackoverflow.com/questions/76856816/python-how-to-create-cpu-load-that-block-asyncio-event-loop?noredirect=1#comment135492684_76856816
'''
def log(msg):
    print(f"{datetime.now()} {msg}")

async def blocking_task(task_name):
    num_iter = 100
    log(f"{task_name} started, thread_id: {threading.get_ident()}\n")
    for i in range(num_iter):
        log(f"{i}/{num_iter} executing {task_name} ...")  # This will show IF taskA and taskB interleave (i.e. not blocking each other)
        # await asyncio.sleep(0) # If you comment this out, taskA and taskB wont be interleaving!!! i.e. won't block each other
    log(f"{task_name} finished")

async def taskA(loop):
    await blocking_task("Blocking Task A")

async def taskB(loop):
    await blocking_task("Blocking Task B")

async def async_main():
    log(f"async_main thread_id: {threading.get_ident()}")

    loop = asyncio.get_event_loop()
    '''
    If you do this, tasks will be on separate threads and you wont see blocking happen.
        task_a = asyncio.create_task(taskA(loop))
        task_b = asyncio.create_task(taskB(loop))
        await asyncio.gather(task_a, task_b)
    '''
    await asyncio.gather(taskA(loop), taskB(loop))
    
    log("Async main finished")

if __name__ == "__main__":
    log(f"Main thread ID: {threading.get_ident()}")
    
    asyncio.run(async_main())
    
    log("Main thread finished")