Как остановить asyncio server.serve_forever() из другого потока?

Question

[Денис]

Есть 2 потока. В одном отлавливаются сигналы (и еще много всего, но для вопроса это всё лишнее), а в другом - происходит server.serve_forever():

import sys
import signal
import asyncio
from threading import Thread

class Server:
    def run_serve(self):  # функция потока
        asyncio.run(self.serve())

    async def serve(self):  # функция запуска сервера, вызывается асинхронно в потоке
        self.server = await asyncio.start_server(self.handle_connection, '0.0.0.0', 5050)
        async with self.server:
            await self.server.serve_forever()

    async def handle_connection(self, reader, writer):  # просто эхо-заглушка принятия соединения
        while True:
            data = await reader.read(2048)
            writer.write(data)

    def close(self):  # должна закрыть self.server и завершить выполнение потока (функции run_serve)
        self.server.close()  # Работает только со второго раза, принудительным завершением

# Функция обработчика
def signal_handler(signal, frame):
    server.close()
    sys.exit(0)

# Обработчик
signal.signal(signal.SIGINT, signal_handler)

# Запускаем сервер
server = Server()
thread = Thread(target=server.run_serve)
thread.start()

while True:
    pass  # здесь что-то происходит

Как корректно завершить работу сервера и asyncio.run()? Можно ли так сделать вообще? Пробовал останавливать сам loop, получая его в том потоке. Не вышло. Помогите пожалуйста, хочу разобраться, может я вообще построил некорректно всё и нужно по другому? Но у меня запускается несколько экземпляров класса Server, т.е. несколько потоков, так что думаю тут по другому никак

Answer 1

[Владимир Олохтонов]

Существует только один правильный способ остановить поток снаружи — взвести какой-нибудь флажок, на который поток будет смотреть в своём основном цикле и остановится сам.

А зачем вам много потоков с asyncio? Они вообще-то плохо сочетаются, а профита приносят мало, ведь ожидание и так асинхронное, а вычислительные операции в основном держат GIL.

Если всё же надо, то почему бы не воспользоваться ThreadPoolExecutor?

Comments

[Денис]

Спасибо за ответ. По поводу флажка - понимаю, только вот основной цикл сервера спрятан внутри self.server.serve_forever(), так что я как-то не особо понимаю, как это можно сделать (неужели придется на более низкий уровень переходить с asyncio?). Соединения все получается закрыть флажками, но сам сервер не останавливается.

По поводу потоков - их не так много. Суть в том, чтобы открыть пару серверов на разных портах и распределенно принимать клиентов. А асинхронность, соответственно, чтобы не было блокировки при принятии соединения и отсылки/получения данных.

Про ThreadPoolExecutor не слышал, посмотрю

[Денис]

Смог решить проблему.

Закрывать сервер надо асинхронно:

class Server:

    .........
    
    async def close(self):
        async with self.server:
            self.server.close()
        await self.server.wait_closed()

Соответственно переписать signal_handler, чтобы вызывал асинхронную функцию закрытия сервера:

def signal_handler(signal, frame):
    asyncio.run(server.close())
    sys.exit(0)

Так сервер перестает работать. Но при этом, нужно демонизировать поток, чтобы он закрывался вместе с основным (это кстати странно немного для меня, раньше вроде и без этого всё закрывалось, видимо что-то на низком уровне, разобраться надо):

server = Server()
thread = Thread(target=server.run_serve)
thread.daemon = True
thread.start()
thread.join()  # после старта всех потоков