如何等待带有 asyncio 的套接字的错误/关闭事件？

Question

我正在使用一个网络库，它提供了一个包装器，用于将其协程函数与asyncio 一起使用。当我写了一个随机关闭连接的测试（看看我的程序在恶劣条件下是否有弹性）时，我发现它无限期挂起。

这似乎是我正在使用的库提供的包装器中的一个错误，因为程序挂起等待来自loop.add_reader() 或loop.add_writer() 的回调，但后来我找不到如何在套接字时收到通知已关闭。

这是一个显示我的程序发生了什么的最小程序：

import asyncio
import socket

async def kill_later(c):
    await asyncio.sleep(0.1)
    c.close()

async def main():
    loop = asyncio.get_running_loop()

    c = socket.create_connection(('www.google.com', 80))
    c.setblocking(0)

    ev = asyncio.Event()
    loop.add_reader(c, ev.set)

    # Closes the socket after 0.1 ms:
    asyncio.create_task(kill_later(c))

    print("waiting...")

    #### ↓ THIS WAITS FOREVER ↓ ####
    await ev.wait()

asyncio.run(main())

我的问题：如何通知套接字被asyncio 循环关闭？

编辑：由于大众的需求，使套接字成为非阻塞的，但这并没有什么区别，因为add_reader() 不会尝试在套接字上执行任何 IO，只是观察它何时准备好。

Answer 1

您的测试程序有缺陷。对c.close() 的调用不会模拟另一端关闭的套接字，它会关闭您自己的文件描述符并使其无法访问。您可以将close(fd) 视为断开数字 fd 和底层操作系统资源之间的链接。在那之后，读取和轮询 fd 变得毫无意义，因为这个数字不再代表任何东西。因此，epoll() 不能也不会将关闭的文件描述符报告为“可读”。

测试您要测试的条件的方法是让另一端关闭连接。最简单的方法是生成另一个进程或线程作为模拟服务器。例如：

import asyncio, threading, socket, time

def start_mock_server():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(('localhost', 10000))
    s.listen(1)
    def serve():
        conn, addr = s.accept()
        time.sleep(1)
        conn.close()
        s.close()
    threading.Thread(target=serve).start()

async def main():
    loop = asyncio.get_running_loop()
    start_mock_server()

    c = socket.create_connection(('localhost', 10000))
    c.setblocking(0)

    ev = asyncio.Event()
    loop.add_reader(c.fileno(), ev.set)

    print("waiting...")
    await ev.wait()
    print("done")

asyncio.run(main())

Answer 2

至少在您的示例中，问题是您的套接字永远不会收到任何东西，因此永远不会设置事件。 example 中提到的文档：

等到文件描述符使用 loop.add_reader() 方法，然后关闭事件循环

为了让您的示例正常工作，您必须先向 Google 发送请求：

c = socket.create_connection(('www.google.com', 80))
c.sendall("GET /\r\n".encode())

这将设置您的await 稍后在您的main coro 中的事件。

epoll 认为如果文件描述符上的数据可用或发出 EOF 信号，则文件描述符已准备好读取，如this answer 中所述。

add_writer() 在这种情况下不会阻塞，因为只要输入缓冲区上仍有可用空间，就认为文件描述符已准备好写入。

调用recv()，正如我在上一个版本的答案中提到的，不是必需的。