连接诊断器：并发尝试与安全取消

目标

构建一个诊断工具：在有限的预算内推进多个 TCP 连接尝试，区分拒绝、截止时间与取消，并回收所有资源。

这是 90 分钟的实验。默认会话为 60 分钟，所以请在到期之前用“＋时间”按钮延长（最长 180 分钟）。会话结束后文件会消失。重要的成果请在结束之前另行保存。

为什么重要

只看连接成功的诊断工具会掩盖失败原因，无法取消的工具则会迫使用户强制终止。本实验把前面的公平性、SO_ERROR 和控制通道，连接到同一个循环中。真实通信只使用实验环境的 loopback，不检查外部服务器。数据传递、HTTP、TLS、重试、DNS 以及服务器的 graceful shutdown，都不在本实验的范围之内。这是基于 DefaultSelector 的完成处理，而不是 ET 读取服务器。

检查会导入源文件来测试其行为。不要在文件最上层启动服务器或等待输入。不需要额外的软件包和权限。/opt/fixtures/reactor 中的三个 probe.py 用于单独观察概念，不是需要修改或提交的材料文件。

步骤

定义连接尝试的输入与初始状态——在 /root/reactor/client.py 中实现 Dial(sock, host, port, deadline=None)。输入 sock 是调用方打开的、互不相同的 Linux AF_INET/SOCK_STREAM 套接字。host 只接受数字 IPv4 字符串，port 只接受不含 bool 的 1–65535 int，违反则为 ValueError。deadline=None 时为 time.monotonic()+30，指定值必须是不含 bool 的有限非负 int/float。把 sock 改为非阻塞，并保存 sock、endpoint=(规范化后的 host, port)、float 类型的 deadline、state='new'、error=None。创建失败时，套接字的回收是调用方的责任。
区分立即成功与连接进行中的状态——在 /root/reactor/client.py 中，Dial.start() 只在 new 状态下调用一次 sock.connect_ex(endpoint)。0/EISCONN 为 connected、error=0，EINPROGRESS/EALREADY 为 pending、error=None，其他代码为 failed、error=该代码。OSError 记录 errno，没有 errno 时为 EIO。如果不是 new，则不调用而直接返回已有的 state。所有路径都返回 state 字符串，在这一 Linux IPv4 契约中，EAGAIN 不属于 pending。
保存最初的 SO_ERROR——在 /root/reactor/client.py 中，Dial.writable() 只在 pending 时读取一次 getsockopt(SOL_SOCKET, SO_ERROR)，0/EISCONN 记为 connected、0，其他代码记为 failed、对应的代码。OSError 以 errno 或 EIO 按失败处理。如果不是 pending，则不查询。返回当前的 state。调用方只有在观察到 WRITE 就绪之后，才能对 pending 调用这个方法。同时也要检查真实的 TCP listen 端口，以及只做了 bind 的拒绝端口。
保存队列截止时间与取消原因——在 /root/reactor/client.py 中，Dial.expire(now) 只在 new/pending 且 now >= deadline 时，改为 timed_out、ETIMEDOUT。now 是调用方提供的有限单调时刻。Dial.cancel() 只把 new/pending 改为 cancelled、ECANCELED。这两个方法不会覆盖已经终止的状态和错误，也不会直接关闭套接字。也不延长 deadline。
构建不重复的就绪队列——在 /root/reactor/client.py 中，为 ReadyQueue() 实现 push(item)、pop()、discard(item) 和 len。item 是不为 None 的可哈希连接标识符。push 在已存在时忽略，pop 按 FIFO 移除并返回、为空时返回 None，discard 要移除对应的条目，且条目不存在时也必须安全。取出的条目可以再次添加，相同 FD 的不同代次的标识符要区分开。在 run 中，用 Dial 对象本身作为键。
记录停止状态并唤醒内核等待——在 /root/reactor/client.py 中，Control() 创建非阻塞 socketpair 的 reader/writer，以及最初为 False 的 threading.Event stop_event。request_stop() 在 Event.set 之后调用一次 writer.send(b'Q')，只忽略 BlockingIOError。drain() 调用一次 reader.recv(4096) 并返回 bytes，其中 BlockingIOError 返回 b''，EOF 为 ConnectionError。close() 即使在关闭 reader 时出现异常，也要关闭 writer。run 结束之后调用 request_stop 属于契约之外。
遵守上限与预算，回收所有套接字——在 /root/reactor/client.py 中实现 run(dials, control, limit=8, budget=4, ready=None)。dials 是由持有互不相同的已打开套接字的、互不相同的新 Dial 构成的 list（最多 128 个），limit 是不含 bool 的 1–128 int，budget 是不含 bool 的 1–64 int。要验证列表类型、个数、Dial 是否重复、是否为 new 状态以及两个预算，违反则以 ValueError 拒绝，此时所有权仍属于调用方。一旦成功创建 DefaultSelector，函数就接管所有数据套接字和 Control。把控制 reader 以 READ/data=None 注册，如果有 ready 则调用一次。每一轮先检查停止，并对 new/pending 应用截止时间。如果已停止，就取消全部未完成的项并结束。每一轮最多启动 budget 个，并把最多 limit 个 pending 以 WRITE/data=Dial 注册。立即终止的条目要回收。等待时间取距最近截止时间的时长与 1 秒中较小的值（最小为 0），如果就绪队列还有剩余，或等待列表中还有可启动的余量，则为 0。select 结果中的控制事件做 drain，WRITE 则放入不重复的就绪队列。再次检查停止之后，从队列中最多取出 budget 个，在 I/O 之前检查截止时间，然后对 pending 调用 writable。已终止的条目要从队列中删除、unregister 之后再 close。即使出现异常，也必须回收其他数据套接字、Control 乃至 selector，并传播错误。不要让某一次回收的失败妨碍其余的回收。空列表也要回收 Control 并返回 []。正常终止时，返回按输入顺序的 {'state': 状态, 'error': 整数} 的 list。诊断连接即使成功也要关闭，不会复用。
无损地汇总成功与失败原因——在 /root/reactor/client.py 中，summarize(results) 只接收由 dict 构成的 list，违反则为 ValueError。每一行的 state 必须是 connected/failed/timed_out/cancelled，error 必须是不含 bool 的 int。connected 为 0，其他状态为正数，timed_out 必须是 ETIMEDOUT，cancelled 必须是 ECANCELED。pending 等未完成状态也以 ValueError 拒绝。返回 total、四种状态各自的数量以及 errors 字典。errors 以 connected 之外的 errno 的字符串为键进行累加，并对键排序。空输入为 total 与四种状态均为 0，且 errors={}。不修改原列表和各行。这一步也会重新检查前面的真实 TCP、截止时间、取消、预算和异常回收。

参考

请先用 mkdir -p /root/reactor 创建工作目录。格式示例中的 NotImplementedError 不是完整答案。
可以像 python3 /opt/fixtures/reactor/check.py 7 /root/reactor/client.py 这样直接运行累计检查。步骤数字为 1–8。
除了真实的 TCP 成功与拒绝之外，还通过合成的连接状态和真实 socketpair 的写等待来验证截止时间和取消。不要把它解读为重现了远程网络故障。
limit 是 pending 的上限，并不是预先打开的套接字总数。如果初始化失败或收到无效参数，调用方必须清理资源。
使用“对照答案操作”会覆盖累计的源代码，所以请先保存自己的代码。必须能亲自说明取消注册与 close 的顺序，以及所有异常路径，才算完成。

定义连接尝试的输入与初始状态

用 ipaddress.IPv4Address 验证地址，而不插入名称解析。注意 bool 是 int 的子类型。

区分立即成功与连接进行中的状态

在等待写就绪期间，不要反复重新连接。一个 Dial 就是一次连接尝试。

保存最初的 SO_ERROR

SO_ERROR 在读取时会被清除。不要用下一次查询得到的 0 覆盖已经确定的失败。

保存队列截止时间与取消原因

在队列中停留的时间也计入总生存时长。状态终止与真实的 FD 回收是不同的步骤。

构建不重复的就绪队列

要同时管理 deque 和 set 的添加与移除。不要让一个任务的重复位置抢走其他连接的轮次。

记录停止状态并唤醒内核等待

先记录停止状态，并让通知可以被合并。不要把一个通知字节解读为一条单独的用户命令。

遵守上限与预算，回收所有套接字

分别统计连接启动预算与完成处理预算。在完成事件集中涌现的测试中，两项限制都必须得到遵守。

无损地汇总成功与失败原因

0 是成功的观察结果，而不是空信息。要区分“汇总已完成”这一事实与整个生产服务的健康状况。