3

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

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尚未发送。

有任何想法吗？

4

1 回答 1

6

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

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

s 表示 4 个三*重奏任务。该_connect任务位于托儿所的尽头，等待子任务完成。_watchdog任务被阻塞，任务被阻塞await self._end_session.wait()，任务被阻塞。_senderawait q.get()_receiverawait client_stream.receive_some(...)

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

  • 如果_watchdog's waitcall raises KeyboardInterrupt，则该_watchdog方法立即退出，因此它甚至不会尝试 send logout。然后作为展开堆栈的一部分，三重奏取消所有其他任务，一旦它们退出，然后KeyboardInterrupt继续传播直到它到达你的finally块connect。此时，您尝试使用 通知看门狗任务self._end_session.set()，但它不再运行，因此不会通知。

  • 如果_sender's q.get()call raises KeyboardInterrupt，则该_sender方法立即退出，因此即使_watchdog确实要求它发送注销消息，它也不会注意到。在任何情况下，三重奏都会继续取消看门狗和接收器任务，事情如上进行。

  • 如果_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 聊天频道。

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

于 2018-02-15T06:54:03.800 回答