如何停止()多个嵌套异步循环中的最后一个循环？

Question

我有时需要多个异步协程，其中例程将被嵌套:协程B在协程A中运行，C在B中运行，依此类推。问题在于停止给定的循环。例如，在最后一个顶级循环中使用loop.stop()，比如循环'C‘，实际上会杀死所有异步协程--而不仅仅是这个循环'C’。我怀疑stop()实际上杀死了协程A，并且通过这样做它消除了所有其他依赖的例程。嵌套的例程是call_soon_threadsafe，并且所有例程都以run_forever开头。

我尝试使用特定的循环名称，或'return'，或'break‘(在协程内的while循环中)，但没有任何东西退出循环--除了stop()，它会同时终止非特定的所有循环。

我在这里描述的问题实际上与我之前的一个问题有关……python daemon server crashes during HTML popup overlay callback using asyncio websocket coroutines ...which我以为我已经解决了--直到遇到了这个loop.stop()问题。

下面是我的Python3.4.3示例代码，其中我尝试在websocket作业完成后立即停止() coroutine_overlay_websocket_server循环。如上所述，我在当前状态下的代码会中断所有正在运行的循环。此后，fmDaemon重新创建一个新的异步循环，该循环对之前计算的内容一无所知：

import webbrowser
import websockets
import asyncio

class fmDaemon( Daemon):

    # Daemon - see : http://www.jejik.com/articles/2007/02/a_simple_unix_linux_daemon_in_python/
    # Daemon - see : http://www.jejik.com/files/examples/daemon3x.py

    def __init__( self, me):
        self.me = me

    def run( self):

        while True:

            @asyncio.coroutine
            def coroutine_daemon_websocket_server( websocket, path):

                msg = yield from websocket.recv()

                if msg != None:
                    msg_out = "{}".format( msg)
                    yield from websocket.send( msg_out)
                    self.me = Function1( self.me, msg)

            loop = asyncio.get_event_loop()
            loop.run_until_complete( websockets.serve( coroutine_daemon_websocket_server, self.me.IP, self.me.PORT))
            loop.run_forever()


def Function1( me, msg):

    # doing some stuff :
    # creating HTML file,
    # loading HTML webpage with a webbrowser call,
    # awaiting HTML button press signal via websocket protocol :

    @asyncio.coroutine
    def coroutine_overlay_websocket_server( websocket, path):

        while True:
            msg = yield from websocket.recv()
            msg_out = "{}".format( msg)
            yield from websocket.send( msg_out)

            if msg == 'my_expected_string':
                me.flags['myStr'] = msg
                break

        loop.call_soon_threadsafe( loop.stop)

    loop = asyncio.get_event_loop()
    loop.call_soon_threadsafe( asyncio.async, websockets.serve( coroutine_overlay_websocket_server, me.IP, me.PORT_overlay))
    loop.run_forever()
    loop.call_soon_threadsafe( loop.close)

    # program should continue here...

我的两个问题: 1)有没有一种方法可以退出给定的协程而不杀死底层的协程? 2)或者，你是否知道一种不使用异步的方法来读取websocket调用？

Answer 1

我仍然对您要做的事情感到有点困惑，但绝对没有必要尝试嵌套事件循环-您的程序是单线程的，所以当您多次调用asyncio.get_event_loop()时，您总是会得到相同的事件循环。因此，您实际上并没有在示例中创建两个不同的循环；fmDaemon.run和Function1都使用同一个循环。这就是为什么在Function1中停止loop也会杀死你在run中启动的协程。

也就是说，没有理由一开始就试图创建两个不同的事件循环。Function1是从协程调用的，并且想要调用其他协程，所以为什么不把它也设为协程呢？然后，您可以直接调用yield from websockets.serve(...)，并使用asyncio.Event等待coroutine_overlay_websocket_server完成：

import webbrowser
import websockets
import asyncio

class fmDaemon( Daemon):
    def __init__( self, me):
        self.me = me

    def run( self):
        @asyncio.coroutine
        def coroutine_daemon_websocket_server(websocket, path):
            msg = yield from websocket.recv()

            if msg != None:
                msg_out = "{}".format( msg)
                yield from websocket.send( msg_out)
                self.me = Function1( self.me, msg)

        loop = asyncio.get_event_loop()
        loop.run_until_complete(websockets.serve(coroutine_daemon_websocket_server, 
                                                 self.me.IP, 
                                                 self.me.PORT))
        loop.run_forever()


@asyncio.coroutine
def Function1(me, msg):
    @asyncio.coroutine
    def coroutine_overlay_websocket_server(websocket, path):

        while True:
            msg = yield from websocket.recv()
            msg_out = "{}".format( msg)
            yield from websocket.send( msg_out)

            if msg == 'my_expected_string':
                me.flags['myStr'] = msg
                break

        event.set() # Tell the outer function it can exit.

    event = asyncio.Event()
    yield from websockets.serve(coroutine_overlay_websocket_server, me.IP, me.PORT_overlay))
    yield from event.wait() # This will block until event.set() is called.