ждать asyncio сопрограмм закончить в определенном порядке

Question

У меня есть программа-демон с кучей asyncio.coroutine-х, которые могут быть подведены к чему-то вроде этого

import asyncio 
import signal 

class Daemon: 

    def __init__(self, loop=asyncio.get_event_loop()): 
     self.loop = loop 
     self.running = False 
     self.tasks = { 
      'coroutine1': asyncio.ensure_future(self.coroutine1()), 
      'coroutine2': asyncio.ensure_future(self.coroutine2()), 
     } 

    def run(self): 
     self.running = True 
     for task in self.tasks.values(): 
      task.add_done_callback(self.task_done_callback) 
     # gracefuly close everything when SIGINT (could be ^C) is received 
     self.loop.add_signal_handler(signal.SIGINT, self.close) 
     self.loop.run_forever() 

    def close(self): 
     self.running = False 
     self.loop.run_until_complete(self.tasks['coroutine1']) 
     self.loop.run_until_complete(self.tasks['coroutine2']) 

    def task_done_callback(self, future): 
     for task in self.tasks.values(): 
      if not task.done(): 
       return 
     self.loop.stop() 

    @asyncio.coroutine 
    def coroutine1(self): 
     while self.running: 
      print('coroutine1: do stuff') 
      yield from asyncio.sleep(1) 

    @asyncio.coroutine 
    def coroutine2(self): 
     while self.running: 
      print('coroutine2: do some other stuff') 
      yield from asyncio.sleep(3) 

if __name__ == '__main__': 
    loop = asyncio.get_event_loop() 
    daemon = Daemon(loop) 
    daemon.run() 
    loop.close() 

демона корректно завершить работу, когда SIGINT является полученных программой. Когда это происходит, срабатывает метод close(), который отвечает за уведомление всех запущенных задач, что они должны закончить свою работу и остановиться. Это делается просто, установив running в False. Всякий раз, когда выполняется задача, запускается Daemon.task_done_callback. Он проверяет, выполняются ли все задачи, если это так, то он останавливает цикл.

Проблема в том, что метод close() не работает. Это потому, что я звоню loop.run_until_complete, когда цикл уже запущен (через run_forever). Это дает RuntimeError('This event loop is already running').

Важно: coroutine1 потребности закончить до coroutine2 потому coroutine1 могут возникнуть проблемы делают свои вещи, если coroutine2 не делает его вещи больше.

Мой вопрос здесь как я убеждаюсь coroutine1 заканчивается перед тем coroutine2?

Answer 1

Это способ достичь этого. (я удаляю часть signal и ограничил время работы coroutine1).

import asyncio 

class Daemon: 

    def __init__(self, loop=asyncio.get_event_loop()): 
     self.loop = loop 
     self.tasks = { 
      'coroutine1': asyncio.ensure_future(self.coroutine1()), 
      'coroutine2': asyncio.ensure_future(self.coroutine2())} 

    def run(self): 
     for task in self.tasks.values(): 
      task.add_done_callback(self.task_done_callback) 
     self.loop.run_forever() 

    def task_done_callback(self, future): 
     if all(task.done() for task in self.tasks.values()): 
      self.loop.stop() 

    @asyncio.coroutine 
    def coroutine1(self): 
     for _ in range(5): 
      print('coroutine1: doing stuff') 
      yield from asyncio.sleep(0.2) 
     print('coroutine1: done!') 

    @asyncio.coroutine 
    def coroutine2(self): 
     while not self.tasks['coroutine1'].done(): 
      print('coroutine2: doing stuff while coro1 is running') 
      yield from asyncio.sleep(0.2) 
     print('coroutine2: doing stuff after coro1 has ended') 
     yield from asyncio.sleep(1) 
     print('coroutine2: done!') 

if __name__ == '__main__': 
    loop = asyncio.get_event_loop() 
    daemon = Daemon(loop) 
    daemon.run() 

главное здесь, чтобы проверить, является ли coroutine1 все еще работает (ваш атрибут self.tasks может быть запрошен только для этого).

сейчас, чтобы интегрировать это с signal, чтобы остановить coroutine1 Я предлагаю вам зарегистрировать простую функцию, которая устанавливает флаг (например, self.signal_flag). а затем петля внутри coroutine1 с чем-то вроде while not self.signal_flag: .... эти изменения я хотел бы предложить для полного решения:

class Daemon: 

    def __init__(self, loop=asyncio.get_event_loop()): 
     ... 
     self.signal_flag = False 

    def run(self): 
     ... 
     self.loop.add_signal_handler(signal.SIGINT, self.set_signal_flag) 
     self.loop.run_forever() 

    def set_signal_flag(self): 
     print('caught signal!') 
     self.signal_flag = True 

    async def coroutine1(self): 
     while not self.signal_flag: 
      print('coroutine1: doing stuff') 
      await asyncio.sleep(0.2) 
     print('coroutine1: done!') 

также отметить, что нет никакой необходимости close петли. в IDE, где интерпретатор продолжает работать это будет делать вещи более сложными, чем они должны быть ...

начиная с питона 3.5 вы можете (и, вероятно, должны) использовать этот синтаксис для сопрограмм:

async def coroutine1(self): 
    for _ in range(5): 
     print('coroutine1: doing stuff') 
     await asyncio.sleep(0.2) 
    print('coroutine1: done!')