python - 如何监控 Python 事件循环的繁忙程度？

Question

我有一个异步应用程序，它通过 aiohttp 处理请求并执行其他异步任务(与数据库交互、处理消息、将请求本身作为 HTTP 客户端)。我想监视事件循环的繁忙程度，也许是执行代码与等待 select 完成所花费的时间。

有没有办法通过标准库事件循环或其他第三方循环(uvloop)来衡量这一点？

具体来说，我想要一个持续的饱和度百分比度量，而不仅仅是 this question 的二进制“是否繁忙”。似乎解决了。

Answer 1

挖掘源代码显示如下:

1. 事件循环基本上是executing while True 循环中的 _run_once
2. _run_once 执行所有操作 including等待选择完成
3. 超时等待选择 isn't存储在_run_once
之外的任何地方

没有什么可以阻止我们重新实现 _run_once，这样我们就可以根据需要计时。

我们可以在 select 之前计时，即 _run_once 启动时(因为在 select 没有发生任何耗时的事情)以及 select 之后的时间 is when _process_events 已启动。

从言语到行动:

import asyncio

class MeasuredEventLoop(asyncio.SelectorEventLoop):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._total_time = 0
        self._select_time = 0

        self._before_select = None

    # TOTAL TIME:
    def run_forever(self):
        started = self.time()
        try:
            super().run_forever()
        finally:
            finished = self.time()
            self._total_time = finished - started

    # SELECT TIME:
    def _run_once(self):
        self._before_select = self.time()
        super()._run_once()

    def _process_events(self, *args, **kwargs):
        after_select = self.time()
        self._select_time += after_select - self._before_select
        super()._process_events(*args, **kwargs)

    # REPORT:
    def close(self, *args, **kwargs):
        super().close(*args, **kwargs)

        select = self._select_time
        cpu = self._total_time - self._select_time
        total = self._total_time

        print(f'Waited for select: {select:.{3}f}')
        print(f'Did other stuff: {cpu:.{3}f}')
        print(f'Total time: {total:.{3}f}')

<小时/>

让我们测试一下:

import time


async def main():
    await asyncio.sleep(1)  # simulate I/O, will be handled by selectors
    time.sleep(0.01)        # CPU job, executed here, outside event loop
    await asyncio.sleep(1)
    time.sleep(0.01)


loop = MeasuredEventLoop()
asyncio.set_event_loop(loop)
try:
    loop.run_until_complete(main())
finally:
    loop.close()

结果:

Waited for select: 2.000
Did other stuff: 0.032
Total time: 2.032

<小时/>

让我们用真实的 I/O 来测试它:

import aiohttp


async def io_operation(delay):
    async with aiohttp.ClientSession() as session:
        async with session.get(f'http://httpbin.org/delay/{delay}') as resp:
            await resp.text()


async def main():
    await asyncio.gather(*[
        io_operation(delay=1),
        io_operation(delay=2),
        io_operation(delay=3),
    ])

结果:

Waited for select: 3.250
Did other stuff: 0.016
Total time: 3.266