【问题标题】:How to stop a thread which is blocking on a named pipe in Python?如何停止在 Python 中的命名管道上阻塞的线程?
【发布时间】:2019-05-10 21:15:28
【问题描述】:

我有一个子类threading.Thread 的类。它的唯一职责就是将从 UNIX 命名管道读取的消息放入 queue.Queue 对象(以便其他线程稍后可以处理这些值)。

示例代码:

class PipeReaderThread(Thread):
    def __init__(self, results_queue, pipe_path):
        Thread.__init__(self)
        self._stop_event = Event()
        self._results_queue = results_queue
        self._pipe_path = pipe_path

    def run(self):
        while not self._stop_event.is_set():
            with open(self._pipe_path, 'r') as pipe:
                message = pipe.read()
            self._results_queue.put(message, block=True)

    def stop(self):
        self._stop_event.set()

如您所见,我想使用 threading.Event 对象来停止循环,但是由于命名管道上的 open()read() 调用将阻塞(直到有人打开管道以进行写入/写入然后关闭它),线程永远没有机会停止。

我不想对命名管道使用非阻塞模式,因为阻塞实际上是我想要的,从某种意义上说,我想等待有人打开并写入管道。

对于套接字,我会尝试在套接字上设置超时标志,但我找不到任何方法来为命名管道执行此操作。 我也考虑过只是冷血地杀死线程而不给它一个优雅地停止的机会,但这并不是我应该做的事情,我什至不知道 Python 是否提供任何这样做的方法.

我应该如何正确停止这个线程,以便之后我可以调用join()

【问题讨论】:

    标签: python-3.x multithreading pipe named-pipes blocking


    【解决方案1】:

    执行此操作的经典方法是使用未命名的管道发出关闭信号,并使用select 知道要使用哪个管道。

    select 将阻塞,直到其中一个描述符准备好读取,然后您可以使用os.read,在这种情况下不会阻塞。

    演示代码(不处理错误,可能泄漏描述符):

    class PipeReaderThread(Thread):
        def __init__(self, results_queue, pipe_path):
            Thread.__init__(self)
            self._stop_pipe_r, self._stop_pipe_w = os.pipe()
            self._results_queue = results_queue
            self._pipe = os.open(pipe_path, os.O_RDONLY) # use file descriptors directly to read file in parts
            self._buffer = b''
    
        def run(self):
            while True:
                result = select.select([self._stop_pipe_r, self._pipe], [], [])
                if self._stop_pipe_r in result[0]:
                    os.close(self._stop_pipe_r)
                    os.close(self._stop_pipe_w)
                    os.close(self._pipe)
                    return
                self._buffer += os.read(self._pipe, 4096) # select above guarantees read is noblocking
                self._extract_messages_from_buffer() # left as an exercise
    
        def stop(self):
            os.write(self._stop_pipe_w, b'c')
    

    【讨论】:

    • 应该想到这一点。我喜欢这种方法完全符合 Unix 哲学。谢谢。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2010-10-09
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多