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

Question

Ниже приведен простой эхо-сервер.Но если клиент не отправляет ничего в течение 10 секунд, я хочу закрыть соединение.

import asyncio


async def process(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    print("awaiting for data")
    line = await reader.readline()
    print(f"received {line}")
    writer.write(line)
    print(f"sent {line}")
    await writer.drain()
    print(f"Drained")


async def timeout(task: asyncio.Task, duration):
    print("timeout started")
    await asyncio.sleep(duration)
    print("client unresponsive, cancelling")
    task.cancel()
    print("task cancelled")


async def new_session(reader, writer):
    print("new session started")
    task = asyncio.create_task(process(reader, writer))
    timer = asyncio.create_task(timeout(task, 10))
    await task
    print("task complete")
    timer.cancel()
    print("timer cancelled")
    writer.close()
    print("writer closed")


async def a_main():
    server = await asyncio.start_server(new_session, port=8088)
    await server.serve_forever()


if __name__ == '__main__':
    asyncio.run(a_main())

Если клиент отправляет сообщение, оно работает нормально.Но в другом случае, когда клиент молчит, он не работает

Когда клиент отправляет сообщение:

new session started
awaiting for data
timeout started
received b'slkdfjsdlkfj\r\n'
sent b'slkdfjsdlkfj\r\n'
Drained
task complete
timer cancelled
writer closed

Когда клиент молчит после открытия соединения

new session started
awaiting for data
timeout started
client unresponsive, cancelling
task cancelled

Нет task complete, timer cancelled, writer closed.

1. В чем проблема с приведенным выше кодом?
2. Есть ли лучший способ реализовать тайм-ауты?

---

Обновление

Разобрался с проблемой, похожезадача была фактически отменена, но исключение было беззвучно проигнорировано. Исправлена ​​проблема с перехватом CancelledError

async def new_session(reader, writer):
    print("new session started")
    task = asyncio.create_task(process(reader, writer))
    timer = asyncio.create_task(timeout(task, 10))
    try:
        await task
    except asyncio.CancelledError:
        print(f"Task took too long and was cancelled by timer")
    print("task complete")
    timer.cancel()
    print("timer cancelled")
    writer.close()
    print("writer closed")

Вторая часть все еще остается.Есть ли лучший способ реализовать тайм-ауты?

---

Update2

Полный код с использованием wait_for.Код тайм-аута больше не нужен.Проверка принята решение ниже:

async def new_session(reader, writer):
    print("new session started")
    try:
        await asyncio.wait_for(process(reader, writer), timeout=5)
    except asyncio.TimeoutError as te:
        print(f'time is up!{te}')
    finally:
        writer.close()
        print("writer closed")

Answer 1

Есть ли лучший способ реализовать тайм-ауты?

Вы можете использовать asyncio.wait_for вместо timeout.Он имеет похожую семантику, но уже поставляется с asyncio.Кроме того, вы можете дождаться будущего, которое он вернет, чтобы определить, истекло ли время ожидания.

Answer 2

Я использую следующий код при установлении соединения.Я бы предложил использовать wait_for аналогично для вашего кода.

fut = asyncio.open_connection( self.host, self.port, loop=self.loop )
try:
   r, w = await asyncio.wait_for(fut, timeout=self.connection_timeout)
except asyncio.TimeoutError:
   pass