多次调用asyncio.open_connection为何引发OSError？

当尝试用asyncio批量创建TCP客户端连接时，多次调用asyncio.open_connection会抛出OSError: [Errno 22] Invalid argument，单任务运行则正常，Linux和Mac系统都能复现。

最小复现代码 #

import asyncio


async def supervisor():
    async with asyncio.TaskGroup() as tg:
        for _ in range(20):
            tg.create_task(send())


async def send():
    _, writer = await asyncio.open_connection("localhost", 8888)
    writer.write(b"test")
    await writer.drain()


if __name__ == "__main__":
    asyncio.run(supervisor())

运行后错误输出 #

+ Exception Group Traceback (most recent call last):
  |   File "/tmp/repro.py", line 17, in <module>
  |     asyncio.run(supervisor())
  |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/runners.py", line 190, in run
  |     return runner.run(main)
  |            ^^^^^^^^^^^^^^^^
  |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/runners.py", line 118, in run
  |     return self._loop.run_until_complete(task)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 654, in run_until_complete
  |     return future.result()
  |            ^^^^^^^^^^^^^^^
  |   File "/tmp/repro.py", line 5, in supervisor
  |     async with asyncio.TaskGroup() as tg:
  |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/taskgroups.py", line 145, in __aexit__
  |     raise me from None
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/tmp/repro.py", line 11, in send
    |     _, writer = await asyncio.open_connection("localhost", 8888)
    |                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/streams.py", line 48, in open_connection
    |     transport, _ = await loop.create_connection(
    |                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 1113, in create_connection
    |     transport, protocol = await self._create_connection_transport(
    |                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 1143, in _create_connection_transport
    |     transport = self._make_socket_transport(sock, protocol, waiter)
    |                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/selector_events.py", line 61, in _make_socket_transport
    |     return _SelectorSocketTransport(self, sock, protocol, waiter,
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/selector_events.py", line 936, in __init__
    |     base_events._set_nodelay(self._sock)
    |   File "/opt/local/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 196, in _set_nodelay
    |     sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    | OSError: [Errno 22] Invalid argument
    +------------------------------------

问题原因 #

这个错误的核心是没有对应端口的监听服务端，客户端连接失败后，asyncio内部尝试对无效套接字设置TCP_NODELAY选项，触发Invalid argument错误。批量连接时，系统快速创建大量失败的套接字，这些套接字处于错误状态，调用setsockopt就会抛出异常；单任务场景下错误未被批量放大，不会触发底层报错。

另外代码存在资源泄漏问题：未关闭StreamWriter，批量场景下会加剧资源消耗，进一步触发异常。

解决方案 #

1. 启动对应端口的服务端：确保localhost:8888有TCP服务监听，让客户端能成功建立连接。
2. 正确关闭StreamWriter：用async with上下文管理器管理，或手动调用writer.close()并await writer.wait_closed()，释放资源。
3. 添加错误捕获：批量连接时捕获异常，避免单个连接失败导致整个TaskGroup崩溃。

修改后的示例代码：

import asyncio


async def supervisor():
    async with asyncio.TaskGroup() as tg:
        for _ in range(20):
            tg.create_task(send())


async def send():
    try:
        _, writer = await asyncio.open_connection("localhost", 8888)
        async with writer:
            writer.write(b"test")
            await writer.drain()
    except Exception as e:
        print(f"连接失败: {e}")


if __name__ == "__main__":
    asyncio.run(supervisor())

补充说明 #

你没有误解StreamWriter的使用，但必须保证连接成功后再执行写操作，且用完后关闭。批量创建连接时，系统的文件描述符、套接字资源有限，正确的资源管理和错误处理是避免异常的关键。

内容的提问来源于stack exchange，提问作者Bruce Johnson