python - AsyncIO - 如何测试我的函数是否阻塞？

Question

我正在尝试测试我是否正确使用异步库 (aiofiles) 以查看它是否阻塞。

我该怎么做？我的想法是让 asyncio.sleep 无限循环检查每次调用之间的延迟，同时我尝试在两者之间插入阻塞调用

这是我目前的尝试，但从未调用过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())

Answer 1

您的 test_blocking 协程实际上并未阻塞事件循环 - 它已正确实现。但是，run_until_complete 是一个阻塞调用。因此，您的脚本实际上从未到达 run_until_complete(blocking()) 行 - 它正在等待 test_blocking 返回，这永远不会发生，因为它是一个无限循环。使用 loop.create_task相反，您会看到预期的行为。

此外，您可能需要考虑使用 asyncio's debug mode实现你在这里想要做的事情。启用后，它可以在特定回调花费的时间超过给定时间时提醒您:

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

在您的测试脚本上启用时，它会打印:

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