在给定的multiprocessing.connection.Listener.accept()给定时间后引发TimeOutError

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了在给定的multiprocessing.connection.Listener.accept()给定时间后引发TimeOutError相关的知识,希望对你有一定的参考价值。

我试图打断multiprocessing.connection.Listener.accept(),但到目前为止还没有成功。因为它没有提供timeout参数,我想也许我可以使用socket.setdefaulttimeout()来打断它,正如帖子中所建议我再也找不到了,这里就是SO。

这没用。然后我尝试在close()对象上调用Listener()。根据this post's answer,这应该有效。

然而,似乎这些对象不能与通常的socket相关的解决方案一起发挥作用。

我可以确认ListenerTimer对象按预期关闭,但accept()呼叫不会中断。

代码:

import logging
import socket
import os
from multiprocessing.connection import Listener
from queue import Queue, Empty
from threading import Thread, Event, Timer

class Node(Thread):
    """Base Class providing a AF_INET, AF_UNIX or AF_PIPE connection to its
    data queue. It offers put() and get() method wrappers, and therefore
    behaves like a Queue as well as a Thread.

    Data from the internal queue is automatically fed to any connecting client.
    """
    def __init__(self, sock_name, max_q_size=None, timeout=None,
                 *thread_args, **thread_kwargs):
        """Initialize class.

        :param sock_name: UDS, TCP socket or pipe name
        :param max_q_size: maximum queue size for self.q, default infinite
        """
        self._sock_name = sock_name
        self.connector = Listener(sock_name)
        max_q_size = max_q_size if max_q_size else 0
        self.q = Queue(maxsize=max_q_size)
        self._running = Event()
        self.connection_timer = Timer(timeout, self.connection_timed_out)
        super(Node, self).__init__(*thread_args, **thread_kwargs)

    def connection_timed_out(self):
        """Closes the Listener and shuts down Node if no Client connected.

        :return:
        """
        self.connector.close()
        self.join()

    def _start_connection_timer(self):
        self.connection_timer.start()

    def start(self):
        self._running.set()
        super(Node, self).start()

    def join(self, timeout=None):
        print("clearing..")
        self._running.clear()
        print("internal join")
        super(Node, self).join(timeout=timeout)
        print("Done")

    def run(self):
        while self._running.is_set():
            print("Accepting connections..")
            self._start_connection_timer()
            try:
                client = self.connector.accept()
                self.connection_timer.cancel()
                self.feed_data(client)
            except (TimeoutError, socket.timeout):
                continue
            except Exception as e:
                raise
        print("Run() Terminated!")

    def feed_data(self, client):
        try:
            while self._running.is_set():
                try:
                    client.send(self.q.get())
                except Empty:
                    continue
        except EOFError:
            return


if __name__ == '__main__':
    import time
    n = Node('/home/nils/git/spab2/test.uds', timeout=10)
    n.start()
    print("Sleeping")
    time.sleep(15)
    print("Manual join")
    n.join()

我意识到我的问题是这个question的重复 - 然而,它已经差不多一年了,甚至没有收到评论。另外,我使用Unix Domain Sockets,而不是链接帖子的TCP连接。

答案

我设法在Python 2.7中以下列方式设置超时:

self.listener = mpc.Listener((address, port))
self.listener._listener._socket.settimeout(3)

这样,accept()呼叫被中断。

结果:

conn = self.listener.accept()
File "/usr/lib/python2.7/multiprocessing/connection.py", line 145, in accept
c = self._listener.accept()
File "/usr/lib/python2.7/multiprocessing/connection.py", line 275, in accept
s, self._last_accepted = self._socket.accept()
File "/usr/lib/python2.7/socket.py", line 202, in accept
sock, addr = self._sock.accept()
timeout: timed out

此致,亨利

以上是关于在给定的multiprocessing.connection.Listener.accept()给定时间后引发TimeOutError的主要内容,如果未能解决你的问题,请参考以下文章

算法:按字典顺序在给定索引处查找给定字符串的字谜

如何确定给定日期是不是在 .NET 2.0 中给定时区的夏令时?

PHP锁定/确保给定脚本在任何给定时间只运行一次

黄瓜订购给定,何时,然后(给定,何时,然后,何时,然后)

如何在不更改其他通道的情况下有效地将 cv::Mat 的给定通道设置为给定值?

在给定数字之后查找质数