我有两个 Python 程序正在运行。程序 A 通过 连接到程序 B多处理 模块:
# Connection code in program A
# -----------------------------
import multiprocessing
import multiprocessing.connection
...
connection = multiprocessing.connection.Client(
('localhost', 19191), # <- address of program B
authkey='embeetle'.encode('utf-8') # <- authorization key
)
...
connection.send(send_data)
recv_data = connection.recv()
它在大多数情况下都能完美运行。但是,有时程序 B 会被卡住(细节并不重要,但通常会在程序 B 的 GUI 生成模式窗口时发生)。
当程序 B 被卡住时,程序 A 在以下行挂起:
connection = multiprocessing.connection.Client(
('localhost', 19191), # <- address of program B
authkey='embeetle'.encode('utf-8') # <- authorization key
)
它一直在等待响应。我想放一个超时参数,但调用
multiprocessing.connection.Client(..)
没有。我怎样才能在这里实现超时?
备注:
我正在处理
Windows 10
电脑与 Python 3.7
.
最佳答案
I would like to put a timeout parameter, but the call to
multiprocessing.connection.Client(..)
does not have one. How can I implement a timeout here?
看着source to multiprocessing.connection in Python 3.7 ,
Client()
函数是对 SocketClient()
的一个相当简短的包装。对于您的用例,它反过来包装 Connection()
.起初,写一个
ClientWithTimeout
看起来相当简单。做同样事情的包装器,但另外调用 settimeout()
在它为连接创建的套接字上。但是,这并没有正确的效果,因为:select()
实现其自己的套接字超时行为和一个底层的非阻塞操作系统套接字;此行为是由 settimeout()
配置的。 . Connection
直接对操作系统套接字句柄进行操作,该句柄通过调用 detach()
返回在普通的 Python 套接字对象上。 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
)。 如果您需要处理这些假设不成立的情况,那么您需要从
Client()
复制更多逻辑。和 SocketClient()
.虽然我看了
multiprocessing.connection
要了解如何执行此操作,我的解决方案不使用任何私有(private)实现细节。 Connection
, answer_challenge
和 deliver_challenge
都是 API 的公开和文档化部分。因此,该函数应该可以安全地用于 multiprocessing.connection
的 future 版本。 .请注意
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
关于python - 如何在 Python 3.7 中向 multiprocessing.connection.Client(..) 添加超时?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57817955/