How multiple multiprocessing queues pass a value from one function to another

Question

The code posted below, while does not crash, does not return os.getpid() value from foo function. How to pass a value over the multiple queue objects so it could be printed with a print(queue.get()) command?

import time, multiprocessing, os

def foo(*args):
    outside_queue = args[0]
    logger = multiprocessing.log_to_stderr()
    logger.warning(os.getpid())
    outside_queue.put(os.getpid())
    time.sleep(30)

class Object(object):
    def run(self, *args):
        outside_queue = args[0]

        items = dict()
        for i in range(5):
            queue = multiprocessing.Queue()
            proc = multiprocessing.Process(target=foo, args=(queue,))
            items[proc] = queue
            proc.start()

        for proc, queue in items.items():
            if not queue.empty():
                outside_queue.put(queue.get())

for i in range(2):
    obj = Object()
    queue = multiprocessing.Queue()
    proc = multiprocessing.Process(target=obj.run, args=(queue,))
    proc.start()
    while True:
        proc.join(1)
        if not proc.is_alive():
            break

        if not queue.empty():
            print(queue.get())
        time.sleep(0.1)

Does this [answer](https://stackoverflow.com/a/9928191/7675174) regarding using a `multiprocessing.Manager()` help? I can't execute your script without generating a RuntimeError. — import random, Nov 01 '18 at 04:53

georgexsh · Accepted Answer · 2018-11-02T04:43:11.503

it is a synchronize problem, in Object.run, between starting the child processes and getting results from the queue, there is no guarantee child process has put something into the queue, you have to wait explicitly, use Event, for example:

def foo(outside_queue, evt):
    outside_queue.put(os.getpid())
    evt.set()

class Object(object):
    def run(self, *args):
        outside_queue = args[0]
        items = dict()
        for i in range(5):
            evt = multiprocessing.Event()
            queue = multiprocessing.Queue()
            proc = multiprocessing.Process(target=foo, args=(queue, evt))
            items[proc] = queue
            proc.start()
            evt.wait(None)
            ...

How multiple multiprocessing queues pass a value from one function to another

1 Answers1