如何在python-trio中的KeyboardInterrupt之后清理连接

Question

我的班级在连接到服务器时应立即发送登录字符串，然后在会话结束后应发送退出字符串并清理套接字。下面是我的代码。

import trio

class test:

    _buffer = 8192
    _max_retry = 4

    def __init__(self, host='127.0.0.1', port=12345, usr='user', pwd='secret'):
        self.host = str(host)
        self.port = int(port)
        self.usr = str(usr)
        self.pwd = str(pwd)
        self._nl = b'\r\n'
        self._attempt = 0
        self._queue = trio.Queue(30)
        self._connected = trio.Event()
        self._end_session = trio.Event()

    @property
    def connected(self):
        return self._connected.is_set()

    async def _sender(self, client_stream, nursery):
        print('## sender: started!')
        q = self._queue
        while True:
            cmd = await q.get()
            print('## sending to the server:\n{!r}\n'.format(cmd))
            if self._end_session.is_set():
                nursery.cancel_scope.shield = True
                with trio.move_on_after(1):
                    await client_stream.send_all(cmd)
                nursery.cancel_scope.shield = False
            await client_stream.send_all(cmd)

    async def _receiver(self, client_stream, nursery):
        print('## receiver: started!')
        buff = self._buffer
        while True:
            data = await client_stream.receive_some(buff)
            if not data:
                print('## receiver: connection closed')
                self._end_session.set()
                break
            print('## got data from the server:\n{!r}'.format(data))

    async def _watchdog(self, nursery):
        await self._end_session.wait()
        await self._queue.put(self._logoff)
        self._connected.clear()
        nursery.cancel_scope.cancel()

    @property
    def _login(self, *a, **kw):
        nl = self._nl
        usr, pwd = self.usr, self.pwd
        return nl.join(x.encode() for x in ['Login', usr,pwd]) + 2*nl

    @property
    def _logoff(self, *a, **kw):
        nl = self._nl
        return nl.join(x.encode() for x in ['Logoff']) + 2*nl

    async def _connect(self):
        host, port = self.host, self.port
        print('## connecting to {}:{}'.format(host, port))
        try:
            client_stream = await trio.open_tcp_stream(host, port)
        except OSError as err:
            print('##', err)
        else:
            async with client_stream:
                self._end_session.clear()
                self._connected.set()
                self._attempt = 0
                # Sign in as soon as connected
                await self._queue.put(self._login)
                async with trio.open_nursery() as nursery:
                    print("## spawning watchdog...")
                    nursery.start_soon(self._watchdog, nursery)
                    print("## spawning sender...")
                    nursery.start_soon(self._sender, client_stream, nursery)
                    print("## spawning receiver...")
                    nursery.start_soon(self._receiver, client_stream, nursery)

    def connect(self):
        while self._attempt <= self._max_retry:
            try:
                trio.run(self._connect)
                trio.run(trio.sleep, 1)
                self._attempt += 1
            except KeyboardInterrupt:
                self._end_session.set()
                print('Bye bye...')
                break

tst = test()
tst.connect()

我的逻辑不太行。好吧，如果我杀死netcat侦听器，它会起作用，所以我的会话如下所示：

## connecting to 127.0.0.1:12345
## spawning watchdog...
## spawning sender...
## spawning receiver...
## receiver: started!
## sender: started!
## sending to the server:
b'Login\r\nuser\r\nsecret\r\n\r\n'

## receiver: connection closed
## sending to the server:
b'Logoff\r\n\r\n'

请注意，Logoff字符串已发送出去，尽管在这里没有意义，因为那时连接已经中断。

但是我的目标是Logoff当用户KeyboardInterrupt. 在这种情况下，我的会话看起来类似于：

## connecting to 127.0.0.1:12345
## spawning watchdog...
## spawning sender...
## spawning receiver...
## receiver: started!
## sender: started!
## sending to the server:
b'Login\r\nuser\r\nsecret\r\n\r\n'

Bye bye...

请注意，Logoff尚未发送。

有任何想法吗？

Answer 1

这里你的调用树看起来像：

connect
|
+- _connect*
   |
   +- _watchdog*
   |
   +- _sender*
   |
   +- _receiver*

在*小号指示4级三人的任务。该_connect任务是坐在幼儿园块结束，等待子任务来完成。该_watchdog任务被阻止await self._end_session.wait()，该_sender任务被阻止await q.get()，并且_receiver任务被阻止await client_stream.receive_some(...)。

当你点击 control-C 时，标准的 Python 语义是任何正在运行的 Python 代码都会突然引发KeyboardInterrupt. 在这种情况下，您有 4 个不同的任务在运行，因此这些被阻止的操作之一被随机选取 [1]，并引发KeyboardInterrupt. 这意味着可能会发生一些不同的事情：

• 如果_watchdog的wait调用引发KeyboardInterrupt，则该_watchdog方法立即退出，因此它甚至从不尝试发送logout。然后作为展开堆栈的一部分，trio 取消所有其他任务，一旦它们退出，就会KeyboardInterrupt继续传播，直到它到达您finally的connect. 此时，您尝试使用 通知看门狗任务self._end_session.set()，但它不再运行，因此它不会注意到。

• 如果_sender的q.get()调用产生了KeyboardInterrupt，那么_sender马上方法退出，所以即使_watchdog没有要求它发送注销消息，它不会在那里的通知。在任何情况下，trio 都会继续取消看门狗和接收器任务，事情如上进行。

• 如果_receiver's receive_allcall raise KeyboardInterrupt...同样的事情发生。

• 细微之处：_connect也可以接收KeyboardInterrupt，它做同样的事情：取消所有孩子，然后等待他们停止，然后让KeyboardInterrupt继续传播。

如果你想可靠地捕获 control-C 然后用它做一些事情，那么在某个随机点提出它的业务是相当麻烦的。最简单的方法就是利用Trio 对捕获信号的支持来捕获signal.SIGINT信号，这也是 Python 通常转换成KeyboardInterrupt. （“INT”代表“中断”。）类似于：

async def _control_c_watcher(self):
    # This API is currently a little cumbersome, sorry, see
    # https://github.com/python-trio/trio/issues/354
    with trio.catch_signals({signal.SIGINT}) as batched_signal_aiter:
        async for _ in batched_signal_aiter:
            self._end_session.set()
            # We exit the loop, restoring the normal behavior of
            # control-C. This way hitting control-C once will try to
            # do a polite shutdown, but if that gets stuck the user
            # can hit control-C again to raise KeyboardInterrupt and
            # force things to exit.
            break

然后开始与您的其他任务一起运行。

您还有一个问题，在您的_watchdog方法中，它将logoff请求放入队列 - 从而安排稍后由_sender任务发送的消息- 然后立即取消所有任务，因此该_sender任务可能不会有机会查看消息并对其做出反应！一般来说，我发现只有在必要时才使用任务时，我的代码工作得更好。与其有一个发送者任务，然后在想要发送消息时将消息放入队列，为什么不拥有想要stream.send_all直接发送消息调用的代码呢？您必须注意的一件事是，如果您有多个可能同时发送事物的任务，您可能需要使用 atrio.Lock()来确保它们不会因同时调用而相互碰撞send_all：

async def send_all(self, data):
    async with self.send_lock:
        await self.send_stream.send_all(data)

async def do_logoff(self):
    # First send the message
    await self.send_all(b"Logoff\r\n\r\n")
    # And then, *after* the message has been sent, cancel the tasks
    self.nursery.cancel()

如果您这样做，您可能能够完全摆脱看门狗任务和_end_session事件。

当我在这里时，还有一些关于您的代码的其他注意事项：

• trio.run像这样多次调用是不寻常的。通常的风格是在程序顶部调用一次，然后将所有真实代码放入其中。一旦退出trio.run，三重奏的所有状态都将丢失，您绝对没有运行任何并发任务（因此不可能有任何事情可能正在侦听并注意到您对_end_session.set()!的调用）。一般来说，几乎所有 Trio 函数都假设您已经在调用trio.run. 事实证明，现在您可以trio.Queue()在开始三重奏之前调用而不会出现异常，但这基本上只是巧合。

• 内部屏蔽的使用_sender对我来说看起来很奇怪。屏蔽通常是您几乎从不想使用的高级功能，我认为这也不例外。

希望有帮助！如果您想更多地讨论这样的样式/设计问题，但又担心它们对于堆栈溢出可能过于模糊（“这个程序设计得很好吗？”），那么请随时访问trio chat channel。

[1] 嗯，实际上三人组可能出于各种原因选择主要任务，但这并不能保证，无论如何它在这里没有任何区别。