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

Question

我有一个 class 子 class 是 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()？

Answer 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')

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

How to stop a thread which is blocking on a named pipe in Python?

python-3.x

multithreading

pipe

named-pipes

blocking