Python Multiprocessing Pipe "Deadlock"

Question

I'm facing problems with the following example code:

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()

When the done Queue becomes big enough (a limit about 64k i think), the whole thing freezes without any further notice.

What is the general approach for such a situation when the queue becomes too big? is there some way to remove elements on the fly once they are processed? The Python docs recommend removing the p.join(), in a real application however i can not estimate when the processes have finished. Is there a simple solution for this problem besides infinite looping and using .get_nowait()?

Answer 1

This works for me with 3.4.0alpha4, 3.3, 3.2, 3.1 and 2.6. It tracebacks with 2.7 and 3.0. I pylint'd it, BTW.

#!/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