Можно ли запустить экземпляр asyncio.Server, когда цикл событий уже запущен

Я пытаюсь понять, можно ли запустить экземпляр asyncio.Server, когда цикл событий уже выполняется методом run_forever (из отдельного потока, конечно). Как я понимаю, сервер можно запустить либо по loop.run_until_complete(asyncio.start_server(...)), либо по await asyncio.start_server(...), если цикл уже запущен. Первый способ для меня неприемлем, так как цикл уже запущен методом run_forever. Но я также не могу использовать выражение ожидания, так как я собираюсь запустить его из-за пределов «области цикла» (например, из основного метода, который нельзя пометить как асинхронный, верно?)

def loop_thread(loop):
    asyncio.set_event_loop(loop)
    try:
        loop.run_forever()
    finally:
        loop.close()
        print("loop clesed")

class SchedulerTestManager:
    def __init__(self): 
        ...

        self.loop = asyncio.get_event_loop()
        self.servers_loop_thread = threading.Thread(
            target=loop_thread, args=(self.loop, ))
         ...

    def start_test(self):
        self.servers_loop_thread.start()
        return self.servers_loop_thread

    def add_router(self, router):
        r = self.endpoint.add_router(router)
        host = router.ConnectionParameters.Host
        port = router.ConnectionParameters.Port
        srv = TcpServer(host, port)
        server_coro = asyncio.start_server(
            self.handle_connection, self.host, self.port)
        # does not work since add_router is not async
        # self.server = await server_coro
        # does not work, since the loop is already running
        # self.server = self.loop.run_until_complete(server_coro)
        return r


def maind():
   st_manager = SchedulerTestManager()
   thread = st_manager.start_test()
   router = st_manager.add_router(router)

Конечно, самое простое решение — добавить все маршрутизаторы (серверы) перед запуском теста (запуском цикла). Но я хочу попробовать это реализовать, чтобы можно было добавить маршрутизатор, когда тест уже запущен. Я думал, что методы loop.call_soon (call_soon_threadsafe) могут мне помочь, но, похоже, они не могут запланировать сопрограмму, а просто простую функцию.

Надеюсь, что мое объяснение не очень запутанное. Заранее спасибо!


python python-asyncio
person Alexander Babkin    schedule 12.08.2016    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Для связи между циклом обработки событий, выполняемым в одном потоке, и обычным старым добрым многопоточным кодом, выполняемым в другом потоке, вы можете использовать janusбиблиотека.

Это очередь с двумя интерфейсами: асинхронным и потокобезопасным синхронизирующим.

Это пример использования:

import asyncio
import janus

loop = asyncio.get_event_loop()
queue = janus.Queue(loop=loop)

def threaded(sync_q):
    for i in range(100):
        sync_q.put(i)
    sync_q.join()

@asyncio.coroutine
def async_coro(async_q):
    for i in range(100):
        val = yield from async_q.get()
        assert val == i
        async_q.task_done()

fut = loop.run_in_executor(None, threaded, queue.sync_q)
loop.run_until_complete(async_coro(queue.async_q))
loop.run_until_complete(fut)

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

person Andrew Svetlov    schedule 14.08.2016
comment
Спасибо! Кажется, это то, что я ищу. - person Alexander Babkin; 14.08.2016