Как бы вы реализовали собственный асинхронный таймаут?

Асинхронный 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, а не собственная реализация. Задачи и ожидание.

Практика в реальном времени

Подготовьтесь к следующему собеседованию

Interview Boost учитывает вакансию, резюме и технологии и помогает сформулировать ответ прямо во время интервью.

Начать подготовку