Two python pools, two queues
Question: I am trying to understand how pool and queue work in Python, and the following
example doesn't work as expected. I expect the program to end, but it's stuck
in an infinite loop because the second queue isn't getting emptied.

    
    
    import multiprocessing
    import os
    import time
    
    inq = multiprocessing.Queue()
    outq = multiprocessing.Queue()
    
    def worker_main(q1, q2):
        while True:
            i = q1.get(True)
            time.sleep(.1) 
            q2.put(i*2)
    
    def worker2(q):
        print q.get(True)
    
    p1 = multiprocessing.Pool(3, worker_main,(inq, outq,))
    p2 = multiprocessing.Pool(2, worker2,(outq,))
    
    
    for i in range(50):
        inq.put(i)
    
    
    while inq.qsize()>0 or outq.qsize()>0:
        print 'q1 size', inq.qsize(), 'q2 size', outq.qsize()
        time.sleep(.1)
    

the output shows that the second queue (outq) is .get once, but that's all.

output:

>
>     q1 size 49 q2 size 0
>     q1 size 47 q2 size 0
>     2
>     4
>     q1 size 44 q2 size 1
>     q1 size 41 q2 size 4
>     q1 size 38 q2 size 7
>     q1 size 35 q2 size 11
>     q1 size 31 q2 size 14
>     q1 size 27 q2 size 18
>     q1 size 24 q2 size 21
>     q1 size 22 q2 size 23
>     q1 size 19 q2 size 26
>     q1 size 15 q2 size 30
>     q1 size 12 q2 size
>  

Why isn't the worker2 getting called until the outq is empty?


Answer: This is a very odd way to use a `Pool`. The function passed to the constructor
is called only once per process in the pool. It's intended for one-time
initialization tasks, and is rarely used.

As is, your `worker2` is called exactly twice, one time for each process in
yout `p2` pool. Your function gets one value from a queue and then exits. The
process never does anything else. So it's doing exactly what you coded it to
do.

There's no evident reason to use a `Pool` here at all; creating 5
`multiprocessing.Process` objects instead would be more natural.

If you feel you have to do it this way, then you need to put a loop in
`worker2`. Here's one way:

    
    
    import multiprocessing
    import time
    
    def worker_main(q1, q2):
        while True:
            i = q1.get()
            if i is None:
                break
            time.sleep(.1) 
            q2.put(i*2)
    
    def worker2(q):
        while True:
            print(q.get())
    
    if __name__ == "__main__":
        inq = multiprocessing.Queue()
        outq = multiprocessing.Queue()
        p1 = multiprocessing.Pool(3, worker_main,(inq, outq,))
        p2 = multiprocessing.Pool(2, worker2,(outq,))
    
        for i in range(50):
            inq.put(i)
        for i in range(3): # tell worker_main we're done
            inq.put(None)
    
        while inq.qsize()>0 or outq.qsize()>0:
            print('q1 size', inq.qsize(), 'q2 size', outq.qsize())
            time.sleep(.1)
    

## SUGGESTED

This is a "more natural" way to use `Process` objects instead, and using queue
sentinels (special values - here `None`) to let processes know when to stop.
BTW, I'm using Python 3, so use `print` as a function rather than as a
statement.

    
    
    import multiprocessing as mp
    import time
    
    def worker_main(q1, q2):
        while True:
            i = q1.get()
            if i is None:
                break
            time.sleep(.1) 
            q2.put(i*2)
    
    def worker2(q):
        while True:
            i = q.get()
            if i is None:
                break
            print(i)
    
    def wait(procs):
        alive_count = len(procs)
        while alive_count:
            alive_count = 0
            for p in procs:
                if p.is_alive():
                    p.join(timeout=0.1)
                    print('q1 size', inq.qsize(), 'q2 size', outq.qsize())
                    alive_count += 1
    
    if __name__ == "__main__":
        inq = mp.Queue()
        outq = mp.Queue()
        p1s = [mp.Process(target=worker_main, args=(inq, outq,))
               for i in range(3)]
        p2s = [mp.Process(target=worker2, args=(outq,))
               for i in range(2)]
        for p in p1s + p2s:
            p.start()
    
        for i in range(50):
            inq.put(i)
        for p in p1s: # tell worker_main we're done
            inq.put(None)
    
        wait(p1s)
        # Tell worker2 we're done
        for p in p2s:
            outq.put(None)
        wait(p2s)
    

