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

Question

我有两个 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.

Answer 1

我想设置一个超时参数，但调用multiprocessing.connection.Client(..)没有一个。我如何在这里实现超时？

查看Python 3.7 中 multiprocessing.connection的源代码，该Client()函数是SocketClient()您用例的一个相当简短的包装器，它反过来包装Connection().

起初，编写一个ClientWithTimeout做同样事情的包装器看起来相当简单，但另外调用settimeout()它为连接创建的套接字。但是，这不会产生正确的效果，因为：

1. Python 通过使用select()和底层非阻塞 OS 套接字来实现自己的套接字超时行为；这种行为是由settimeout().

2. Connection直接在 OS 套接字句柄上操作，该句柄通过调用detach()普通 Python 套接字对象返回。

3. 由于 Python 已将 OS 套接字句柄设置为非阻塞模式，因此对其recv()调用会立即返回，而不是等待超时时间。

但是，我们仍然可以使用低级SO_RCVTIMEO套接字选项在底层操作系统套接字句柄上设置接收超时。

因此，我的解决方案的第二个版本：

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

为简洁起见，我假设参数与您的示例相同，即：

• address 是一个元组（暗示地址族是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