如何在Python中停止阻塞在命名管道上的线程?

8

我有一个类,它是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()呢?

1个回答

8

传统的方法是使用未命名管道来发出关闭信号,并使用 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哲学。谢谢。 - krispet krispet

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接