How to use multiprocessing queue in Python?

by

My main problem is that I really don’t know how to implement multiprocessing.queue correctly, you cannot really instantiate the object for each process since they will be separate queues, how do you make sure that all processes relate to a shared queue (or in this case, queues)

This is a simple example of a reader and writer sharing a single queue… The writer sends a bunch of integers to the reader; when the writer runs out of numbers, it sends ‘DONE’, which lets the reader know to break out of the read loop.

You can spawn as many reader processes as you like…

from multiprocessing import Process, Queue
import time
import sys


def reader_proc(queue):
    """Read from the queue; this spawns as a separate Process"""
    while True:
        msg = queue.get()  # Read from the queue and do nothing
        if msg == "DONE":
            break


def writer(count, num_of_reader_procs, queue):
    """Write integers into the queue.  A reader_proc() will read them from the queue"""
    for ii in range(0, count):
        queue.put(ii)  # Put 'count' numbers into queue

    ### Tell all readers to stop...
    for ii in range(0, num_of_reader_procs):
        queue.put("DONE")


def start_reader_procs(qq, num_of_reader_procs):
    """Start the reader processes and return all in a list to the caller"""
    all_reader_procs = list()
    for ii in range(0, num_of_reader_procs):
        ### reader_p() reads from qq as a separate process...
        ###    you can spawn as many reader_p() as you like
        ###    however, there is usually a point of diminishing returns
        reader_p = Process(target=reader_proc, args=((qq),))
        reader_p.daemon = True
        reader_p.start()  # Launch reader_p() as another proc

        all_reader_procs.append(reader_p)

    return all_reader_procs


if __name__ == "__main__":
    num_of_reader_procs = 2
    qq = Queue()  # writer() writes to qq from _this_ process
    for count in [10**4, 10**5, 10**6]:
        assert num_of_reader_procs > 0
        assert num_of_reader_procs < 4
        all_reader_procs = start_reader_procs(qq, num_of_reader_procs)

        writer(count, len(all_reader_procs), qq)  # Queue stuff to all reader_p()
        print("All reader processes are pulling numbers from the queue...")

        _start = time.time()
        for idx, a_reader_proc in enumerate(all_reader_procs):
            print("    Waiting for reader_p.join() index %s" % idx)
            a_reader_proc.join()  # Wait for a_reader_proc() to finish

            print("        reader_p() idx:%s is done" % idx)

        print(
            "Sending {0} integers through Queue() took {1} seconds".format(
                count, (time.time() - _start)
            )
        )
        print("")

More Related Contents:

  1. Multiprocessing vs Threading Python [duplicate]
  2. Can’t pickle when using multiprocessing Pool.map()
  3. How to run functions in parallel?
  4. multiprocessing.Pool: When to use apply, apply_async or map?
  5. What are the differences between the threading and multiprocessing modules?
  6. Processing single file from multiple processes
  7. multiprocessing vs multithreading vs asyncio in Python 3
  8. What kind of problems (if any) would there be combining asyncio with multiprocessing?
  9. multiprocessing vs multithreading vs asyncio
  10. Using multiprocessing.Process with a maximum number of simultaneous processes
  11. How to limit execution time of a function call?
  12. Why does multiprocessing use only a single core after I import numpy?
  13. Python threading multiple bash subprocesses?
  14. Sharing a result queue among several processes
  15. Keyboard Interrupts with python’s multiprocessing Pool
  16. multiprocessing in python – sharing large object (e.g. pandas dataframe) between multiple processes
  17. The right way to limit maximum number of threads running at once?
  18. Can I use a multiprocessing Queue in a function called by Pool.imap?
  19. Accessing an attribute of a multiprocessing Proxy of a class
  20. call multiprocessing in class method Python
  21. Leveraging “Copy-on-Write” to Copy Data to Multiprocessing.Pool() Worker Processes
  22. Tkinter GUI, I/O & Threading: When to use queues, when events?
  23. How to get the return value of a function passed to multiprocessing.Process?
  24. Python thread name doesn’t show up on ps or htop
  25. Threads are printing at the same time messing up the text output
  26. Creating and updating nested dictionaries and lists inside multiprocessing.Manager object
  27. Access a QLCDNumber object from a external function
  28. How do I create a file in python without overwriting an existing file
  29. How to implement a Lock with a timeout in Python 2.7
  30. Trigger an event when clipboard content changes

Leave a Comment