使用 asyncio,为了保持较低的延迟,协程之间不应花费很长时间。
为确保这一点,记录长时间运行可能会有用。
一个可运行的例子展示了我的意思:
#!/usr/bin/env python3
import time
from time import monotonic_ns
import asyncio
from pyaux.runlib import init_logging
init_logging(level=1)
state = {}
class LoggyLoopMixin:
_warn_run_once_time = 0.2
def _run_once(self):
start_time = monotonic_ns()
result = super()._run_once()
end_time = monotonic_ns()
self._check_run_once_time_diff(start_time=start_time, end_time=end_time)
def _check_run_once_time_diff(self, start_time, end_time):
time_diff = (end_time - start_time) / 1e9
if time_diff > self._warn_run_once_time:
self._handle_run_once_time_diff(time_diff=time_diff)
def _handle_run_once_time_diff(self, time_diff, **kwargs):
print("WARNING: large run_once time difference: %.3fs" % (time_diff,))
def mixin_asyncio_loop(mixin_cls):
superclass = asyncio.get_event_loop_policy().__class__
# assuming that `superclass._loop_factory` is a class.
# pylint: disable=protected-access
class MixinnedEventLoop(mixin_cls, superclass._loop_factory):
""" ... """
class MixinnedEventLoopPolicy(superclass):
_loop_factory = MixinnedEventLoop
elp = MixinnedEventLoopPolicy()
asyncio.set_event_loop_policy(elp)
return elp
mixin_asyncio_loop(LoggyLoopMixin)
async def heartbeat(sleep_time=0.1, warn_time=0.2):
"""
An on-top-of-the-loop implementation, which is even worse.
See: https://github.com/pebble/event-loop-lag
"""
prev = monotonic_ns()
warn_time_ns = int(warn_time * 1e9)
while True:
now = monotonic_ns()
if now - prev > warn_time_ns:
print("Hearbeat missed (%.3fs from last)" % ((now - prev) / 1e9,))
else:
print("Heartbeat time: %d" % (now - prev,))
prev = monotonic_ns()
await asyncio.sleep(sleep_time)
async def amain():
# asyncio.ensure_future(heartbeat())
sleep_time = 0.01
while sleep_time < 3.5:
print("Busy for %.3fs" % (sleep_time,))
# await asyncio.sleep(sleep_time)
time.sleep(sleep_time)
await asyncio.sleep(0.001) # yield at all
sleep_time *= 1.5
def main():
loop = asyncio.get_event_loop()
loop.run_until_complete(amain())
if __name__ == '__main__':
main()
但是,我希望它已经存在实现,而且感觉我以前见过这个功能,但找不到它。
并且,as it has been noted elsewhere ,不鼓励子类化 asyncio。
是否有此功能的现有实现?
最佳答案
Are there existing implementations of this feature?
是的,这是一个 debug mode事件循环。
按照查看结果的方式更改代码:
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.run_until_complete(amain())
你会看到这样的东西:
Busy for 0.034s
Busy for 0.051s
Busy for 0.076s
Busy for 0.114s
Executing <Handle ...> took 0.125 seconds
Busy for 0.171s
Executing <Handle ...> took 0.172 seconds
Busy for 0.256s
Executing <Handle ...> took 0.282 seconds
Busy for 0.384s
关于python - 异步 : warn about long-running handlers,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55147495/