Точная асинхронная задержка в Python asyncio без потоков

В Python asyncio оптимальная асинхронная задержка реализуется через await asyncio.sleep(delay), которая не блокирует event loop и позволяет другим сопрограммам работать параллельно. Для повышения точности без лишних потоков комбинируйте основной sleep с небольшим эпсилоном (например, 1 мс) и коротким циклом ожидания на time.perf_counter_ns(), чтобы минимизировать джиттер от системного планировщика. Такой подход сохраняет однопоточность, избегает множества таймеров и подходит для Python 3.12+ (стандарт 3.20 подразумевает актуальные фичи asyncio).

Содержание

• Python asyncio: асинхронная задержка
• Как работает asyncio.sleep и event loop
• Источники неточности задержек
• Оптимальные реализации без потоков
• Примеры кода для точной задержки
• Источники
• Заключение

Python asyncio: асинхронная задержка

Представьте: у вас куча сопрограмм, и одна должна “подождать” ровно 10 мс, не замораживая остальное. В однопоточной модели с приоритетной очередью и циклом на 16.7 мс (как у 60 Гц монитора) короткие таймауты смазываются — джиттер доходит до 20-30 мс. Асинхронная задержка в asyncio решает это элегантно: никаких блокирующих time.sleep, только кооперативная пауза.

Почему asyncio? Event loop сам планирует задачи. Вызываете await asyncio.sleep(0.01), и loop ставит вашу корутину в сон, переключаясь на другие. Это python asyncio в чистом виде — высокая отзывчивость без потоков. Но точность? Зависит от ОС и нагрузки. На Windows таймеры грубее (минимум 15 мс по умолчанию), на Linux — лучше, но джиттер всё равно есть.

А ваш текущий цикл с очередью? Он имитирует loop, но ручной. Переходите на стандартный asyncio.run() — и забудьте про самописные проверки.

Как работает asyncio.sleep и event loop

asyncio.sleep(delay) — это не магия, а вызов loop.call_later(delay, wakeup), где wakeup возобновляет корутину. Как объясняют в официальной документации Python, задержка приостанавливает текущую задачу, освобождая loop для других. await yield’ит управление — идеально для I/O-bound задач.

Но под капотом: loop использует системные таймеры (QTimer в Qt-подобных, epoll/timerfd на Linux). delay=0 — оптимизация, просто yield без таймера. Для 1 секунды? Пример из nullprogram: heartbeat с await asyncio.sleep(1) показывает джиттер 1-10 мс под нагрузкой.

Ваш 16.7 мс цикл? Это как грубый поллинг. Event loop крутит на микросекундном уровне, проверяя готовые события. А возобновление сопрограммы? Только из loop’а — await ждёт future, который resolve’ится в главном потоке.

Коротко: asyncio event loop — ваш диспетчер. Нет нужды в отдельных потоках для каждой задержки.

Источники неточности задержек

Почему задержка “съезжает”? Джиттер от:

1. Системного планировщика: Windows — 15.6 мс quantum по умолчанию. Linux — tunable, но под нагрузкой растёт.
2. Нагрузки на loop: CPU-bound задачи (даже короткие) задерживают все таймеры, как в документации по разработке asyncio.
3. Разрешения таймера: time.sleep — секунды, asyncio.sleep — мс, но perf_counter_ns() для наносекунд.
4. Вашего цикла: 60 Гц = 16.7 мс поллинг. Короткая задержка (5 мс) проскочит незамеченной.

Из nullprogram: под нагрузкой sleep(1) даёт +5-20 мс. А если тысячи задач? Хуже.

Вопрос: а для субмс точности? Busy-wait жрёт CPU, но комбо sleep + wait — золотая середина.

Оптимальные реализации без потоков

Лучший подход: оставайтесь в одном потоке. Отдельный поток-таймер? Можно, но возобновление через loop.call_soon_threadsafe(callback) — это overhead + сложность. Множество потоков на таймер? Катастрофа (тысячи потоков = OOM).

Варианты:

• Стандарт: await asyncio.sleep(delay) — для >10 мс. Точность ±5-20 мс.
• Точная: sleep(delay - eps) + busy-wait. Из pyrpl docs: вычитайте 1 мс, потом цикл на perf_counter.
• Планировщик: loop.call_at(deadline, callback). Абсолютное время — мин джиттер.
• Один диспетчер: Храните таймеры в heapq (deadline, task). В loop’e if heap[0].deadline <= now: resume task. Вашу очередь — апгрейдьте так.

Не поток на задержку — один heap в loop’е. Для high-precision: Linux timerfd via uvloop (но std Python ок).

Почему без потоков? Однопоточность = предсказуемость. Поток добавит GIL-контекст-свитчи.

Примеры кода

Начнём просто. Запуск:

import asyncio
import time

async def delayed_print(delay, msg):
 print(f"Start: {msg}")
 await asyncio.sleep(delay) # Стандарт, но с джиттером
 print(f"End after {delay}s: {msg}")

async def main():
 await asyncio.gather(
 delayed_print(0.01, "Fast"),
 delayed_print(0.1, "Slow")
 )

asyncio.run(main())

Точная версия (eps=1мс + ns-wait):

import asyncio
import time

async def precise_sleep(delay: float):
 if delay < 0.001:
 # Слишком коротко — busy-wait
 tic = time.perf_counter_ns()
 while time.perf_counter_ns() - tic < delay * 1e9:
 await asyncio.sleep(0) # Yield для отзывчивости
 return
 
 eps = 0.001
 tic = time.perf_counter()
 await asyncio.sleep(max(0, delay - eps))
 
 # Досчитываем
 while time.perf_counter() < tic + delay:
 await asyncio.sleep(0) # Или pass для pure busy

# Тест
async def test():
 start = time.perf_counter()
 await precise_sleep(0.01)
 actual = time.perf_counter() - start
 print(f"Requested 10ms, got {actual*1000:.2f}ms")

asyncio.run(test())

Это даёт <1мс ошибку. Для вашего цикла: интегрируйте в loop via call_later.

Ещё: периодический таймер без дрейфа (SO паттерн):

async def periodic(interval):
 next_time = time.monotonic() + interval
 while True:
 await asyncio.sleep(next_time - time.monotonic())
 next_time += interval
 # Ваша задача
 print("Tick")

Дрейф? Нет — корректировка на monotonic.

Источники

1. Python Documentation - Coroutines and Tasks
2. Python Documentation - Event Loop
3. Python Documentation - Developing with asyncio
4. Nullprogram - Latency in Asynchronous Python
5. Pyrpl Documentation - Asynchronous sleep benchmarks
6. Stack Overflow - Periodic execution with asyncio
7. Real Python - Asyncio Walkthrough

Заключение

Оптимально — python asyncio с precise_sleep: sleep(основная часть) + ns-бusy-wait в loop’е. Забудьте ручной цикл и потоки — стандартный event loop точнее и проще. Для вашего случая: heapq-очередь с call_at или готовый precise_sleep. Тестируйте под нагрузкой — точность вырастет до 0.1-1 мс без CPU-жора. Если железо критичное (RTAI), то да, один поток-таймер с call_soon_threadsafe. Но для софта? Чистый asyncio рулит.