快速检测或模拟 WSAECONNREFUSED答案

【问题标题】：Fast detection or simulation of WSAECONNREFUSED快速检测或模拟 WSAECONNREFUSED
【发布时间】：2014-01-16 16:51:42
【问题描述】：

当涉及到 WSAECONNREFUSED 时，Windows 套接字有一些奇怪的行为（这意味着积压已满或端口不可用，请参阅 https://stackoverflow.com/a/10308338/851737）。如果 Windows 检测到其中一种情况，它会以 0.5 秒的间隔重试（最多）两次。这意味着在套接字连接尝试 (http://support.microsoft.com/kb/175523/en-us) 上检测 WSAECONNREFUSED 至少需要 1 秒。

有什么方法可以在不影响注册表值的情况下加快检测速度？我需要在单元测试中模拟拒绝套接字连接。类似使用原始套接字模拟拒绝连接的解决方法也是可以接受的。

下面是一个简单的 Python 脚本来演示这个问题：

import errno
import socket
import time

PORT = 50123


def main():
    s = socket.socket()
    s.bind(('127.0.0.1', PORT))
    s.listen(0)
    client = socket.socket()
    client.connect(('127.0.0.1', PORT))

    client2 = socket.socket()
    start = time.time()

    try:
        client2.connect(('127.0.0.1', PORT))
    except socket.error as e:
        assert e.errno == errno.WSAECONNREFUSED
        print 'connection attempt took', time.time() - start
    finally:
        client2.close()
        client.close()
        s.close()


if __name__ == '__main__':
    main()

【问题讨论】：

标签： python windows sockets winsock

【解决方案1】：

这不是你问的。但如果您只在 unittests 中需要它，mock 库会很有用。

import errno
import socket
import time
import mock

PORT = 50123


def connect_mock(*agrs):
    raise socket.error(errno.WSAECONNREFUSED, "Testing")


def main():
    s = socket.socket()
    s.bind(('127.0.0.1', PORT))
    s.listen(0)
    client = socket.socket()
    client.connect(('127.0.0.1', PORT))

    client2 = socket.socket()
    start = time.time()

    with mock.patch('socket.socket.connect', connect_mock):
        try:
            client2.connect(('127.0.0.1', PORT))
            print "done"
        except socket.error as e:
            assert e.errno == errno.WSAECONNREFUSED
            print 'connection attempt took', time.time() - start
        finally:
            client2.close()
            client.close()
            s.close()


if __name__ == '__main__':
    main()

【讨论】：

实际上，我正在测试重新连接例程，因此第二次调用 connect 必须成功（因为我在注意到第一次连接尝试失败后启动服务器）。但这应该可以通过更智能的connect_mock 版本实现。我将对此进行测试并报告我的进度。感谢到目前为止的灵感。
啊，这在我的情况下非常有效 :) 你肯定赢得了赏金，但我将继续开放几天，因为可能会有更通用的解决方案。

【解决方案2】：

这是我基于 dmitry-vakhrushev 的answer 的解决方案，它更智能地修补了连接方法：

if sys.platform == 'win32':
    n_calls = [0]
    org_connect = socket.socket.connect

    def refusing_connect(*args):
        if n_calls[0] < 2:
            n_calls[0] += 1
            raise socket.error(errno.WSAECONNREFUSED, "Testing")
        return org_connect(*args)

    # patch socket.connect to speed up WSAECONNREFUSED detection
    patcher = mock.patch('socket.socket.connect', refusing_connect)
    patcher.start()
    self.addCleanup(patcher.stop)

【讨论】：