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

问题描述 投票：0回答：2

我正在运行两个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
计算机工作。

python python-3.x sockets multiprocessing python-3.7
2个回答
9
投票

我想设置一个超时参数，但是对

multiprocessing.connection.Client(..)
的调用没有超时参数。我如何在这里实现超时？

查看 Python 3.7 中 multiprocessing.connection 的 源，

Client()
函数是针对您的用例的 
SocketClient()
的一个相当简短的包装器，而它又包装了 
Connection()
。

起初，编写一个执行相同操作的 

ClientWithTimeout
包装器看起来相当简单，但另外在它为连接创建的套接字上调用 
settimeout()
。然而，这并没有达到正确的效果，因为：

  1. Python 通过使用 

    select()
    和底层非阻塞操作系统套接字来实现自己的套接字超时行为；此行为是由 
    settimeout()
    配置的。

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

  3. 由于Python已将操作系统套接字句柄设置为非阻塞模式，因此

    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

为了简洁起见，我假设参数按照您的示例，即：

  • 地址是一个元组（意味着地址族是
    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
0
投票

另请参阅https://github.com/python/cpython/issues/76425

Python 3.12 还没有解决

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.