Asyncio Loop.call_later 计时精度

Question

• 我正在尝试让一个函数在事件循环上定期运行
• 以指定的速率调用函数至关重要
• 我正在使用

  asyncio.call_later

递归地调度函数运行
• 看起来

  asyncio.call_later

  没有以正确的延迟调用函数

• call_later

  返回的计时器句柄确认延迟大于预期

为什么函数没有以指定的速率调用？

import asyncio
import time

def print_fps():
    global last_time

    time_now = time.time()
    dt = time_now - last_time
    if dt > 0:
        print(f"fps: {round(1/dt, 3)} [{round(dt,3)}s], target: {fps} [{round(1/fps,3)}s]")
    else:
        print("dt = 0")
    last_time = time_now

def loop_func():
    # print fps info
    print_fps()

    # schedule next iteration of loop
    loop.call_later(1/fps, loop_func)

async def main():
    loop_func()

last_time = time.time()
fps = 60

loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()

输出：

fps: 31.446 [0.032s], target: 60 [0.017s]
fps: 32.693 [0.031s], target: 60 [0.017s]
fps: 32.172 [0.031s], target: 60 [0.017s]
fps: 31.663 [0.032s], target: 60 [0.017s]
fps: 32.001 [0.031s], target: 60 [0.017s]
fps: 30.549 [0.033s], target: 60 [0.017s]
fps: 34.486 [0.029s], target: 60 [0.017s]
fps: 31.366 [0.032s], target: 60 [0.017s]
fps: 31.743 [0.032s], target: 60 [0.017s]
fps: 30.057 [0.033s], target: 60 [0.017s]

Answer 1

我在寻找一种使 asyncio.sleep 在 Windows 上准确的方法时发现了这个问题，它默认为 16ms 准确度。我开发了一个解决方案，所以我把它放在这里，以防其他人在寻找与我相同的东西时，在这里找到方法 - 抱歉，它不太具体针对您的问题。

这是一个在 Windows 上与 asyncio 配合使用的精确睡眠函数 - 您可能需要在其他平台上修改延迟时间：

import sys
import asyncio
import time

async def accurate_sleep(s):
    start = time.perf_counter()
    if s > 0.016: # 16ms for windows sleep accuracy
        await asyncio.sleep(s - 0.016)
    while (time.perf_counter() - start) < s:
        await asyncio.sleep(0)


async def sleep_test():
    samples = 50
    err = 0.0
    maxerr = 0
    for i in range(0, samples):
        start_ns = time.perf_counter_ns()
        await asyncio.sleep(i * 0.001)
        result_us = (time.perf_counter_ns() - start_ns) // 1_000
        # print("    " + str(i) + " ms\t" + str(result_ms - i) + " ms err")
        err += abs(result_us - i * 1000)
        maxerr = max(maxerr, abs(result_us - i * 1000))

    ferr = 0.0
    fmaxerr = 0
    for i in range(0, samples):
        start_ns = time.perf_counter_ns()
        await accurate_sleep(i * 0.001)
        result_us = (time.perf_counter_ns() - start_ns) // 1_000
        # print("    " + str(i) + " ms\t" + str(result_ms - i) + " ms err")
        ferr += abs(result_us - i * 1000)
        fmaxerr = max(fmaxerr, abs(result_us - i * 1000))

    print(
        "  Basic asyncio sleep err avg "
        + str(err / samples)
        + " us, max err "
        + str(maxerr)
        + " us"
    )
    print(
        "  Fancy Asyncio sleep err avg "
        + str(ferr / samples)
        + " us, max err "
        + str(fmaxerr)
        + " us"
    )


if __name__ == "__main__":
    asyncio.run(sleep_test())

我的 Win 11 盒子上的测试结果：

基本异步睡眠错误平均 8215.12 us，最大错误 16463 us

花哨的 Asyncio sleep err 平均 1.8 us，最大 err 4 us