基本上我有以下代码:
import multiprocessing
import time
class MyProcess(multiprocessing.Process):
def __init__(self, ):
multiprocessing.Process.__init__(self)
self.queue = multiprocessing.Queue()
def run(self):
print "Subprocess starting!"
time.sleep(4)
print "Subprocess exiting!"
def addToQueue(self):
starttime = time.time()
count=0
print "Adding stuff to queue..."
while time.time()-starttime < 4:
self.queue.put("string")
count += 1
print "Added %d objects!" % count
#self.queue.close()
if __name__ == "__main__":
process = MyProcess()
process.start()
print "Waiting for a while"
time.sleep(2)
process.addToQueue()
time.sleep(1)
print "Child process state: %d" % process.is_alive()
当主进程完成时,它不会退出。什么也没有发生,它只是阻塞。我发现唯一的退出方法是杀死它(不是 SIGTERM、SIGKILL)。
如果我使用该注释行,它会退出但会发出 IOError。
我查看了 multiprocessing.queue 的代码,它使用在另一个线程 (threading.Thread) 中生成的 os.pipe()。我怀疑线程在写入管道时阻塞,并且在使用 close() 方法时引发 IOError。
所以我的问题是:有没有更简洁的方法来处理这个问题?
我的意思是,我有一个队列不断被写入的场景。当接收进程退出时(干净或不干净),我应该关闭队列并在发送进程上得到一个 IOError 吗?
编辑:过程的输出
Waiting for a while
Subprocess starting!
Adding stuff to queue...
Subprocess exiting!
Added 1822174 objects!
Traceback (most recent call last):
File "/usr/lib/python2.7/multiprocessing/queues.py", line 266, in _feed
send(obj)
IOError: [Errno 32] Broken pipe
Child process state: 0
这部分仅在使用注释的 self.queue.close() 时发生:
Traceback (most recent call last):
File "/usr/lib/python2.7/multiprocessing/queues.py", line 266, in _feed
send(obj)
IOError: [Errno 32] Broken pipe
最佳答案
我正在回答我自己的问题,因为不是每个人都会阅读评论。在评论中收到用户 mata 的提示后,我测试了问题中的示例代码,在将对象添加到队列的循环中添加了对 time.sleep(0.01)
的调用,因此我可以限制将添加到队列中的对象数:
def addToQueue(self):
starttime = time.time()
count=0
print "Adding stuff to queue..."
while time.time()-starttime < 4:
self.queue.put("string")
count += 1
time.sleep(0.01)
print "Added %d objects!" % count
因此,当对象数量较少(在此示例中少于 3800)时,进程会正常退出。但是当有很多对象时,进程之间的管道中似乎有一些锁定。
但这对我提出了另一个问题:这是一个错误吗?我应该报告吗?或者这只是正常的预期行为?
非常感谢用户 mata 指出这种可能性!
关于Python 多处理队列 : what to do when the receiving process quits?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/10607553/