python - 如何在 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

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()在它为连接创建的套接字上。但是，这并没有正确的效果，因为:

Python 通过使用 select() 实现其自己的套接字超时行为和一个底层的非阻塞操作系统套接字；此行为是由 settimeout() 配置的。 .

Connection直接对操作系统套接字句柄进行操作，该句柄通过调用 detach() 返回在普通的 Python 套接字对象上。

由于 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

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

address 是一个元组(暗示地址族是 AF_INET )。

authkey 是一个字节串。

如果您需要处理这些假设不成立的情况，那么您需要从 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