pipe(7)
Linux 手册页面 指定管道具有有限的容量(默认为 65,536 字节),并且在向已满的管道写入数据时会阻塞,直到从管道中读取足够的数据以允许写入完成为止:
管道和FIFO上的I/O
[...]
如果一个进程试图从空管道中读取数据,那么read(2)
将会阻塞直到有数据可用。如果一个进程试图向满管道中写入数据(见下文),那么write(2)
将会阻塞直到足够的数据被从管道中读取以允许写入完成。可以通过使用fcntl(2)
的F_SETFL
操作来启用O_NONBLOCK
打开文件状态标志来实现非阻塞I/O。
[...]
管道容量
管道具有有限的容量。如果管道已满,则write(2)
将会阻塞或失败,这取决于是否设置了O_NONBLOCK
标志(见下文)。不同的实现对管道容量有不同的限制。应用程序不应该依赖于特定的容量:应该设计一个读取进程在数据可用时立即消耗数据的应用程序,以便写入进程不会保持阻塞。
在Linux 2.6.11之前的版本中,管道的容量与系统页面大小相同(例如,在i386上为4096字节)。自Linux 2.6.11以来,管道容量为16页(即,在页面大小为4096字节的系统中为65536字节)。自Linux 2.6.35以来,默认管道容量为16页,但可以使用fcntl(2)
的F_GETPIPE_SZ
和F_SETPIPE_SZ
操作查询和设置容量。有关更多信息,请参见fcntl(2)
。
这就是为什么
multiprocessing
Python库文档建议在生产者进程中使用
Queue.join_thread
调用(隐式地进行垃圾回收或显式地进行)之前,通过
Queue.get
调用使每个
Queue
对象的消费者进程为空:
Joining processes that use queues
Bear in mind that a process that has put items in a queue will wait before terminating until all the buffered items are fed by the “feeder” thread to the underlying pipe. (The child process can call the Queue.cancel_join_thread
method of the queue to avoid this behaviour.)
This means that whenever you use a queue you need to make sure that all items which have been put on the queue will eventually be removed before the process is joined. Otherwise you cannot be sure that processes which have put items on the queue will terminate. Remember also that non-daemonic processes will be joined automatically.
An example which will deadlock is the following:
from multiprocessing import Process, Queue
def f(q):
q.put('X' * 1000000)
if __name__ == '__main__':
queue = Queue()
p = Process(target=f, args=(queue,))
p.start()
p.join()
obj = queue.get()
A fix here would be to swap the last two lines (or simply remove the p.join()
line).
在某些应用程序中,消费者进程可能不知道生产者进程已经向队列添加了多少项。在这种情况下,一种可靠的清空队列的方法是让每个生产者进程在完成时添加一个哨兵项,并让消费者进程删除项目(常规和哨兵项),直到它删除了与生产者进程数量相同的哨兵项为止:
import multiprocessing
def f(q, e):
while True:
q.put('X' * 1000000)
if e.is_set():
break
q.put(None)
if __name__ == '__main__':
start_count = 5
stop_count = 0
q = multiprocessing.Queue()
e = multiprocessing.Event()
for _ in range(start_count):
multiprocessing.Process(target=f, args=(q, e)).start()
e.set()
while stop_count < start_count:
if q.get() is None:
stop_count += 1
此解决方案使用 阻塞式 的 Queue.get
调用来清空队列。这可以确保 所有 项目都已添加到队列并已被移除。
@DanH 的解决方案 使用 非阻塞式 的 Queue.get_nowait
调用来清空队列。该解决方案的问题在于,生产者进程仍然可以在消费者进程清空队列后向队列添加项目,这将导致死锁(消费者进程将等待生产者进程终止,每个生产者进程将等待其馈线程终止,每个生产者进程的馈线程将等待消费者进程移除添加到队列中的项目):
import multiprocessing.queues
def f(q):
q.put('X' * 1000000)
if __name__ == '__main__':
q = multiprocessing.Queue()
p = multiprocessing.Process(target=f, args=(q,))
p.start()
try:
while True:
q.get_nowait()
except multiprocessing.queues.Empty:
pass
p.join()
或者,如果与其作为属性一起提供的队列的同步资源在垃圾回收之前被清除,新创建的生产者进程可能无法反序列化消费者进程的Process
对象,从而引发FileNotFoundError
异常:
import multiprocessing.queues
def f(q):
q.put('X' * 1000000)
if __name__ == '__main__':
q = multiprocessing.Queue()
multiprocessing.Process(target=f, args=(q,)).start()
try:
while True:
q.get_nowait()
except multiprocessing.queues.Empty:
pass
标准错误:
Traceback (most recent call last):
File "<string>", line 1, in <module>
File "/usr/local/Cellar/python@3.9/3.9.12/Frameworks/Python.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 116, in spawn_main
exitcode = _main(fd, parent_sentinel)
File "/usr/local/Cellar/python@3.9/3.9.12/Frameworks/Python.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 126, in _main
self = reduction.pickle.load(from_parent)
File "/usr/local/Cellar/python@3.9/3.9.12/Frameworks/Python.framework/Versions/3.9/lib/python3.9/multiprocessing/synchronize.py", line 110, in __setstate__
self._semlock = _multiprocessing.SemLock._rebuild(*state)
FileNotFoundError: [Errno 2] No such file or directory
import
语句会相互覆盖。如果你使用上面写的代码,multi_q
和normal_q
都将成为常规的Queue.Queue
实例。为了使其工作,你需要只导入模块并使用完全限定名称来调用类(例如:multi_q = multiprocessing.Queue()
),或者使用as
关键字在导入时为它们命名(例如:from Queue import Queue as qQueue
)。 - Blckknght