Python Multiprocessing Using a Queue List

Question

I am trying to create a queue list using the Python 2.7 multiprocessing package. Each subprocess owns a separate queue and has two tasks: obtaining elements from its own queue and placing elements in the queue of other subprocesses. Thus, each subprocess must know which one belongs to it, so I use a list of queues.

I made the 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 a Queue list in the parent, from a few to each worker at creation time. Each employee will perform tasks from one of his queues, withdrawal 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') 

Execution Example:

 [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