如何在 Python 3.7 中向 multiprocessing.connection.Client(..) 添加超时？

Posted 2023-02-25

【中文标题】如何在 Python 3.7 中向 multiprocessing.connection.Client(..) 添加超时？

【英文标题】：How do I add a timeout to multiprocessing.connection.Client(..) in Python 3.7?

【发布时间】：2020-01-09 02:12:58

【问题描述】：

我有两个 Python 程序正在运行。程序 A 使用 multiprocessing 模块连接到程序 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
)

它一直在等待响应。我想设置一个 timeout 参数，但对multiprocessing.connection.Client(..) 的调用没有。

如何在这里实现超时？

 注意事项： 我正在使用Windows 10 使用Python 3.7 的计算机。

【问题讨论】：

.Client is a function。 Timeout on a function call 的答案描述了许多方法（通过搜索可以找到其他类似的 SO Q&A）。许多方法都涉及到装饰器；如果一个有用，你可以用它包装/重新定义.Client - Client = timeout_decorator(...Client) 并使用包装的功能。

如果您认为您的问题与Timeout on a function call 重复，请发表评论，以便我们将其标记为此类。

嗨@wwii，函数调用超时确实是一个非常好的方法！非常感谢。但是，我注意到该问题的公认答案仅适用于 UNIX。您认为什么是 Windows 用户的最佳答案？

警告：除了玩我在 SO 上找到的东西外，我没有这方面的实际经验。我想我会尝试使用这个答案 - ***.com/a/31667005/2823755。它使用threading.Timer 来提升可能被捕获的KeyboardInterrupt。最好有一个最小的工作示例来玩：模仿您的程序 A 和 B 的东西，它们试图相互连接而没有其他任何东西 - 确保使用 multiprocessing.connection.Client 执行此操作不会导致任何不必要的副作用。

特别是-***.com/questions/34871191/…、***.com/questions/47847392/…。可能需要使用***.com/a/31667005/2823755 作为起点进行更详细的说明。

【参考方案1】：

我想设置一个超时参数，但是对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

【讨论】：

你测试了吗？如果可能并且不太长，您可以添加代码来测试它吗？

非常好，谢谢。希望 OP 能够使用它。

谢谢，这有助于加载！我稍微调整了一下，1）添加了一个发送超时（因为这也可以阻止），2）在s.connect之前设置两个选项（因为这是发送时的阻止功能）。您可以类似地设置发送超时：s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDTIMEO, timeval)