Асинхронный timeout ждёт завершения задачи до дедлайна, затем запрашивает отмену и корректно освобождает ресурсы. Это кооперативная отмена, а не принудительное завершение потока.
Как бы вы реализовали собственный асинхронный таймаут?
Асинхронный timeout ждёт завершения задачи до дедлайна, затем запрашивает отмену и корректно освобождает ресурсы. Это кооперативная отмена, а не принудительное завершение потока.
Короткий ответ
Что ответить на собеседовании
Подробный разбор
Ответ с пояснениями
Асинхронный timeout ждёт завершения задачи до дедлайна, затем запрашивает отмену и корректно освобождает ресурсы. Это кооперативная отмена, а не принудительное завершение потока.
Упрощённый вариант для корутины, которая корректно поддерживает отмену:
import asyncio
from contextlib import suppress
async def run_with_timeout(coro, seconds):
task = asyncio.create_task(coro)
try:
done, _ = await asyncio.wait({task}, timeout=seconds)
if task in done:
return task.result()
raise TimeoutError('Deadline exceeded')
finally:
if not task.done():
task.cancel()
with suppress(asyncio.CancelledError):
await task
Ожидание очистки может занять дополнительное время. Если задача блокирует event loop или подавляет отмену, жёсткого дедлайна нет. В прикладном коде обычно предпочтительнее стандартные asyncio.wait_for или asyncio.timeout, а не собственная реализация. Задачи и ожидание.