python - 如何使用 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())