Python Concurrency: Hangs when using apply_async()
Question: I am trying to learn about Python concurrency. As an experiment, I have the
following program that uses a process pool and calls workers via
apply_async(). To share information between the processes (work and results),
I use a queue from multiprocessing.Manager().

However, this code hangs -- sometimes -- when all items in the work queue have
been processed, and I'm not sure why. I have to run the program a few times to
observe the behavior.

As a side note, I _can_ make this work correctly: I have found some a design
pattern that people sometimes call the "poison pill" method, and it seems to
work. (In my worker() method, I enter an infinite loop and break out of the
loop when my work queue contains a sentinel value. I create as many sentinel
values on the work queue as I have running processes).

But I'm still interested in finding out why this code hangs. I am getting
output like the following (the process ID is in brackets):

    
    
    Found 8 CPUs.  
    Operation queue has 20 items.  
    Will start 2 processes.  
    Joining pool...  
    [5885] entering worker() with work_queue size of 20  
    [5885] processed work item 0  
    [5885] worker() still running because work_queue has size 19  
    [5885] processed work item 1  
    [5885] worker() still running because work_queue has size 18  
    [5885] processed work item 2  
    [5885] worker() still running because work_queue has size 17  
    [5885] processed work item 3  
    [5885] worker() still running because work_queue has size 16  
    [5885] processed work item 4  
    [5885] worker() still running because work_queue has size 15  
    [5885] processed work item 5  
    [5886] entering worker() with work_queue size of 14  
    [5885] worker() still running because work_queue has size 14  
    [5886] processed work item 6  
    [5886] worker() still running because work_queue has size 13  
    [5885] processed work item 7  
    [5886] processed work item 8  
    [5885] worker() still running because work_queue has size 11  
    [5886] worker() still running because work_queue has size 11  
    [5885] processed work item 9  
    [5886] processed work item 10  
    [5885] worker() still running because work_queue has size 9  
    [5886] worker() still running because work_queue has size 9  
    [5885] processed work item 11  
    [5886] processed work item 12  
    [5885] worker() still running because work_queue has size 7  
    [5886] worker() still running because work_queue has size 7  
    [5885] processed work item 13  
    [5886] processed work item 14  
    [5885] worker() still running because work_queue has size 5  
    [5886] worker() still running because work_queue has size 5  
    [5885] processed work item 15  
    [5886] processed work item 16  
    [5885] worker() still running because work_queue has size 3  
    [5886] worker() still running because work_queue has size 3  
    [5885] processed work item 17  
    [5886] processed work item 18  
    [5885] worker() still running because work_queue has size 1  
    [5886] worker() still running because work_queue has size 1  
    [5885] processed work item 19  
    [5885] worker() still running because work_queue has size 0  
    [5885] worker() is finished; returning results of size 20  
    

(The program hangs on the last line. The other process -- 5886 -- doesn't seem
to finish.)

    
    
    import multiprocessing
    from multiprocessing import Pool
    import os
    
    # Python 2.7.6 on Linux
    
    # Worker (consumer) process
    def worker(work_queue, results_queue):
        print "[%d] entering worker() with work_queue size of %d" % (os.getpid(), work_queue.qsize())
        while not work_queue.empty():
            item = work_queue.get()   
            print "[%d] processed work item %s" % (os.getpid(), item)  
            s = '[%d] has processed %s.' % (os.getpid(), item)
            results_queue.put(s)
            work_queue.task_done()
            print "[%d] worker() still running because work_queue has size %d" % (os.getpid(), work_queue.qsize())
        print "[%d] worker() is finished; returning results of size %d" % (os.getpid(), results_queue.qsize())
    
    if __name__ == '__main__':
    
        MAX_PROCESS = 2     # Max number of processes to create
        MAX_ITEMS = 20      # Max work items to process
    
        m = multiprocessing.Manager()
        results_queue = m.Queue()
        work_queue = m.Queue()
    
        # Assign work
        for x in xrange(MAX_ITEMS):
            work_queue.put(x)     
    
        print "Found %d CPUs." % multiprocessing.cpu_count()
        print "Operation queue has %d items." % work_queue.qsize()
        print "Will start %d processes." % MAX_PROCESS
    
        # Pool method
        pool = Pool(processes=MAX_PROCESS)
        for n in range(0, MAX_PROCESS):
            pool.apply_async(worker, args=(work_queue, results_queue))
        pool.close()
        print "Joining pool..."
        pool.join()
        print "Joining pool finished..."
        print "--- After pool completion ---"
    
        print "Work queue has %d items." % work_queue.qsize()
        print "Results queue has %d items." % results_queue.qsize()
        print "Results are:"
    
        while not results_queue.empty():
            item = results_queue.get()
            print str(item)
            results_queue.task_done()
        print "--End---"
    

Thanks for the help.


Answer: You're hitting a race condition - process `5886` sees that the `Pool` has an
item in it:

    
    
    [5886] worker() still running because work_queue has size 1
    

So it loops back around to the blocking `get` call:

    
    
    while not work_queue.empty(): # It sees it's not emtpy here
        item = work_queue.get()   # But there's no item left by the time it gets here!
    

However, after it calls `work_queue.empty()`, but _before_ it calls
`work_queue.get()`, the other worker (`5885`) consumes the last item in the
queue:

    
    
    [5885] processed work item 19  
    [5885] worker() still running because work_queue has size 0  
    [5885] worker() is finished; returning results of size 20  
    

So now `5886` will block forever on the `get`. In general, you shouldn't use
the `empty()` method to decide whether or not to make a blocking `get()` call
if there are multiple consumers of a queue, because it's susceptible to this
kind of race condition. Using the "poison pill"/sentinel method you mentioned
is the right way to handle this scenario, or use a non-blocking `get` call,
and catch the `Empty` exception should it occur:

    
    
    try:
        item = work_queue.get_nowait()
        print "[%d] processed work item %s" % (os.getpid(), item)  
        s = '[%d] has processed %s.' % (os.getpid(), item)
        results_queue.put(s)
        work_queue.task_done()
        print "[%d] worker() still running because work_queue has size %d" % (os.getpid(), work_queue.qsize())
    except Queue.Empty:
        print "[%d] worker() is finished; returning results of size %d" % (os.getpid(), results_queue.qsize())
    

Note that you can only use this approach if you know the queue won't ever grow
in size once the workers have begun to consume from it. Otherwise you could
decide the worker should exit when there are still more items to be added to
the queue.

