asyncio: корректная остановка не своих сопрограмм

#python #websocket #python_3x #asyncio

Я опять с той же проблемой. В предыдущем вопросе все таски были под моим контролем,
и я мог их вручную закрывать. Но вот беда: мне тут потребовались вебсокеты. В примерах
в документации к одноимённому модулю никакого корректного закрытия не производится:

start_server = websockets.serve(hello, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()


Я поигрался с примерами, и да, при Ctrl+C ругается, если есть подключенный клиент:

$ python websocket_server.py 
^C
KeyboardInterrupt
Exception ignored in: Task was destroyed but it is pending!
task:  
  wait_for=
  cb=[_wait.._on_completion() at asyncio/tasks.py:399]>
Task was destroyed but it is pending!
task:  
  wait_for=
  cb=[_wait.._on_completion() at asyncio/tasks.py:399]>
Task was destroyed but it is pending!
task:  
  wait_for=>


Так как корректного способа завершения в документации не представлено и быстрое тыкание
наличия всяких server.close() ни к какому продуктивному результату не привело, снова
появляется вопрос о корректном завершении приложения, если таски мной не контролируются.

Это просто эта библиотека кривая? Или это характерно для любых asyncio-библиотек
и я чего-то не понимаю в самой сути asyncio и кто-нибудь ткнёт меня а какую-нибудь
матчасть по этому поводу?

Как всё-таки корректно завершать всё это дело в общем случае?



С общим случаем принцип понятен, но websockets закрываться никак не хочет. Минимальный
пример для повторения (на основе ответа @jfs):

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import asyncio
import websockets


@asyncio.coroutine
def echo(ws, path):
    print('client started')
    while True:
        data = yield from ws.recv()
        if data is None:
            break
        yield from ws.send(data)
    print('client finished')


loop = asyncio.get_event_loop()
start_server = websockets.serve(echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(start_server)

print('Listen')
try:
    server = loop.run_forever()
except KeyboardInterrupt:
    pass

server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
print('Finished')


Достаточно просто подключить к нему любой клиент (хоть из браузера — ws = new WebSocket('ws://127.0.0.1:8888/');)
и нажать Ctrl+C — отпечатается приведённая ранее ошибка (правда, уже без KeyboardInterrupt).
    

Ответы

Ответ 1

asyncio реализует кооперативную многозадачность. Это значит, что websockets библиотека
должна кооперировать (предоставлять возможность чисто завершить соединения). Был открыт
баг на websockets Review cancellation management.

Даже для обычных (preemptive) потоков не существует общего решения, которое бы корректно
остановило бы потоки вне зависимости от кода, который они исполняют (Java выучила этот
урок тяжёлым путём). Безопасным для остановки ресурсом является процесс операционной
системы, если нас не волнует потеря данных из-за неочищенного файлового буфера, уведомления
другой стороны о прекращении сетевого соединения, завершение транзакции базы данных
итд. Если волнует, то процесс также обязан кооперировать, если мы хотим его завершить
преждевременно.

Чтобы обработать Ctrl+C в Питоне, как обычно нужно поймать KeyboardInterrupt или
определить свой обработчик для SIGINT сигнала. Пример echo сервера в asyncio документации
показывает как это можно сделать (KeyboardInterrupt + wait_closed()):

import asyncio

@asyncio.coroutine
def handle_echo(reader, writer):
    data = yield from reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print("Received %r from %r" % (message, addr))

    print("Send: %r" % message)
    writer.write(data)
    yield from writer.drain()

    print("Close the client socket")
    writer.close()

loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(coro)

# Serve requests until CTRL+c is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()


В gevent (кооперативная многозадачность) также необходимо KeyboardInterrtupt ловить.

Документация websockets.server.serve() явно упоминает:


  serve() yields a Server which provides a close() method and a wait_closed() coroutine
to stop serving requests.


то есть для websocket сервера можно точно такой же код использовать для корректной
остановки:

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()