Python multiprocessing using a List of Queues

Question

I try to create a list of queues using Python 2.7 multiprocessing package. Each sub-process owns an individual queue and has two tasks: getting elements from its own queue and putting elements to other sub-processes' queues. So, each sub-process has to know which queue belongs to it, that's why I use a list of queues.

I made code as follows:

mgr = multiprocessing.Manager()

sharedQueueList = mgr.list()

for i in xrange(num_processes):
  sharedQueueList .append(mgr.Queue())

But, I got the following error message:

**raise convert_to_error(kind, result)**
RemoteError: 
---------------------------------------------------------------------------
Unserializable message: ('#RETURN', < Queue.Queue instance at 0x02AD3170 >)
---------------------------------------------------------------------------

Answer 1

Create the list of Queues in the parent, hand a few to each worker at creation time. Each worker will take jobs from one of its queues, output to another queue.

import logging, multiprocessing

def myproc(arg):
    return arg*2

def worker(qlist):
    logger = multiprocessing.get_logger()
    logger.info('start')
    while True:
        job = qlist[0].get()
        logger.info('got %s', job)
        if job is None:
            logger.info('exiting')
            return
        qlist[1].put( myproc(job) )

logger = multiprocessing.log_to_stderr(
    level=logging.INFO,
)
logger.info('setup')

numProcs = 3
queueList = [ multiprocessing.Queue() for i in xrange(numProcs) ]

# prefill with 3 jobs
for num in range(3):
    queueList[0].put(num)
# signal end of jobs
queueList[0].put(None)

worker_p = multiprocessing.Process(
    target=worker, args=( [queueList[0], queueList[1]], ),
    name='worker',
)
worker_p.start()

worker_p.join()

logger.info('done')

Example run:

[INFO/MainProcess] setup
[INFO/worker] child process calling self.run()
[INFO/worker] start
[INFO/worker] got 0
[INFO/worker] got 1
[INFO/worker] got 2
[INFO/worker] got None
[INFO/worker] exiting
[INFO/worker] process shutting down
[INFO/worker] process exiting with exitcode 0
[INFO/MainProcess] done
[INFO/MainProcess] process shutting down