Python Multiprocessing Queue Error

Question

I create 100 child processes

proc_list = [ Process(target = simulator, args=(result_queue,)) for i in xrange(100)] 

and run them

 for proc in proc_list: proc.start() 

Each process places 10,000 tuples in the result_queue (instance of multiprocessing.Queue) after some processing has been performed.

 def simulate(alg_instance, image_ids, gamma, results, simulations, sim_semaphore): (rs, qs, t_us) = alg_instance.simulate_multiple(image_ids, gamma, simulations) all_tuples = zip(rs, qs, t_us) for result in all_tuples: results.put(result) sim_semaphore.release() 

I should (?) Get 1,000,000 tuples in the queue, but after various runs I get these (samples) sizes: 14912 19563 12952 13524 7487 18350 15986 11928 14281 14282 7317

Any suggestions?

Answer 1

My solution to multiprocessing problems almost always uses Manager objects. Although the open interface is the same, the underlying implementation is much simpler and has fewer errors.

 from multiprocessing import Manager manager = Manager() result_queue = manager.Queue() 

Try and see if you can fix your problems.

Answer 2

Multiprocessing .Queue is called thread safe in its documents. But when you do interprocess communication with Queue, it should be used with multiprocessing.Manager (). Queue ()

Answer 3

There are no messages from the OP message that multiprocessing.Queue not working. The code sent by the OP is not at all sufficient to understand what is happening: do they join all the processes? Do they correctly pass the queue to child processes (should be as a parameter if it is on Windows)? do their child processes confirm that they actually got 10,000 tuples? and etc.

There is a possibility that the OP does encounter a hard-to- mp.Queue error in mp.Queue , but given the amount of testing CPython and the fact that I just executed 100 x 10,000 processes without any problems, I suspect the OP really had some The problem is in their own code.

Yes, Manager().Queue() mentioned in other answers is a great way to exchange data, but there is no reason to avoid multiprocessing.Queue() based on unconfirmed reports that “something is wrong with this.”