如何停止在 Python 中的命名管道上阻塞的线程?
How to stop a thread which is blocking on a named pipe in Python?
我有一个 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()
?
执行此操作的经典方法是使用未命名的管道发出关闭信号,并使用 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')
我有一个 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()
?
执行此操作的经典方法是使用未命名的管道发出关闭信号,并使用 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')