AsyncIO - How to test if my function is blocking?

user_3529833 0
Viewed 879

I'm trying to test if I'm correctly using an asyncio library (aiofiles) to see if its blocking.

How can I do this? My idea was to have an infinite loop of asyncio.sleep checking the delay between each call, while I try to insert a blocking call in between

This is my current try, but blocking is never called, I'm assuming the problem is that the infinite loop is actually also blocking.

from asyncio import sleep, get_event_loop
import time


async def test_blocking():
  while True:
    timea = time.time()
    await sleep(1)
    timeb = time.time() - timea
    delay = timeb - 1.00
    if delay >= 0.01:
      print(f'Function blocked by {delay}s')

async def blocking():
  print('Running blocking function')
  time.sleep(5)


loop = get_event_loop()
loop.run_until_complete(test_blocking())
loop.run_until_complete(blocking())
python-asyncio python-3.x python
1 Answers

Your test_blocking coroutine is not actually blocking the event loop - it's implemented correctly. However, run_until_complete is a blocking call. So your script is never actually making it to the run_until_complete(blocking()) line - it's waiting for test_blocking to return, which never happens because it's an infinite loop. Use loop.create_task instead, and you'll see the behavior you expect.

Also, you may want to consider using asyncio's debug mode to achieve what you're trying to do here. When enabled, it can alert you when a particular callback takes longer than a given amount of time:

Callbacks taking longer than 100ms are logged. The loop.slow_callback_duration attribute can be used to set the minimum execution duration in seconds that is considered “slow”.

When enabled on your test script, it prints this:

Executing <Task finished name='Task-2' coro=<blocking() done, defined at a.py:13> result=None created at /home/dan/.pyenv/versions/3.8.3/lib/python3.8/asyncio/base_events.py:595> took 5.005 seconds
user_20735950
Related