Python asyncio：关闭套接字并释放等待 sock_read()

Posted 2023-02-24

【中文标题】Python asyncio：关闭套接字并释放等待 sock_read()

【英文标题】：Python asyncio: closing a socket and releasing an await sock_read()

【发布时间】：2019-10-23 11:30:40

【问题描述】：

我正在编写一个小型多用户游戏。用户通过控制台或套接字登录。我希望能够踢出其他用户。

我使用asyncio 并通过调用await loop.sock_recv(sock, 256) 等待用户输入。现在，如果其他用户（例如，从控制台）关闭套接字，则事件会崩溃，因为 select.select 似乎有问题。

如何终止连接并释放sock_recv()？

附加的是一个小的（ish）MWE。它创建侦听套接字并接受端口 4000 上的连接。之后，您可以通过在控制台上输入 "x" 来终止连接。 logoff() 是我试图终止连接的尝试。

import asyncio
import socket
import sys
import threading


# ------ console input -------------------------------------------------------

async def _ainput(loop):
    fut = loop.create_future()
    def _run():
        line = sys.stdin.readline().strip()
        loop.call_soon_threadsafe(fut.set_result, line)
    threading.Thread(target=_run, daemon=True).start()
    return await fut

async def console_input_loop(loop):
    while True:
        inp = (await _ainput(loop)).strip()
        print(f"[inp.strip()]")
        if inp == "x":
            logoff()


# ------ socket input --------------------------------------------------------

alive = True

async def socket_input_loop(loop, sock):
    print(f"New connection")

    global alive
    while alive:
        try:
            inp = await loop.sock_recv(sock, 256)
        except ConnectionResetError:
            break
        print(inp)

    print("shutting down")
    sock.shutdown(socket.SHUT_RDWR)
    sock.close()
    print(f"Connection closed")
    alive = True

listen_addr = ('', 4000)

async def _run_server(loop, server):
    server.bind(listen_addr)
    server.listen(8)
    server.setblocking(False)

    while loop.is_running():
        global sock
        sock = (await loop.sock_accept(server))[0]
        loop.create_task(socket_input_loop(loop, sock))

async def run_server4(loop):
    await _run_server(loop, socket.socket(socket.AF_INET, socket.SOCK_STREAM))

async def run_server6(loop):
    await _run_server(loop, socket.socket(socket.AF_INET6, socket.SOCK_STREAM))

def async_driver():
    loop = asyncio.get_event_loop()
    loop.create_task(console_input_loop(loop))
    loop.create_task(run_server4(loop))
    loop.create_task(run_server6(loop))
    loop.run_forever()
    print()

def logoff():
    global alive
    alive = False
    #loop = asyncio.get_event_loop()
    #try:
    #    key = loop._selector.get_key(sock.fileno())
    #except KeyError:
    #    pass
    #else:
    #    mask, (reader, writer) = key.events, key.data
    #    #loop._add_callback(reader)
    # the next line is needed, otherwise we get:
    #     r, w, x = select.select(r, w, w, timeout)
    # OSError: [WinError 10038] An operation was attempted on something that is not a socket    
    #loop.remove_reader(sock.fileno())   
    sock.shutdown(socket.SHUT_RDWR)
    sock.close()
    #reader._run()

async_driver()

我遇到了这个崩溃：

Traceback (most recent call last):
  File "C:\xx7.py", line 91, in <module>
    async_driver()
  File "C:\xx7.py", line 69, in async_driver
    loop.run_forever()
  File "C:\Users\chris\Anaconda3\lib\asyncio\base_events.py", line 528, in run_forever
    self._run_once()
  File "C:\Users\chris\Anaconda3\lib\asyncio\base_events.py", line 1728, in _run_once
    event_list = self._selector.select(timeout)
  File "C:\Users\chris\Anaconda3\lib\selectors.py", line 323, in select
    r, w, _ = self._select(self._readers, self._writers, [], timeout)
  File "C:\Users\chris\Anaconda3\lib\selectors.py", line 314, in _select
    r, w, x = select.select(r, w, w, timeout)
OSError: [WinError 10038] An operation was attempted on something that is not a socket

【参考方案1】：

解决方案很简单：只使用shutdown() 而没有close()。