Python的多重管道“死锁”

Question

我面临着下面的示例代码中的问题：

from multiprocessing import Lock, Process, Queue, current_process 

def worker(work_queue, done_queue): 
    for item in iter(work_queue.get, 'STOP'): 
      print("adding ", item, "to done queue") 
      #this works: done_queue.put(item*10) 
      done_queue.put(item*1000) #this doesnt! 
    return True 

def main(): 
    workers = 4 
    work_queue = Queue() 
    done_queue = Queue() 
    processes = [] 

    for x in range(10): 
     work_queue.put("hi"+str(x)) 

    for w in range(workers): 
     p = Process(target=worker, args=(work_queue, done_queue)) 
     p.start() 
     processes.append(p) 
     work_queue.put('STOP') 

    for p in processes: 
     p.join() 

    done_queue.put('STOP') 

    for item in iter(done_queue.get, 'STOP'): 
     print(item) 


if __name__ == '__main__': 
    main() 

当完成队列变得足够大（关于64K，我认为限制），整个事情冻结没有任何进一步的注意。

当队列变得太大时，这种情况的一般方法是什么？一旦它们被处理后，是否有一些方法可以在运行中删除元素？ The Python docs recommend removing the p.join()，但在实际应用中，我无法估计进程何时完成。除了无限循环和使用.get_nowait（）之外，是否还有一个简单的解决方案？

Answer 1

这适用于我3.4.0alpha4,3.3,3.2,3.1和2.6。它追溯到2.7和3.0。顺便说一句，顺便说一句。

#!/usr/local/cpython-3.3/bin/python 

'''SSCCE for a queue deadlock''' 

import sys 
import multiprocessing 

def worker(workerno, work_queue, done_queue): 
    '''Worker function''' 
    #reps = 10 # this worked for the OP 
    #reps = 1000 # this worked for me 
    reps = 10000 # this didn't 

    for item in iter(work_queue.get, 'STOP'): 
     print("adding", item, "to done queue") 
     #this works: done_queue.put(item*10) 
     for thing in item * reps: 
      #print('workerno: {}, adding thing {}'.format(workerno, thing)) 
      done_queue.put(thing) 
    done_queue.put('STOP') 
    print('workerno: {0}, exited loop'.format(workerno)) 
    return True 

def main(): 
    '''main function''' 
    workers = 4 
    work_queue = multiprocessing.Queue(maxsize=0) 
    done_queue = multiprocessing.Queue(maxsize=0) 
    processes = [] 

    for integer in range(10): 
     work_queue.put("hi"+str(integer)) 

    for workerno in range(workers): 
     dummy = workerno 
     process = multiprocessing.Process(target=worker, args=(workerno, work_queue, done_queue)) 
     process.start() 
     processes.append(process) 
     work_queue.put('STOP') 

    itemno = 0 
    stops = 0 
    while True: 
     item = done_queue.get() 
     itemno += 1 
     sys.stdout.write('itemno {0}\r'.format(itemno)) 
     if item == 'STOP': 
      stops += 1 
      if stops == workers: 
       break 
    print('exited done_queue empty loop') 


    for workerno, process in enumerate(processes): 
     print('attempting process.join() of workerno {0}'.format(workerno)) 
     process.join() 

    done_queue.put('STOP') 

if __name__ == '__main__': 
    main() 

HTH