我想设置一个超时参数，但是对multiprocessing.connection.Client(..) 的调用没有。如何在这里实现超时？
查看source to multiprocessing.connection in Python 3.7，Client() 函数是您的用例对SocketClient() 的一个相当简短的包装，而后者又包装了Connection()。
起初，编写一个做同样事情的ClientWithTimeout 包装器看起来相当简单，但另外在它为连接创建的套接字上调用settimeout()。但是，这并没有正确的效果，因为：
Python 通过使用select() 和底层非阻塞操作系统套接字来实现自己的套接字超时行为；这是settimeout()配置的行为。
Connection 直接对 OS 套接字句柄进行操作，该句柄通过在普通 Python 套接字对象上调用 detach() 返回。
由于 Python 已将 OS 套接字句柄设置为非阻塞模式，因此对其进行的recv() 调用会立即返回，而不是等待超时时间。
但是，我们仍然可以通过使用低级 SO_RCVTIMEO 套接字选项在底层 OS 套接字句柄上设置接收超时。
因此我的解决方案的第二个版本：
from multiprocessing.connection import Connection, answer_challenge, deliver_challenge
import socket, struct
def ClientWithTimeout(address, authkey, timeout):
with socket.socket(socket.AF_INET) as s:
s.setblocking(True)
s.connect(address)
# We'd like to call s.settimeout(timeout) here, but that won't work.
# Instead, prepare a C "struct timeval" to specify timeout. Note that
# these field sizes may differ by platform.
seconds = int(timeout)
microseconds = int((timeout - seconds) * 1e6)
timeval = struct.pack("@LL", seconds, microseconds)
# And then set the SO_RCVTIMEO (receive timeout) option with this.
s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVTIMEO, timeval)
# Now create the connection as normal.
c = Connection(s.detach())
# The following code will now fail if a socket timeout occurs.
answer_challenge(c, authkey)
deliver_challenge(c, authkey)
return c
为简洁起见，我假设参数与您的示例相同，即：
- 地址是一个元组（暗示地址族是
AF_INET）。
- authkey 是一个字节串。
如果您需要处理这些假设不成立的情况，那么您需要从Client() 和SocketClient() 复制更多逻辑。
虽然我查看了multiprocessing.connection 源以了解如何执行此操作，但我的解决方案没有使用任何私有实现细节。 Connection、answer_challenge 和 deliver_challenge 都是 API 的公开和文档化部分。因此，该函数应该可以安全地用于 multiprocessing.connection 的未来版本。
请注意，SO_RCVTIMEO 可能并非在所有平台上都受支持，但它至少存在于 Windows、Linux 和 OSX 上。 struct timeval 的格式也是特定于平台的。我假设这两个字段始终是本机 unsigned long 类型。我认为这在通用平台上应该是正确的，但不能保证总是如此。不幸的是，Python 目前不提供独立于平台的方式来做到这一点。
下面是一个测试程序，它显示了这个工作 - 它假设上面的代码保存为client_timeout.py。
from multiprocessing.connection import Client, Listener
from client_timeout import ClientWithTimeout
from threading import Thread
from time import time, sleep
addr = ('localhost', 19191)
key = 'embeetle'.encode('utf-8')
# Provide a listener which either does or doesn't accept connections.
class ListenerThread(Thread):
def __init__(self, accept):
Thread.__init__(self)
self.accept = accept
def __enter__(self):
if self.accept:
print("Starting listener, accepting connections")
else:
print("Starting listener, not accepting connections")
self.active = True
self.start()
sleep(0.1)
def run(self):
listener = Listener(addr, authkey=key)
self.active = True
if self.accept:
listener.accept()
while self.active:
sleep(0.1)
listener.close()
def __exit__(self, exc_type, exc_val, exc_tb):
self.active = False
self.join()
print("Stopped listener")
return True
for description, accept, name, function in [
("ClientWithTimeout succeeds when the listener accepts connections.",
True, "ClientWithTimeout", lambda: ClientWithTimeout(addr, timeout=3, authkey=key)),
("ClientWithTimeout fails after 3s when listener doesn't accept connections.",
False, "ClientWithTimeout", lambda: ClientWithTimeout(addr, timeout=3, authkey=key)),
("Client succeeds when the listener accepts connections.",
True, "Client", lambda: Client(addr, authkey=key)),
("Client hangs when the listener doesn't accept connections (use ctrl-C to stop).",
False, "Client", lambda: Client(addr, authkey=key))]:
print("Expected result:", description)
with ListenerThread(accept):
start_time = time()
try:
print("Creating connection using %s... " % name)
client = function()
print("Client created:", client)
except Exception as e:
print("Failed:", e)
print("Time elapsed: %f seconds" % (time() - start_time))
print()
在 Linux 上运行它会产生以下输出：
Expected result: ClientWithTimeout succeeds when the listener accepts connections.
Starting listener, accepting connections
Creating connection using ClientWithTimeout...
Client created: <multiprocessing.connection.Connection object at 0x7fad536884e0>
Time elapsed: 0.003276 seconds
Stopped listener
Expected result: ClientWithTimeout fails after 3s when listener doesn't accept connections.
Starting listener, not accepting connections
Creating connection using ClientWithTimeout...
Failed: [Errno 11] Resource temporarily unavailable
Time elapsed: 3.157268 seconds
Stopped listener
Expected result: Client succeeds when the listener accepts connections.
Starting listener, accepting connections
Creating connection using Client...
Client created: <multiprocessing.connection.Connection object at 0x7fad53688c50>
Time elapsed: 0.001957 seconds
Stopped listener
Expected result: Client hangs when the listener doesn't accept connections (use ctrl-C to stop).
Starting listener, not accepting connections
Creating connection using Client...
^C
Stopped listener