How to use multiprocessing queue in Python

Question

I m having much trouble trying to understand just how the multiprocessing queue works on python and how to implement it  Lets say I have two python modules that access data from a shared file  let s call these two modules a writer and a reader  My plan is to have both the reader and writer put requests into two separate multiprocessing queues  and then have a third process pop these requests in a loop and execute as such   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

User · Answer

A multi-producers and multi-consumers example, verified. It should be easy to modify it to cover other cases, single/multi producers, single/multi consumers.

from multiprocessing import Process, JoinableQueue
import time
import os

q = JoinableQueue()

def producer():
    for item in range(30):
        time.sleep(2)
        q.put(item)
    pid = os.getpid()
    print(f'producer {pid} done')


def worker():
    while True:
        item = q.get()
        pid = os.getpid()
        print(f'pid {pid} Working on {item}')
        print(f'pid {pid} Finished {item}')
        q.task_done()

for i in range(5):
    p = Process(target=worker, daemon=True).start()

# send thirty task requests to the worker
producers = []
for i in range(2):
    p = Process(target=producer)
    producers.append(p)
    p.start()

# make sure producers done
for p in producers:
    p.join()

# block until all workers are done
q.join()
print('All work completed')

Explanation:

Two producers and five consumers in this example.
JoinableQueue is used to make sure all elements stored in queue will be processed. 'task_done' is for worker to notify an element is done. 'q.join()' will wait for all elements marked as done.
With #2, there is no need to join wait for every worker.
But it is important to join wait for every producer to store element into queue. Otherwise, program exit immediately.

User · Answer

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   from multiprocessing import Process  Queue import time import sys  def reader proc queue          Read from the queue  this will be spawned as a separate Process     while True          msg   queue get             Read from the queue and do nothing         if  msg     DONE                break  def writer count  queue          Write to the queue     for ii in range 0  count           queue put ii                Write  count  numbers into the queue     queue put  DONE    if   name       main         pqueue   Queue     writer   writes to pqueue from  this  process     for count in  10  4  10  5  10  6                            reader proc   reads from pqueue as a separate process         reader p   Process target reader proc  args   pqueue             reader p daemon   True         reader p start            Launch reader proc   as a separate python process           start   time time           writer count  pqueue       Send a lot of stuff to reader           reader p join             Wait for the reader to finish         print  Sending  0  numbers to Queue   took  1  seconds  format count                time time   -  start

User · Answer

I had a look at multiple answers across stack overflow and the web while trying to set-up a way of doing multiprocessing using queues for passing around large pandas dataframes  It seemed to me that every answer was re-iterating the same kind of solutions without any consideration of the multitude of edge cases one will definitely come across when setting up calculations like these  The problem is that there is many things at play at the same time  The number of tasks  the number of workers  the duration of each task and possible exceptions during task execution  All of these make synchronization tricky and most answers do not address how you can go about it  So this is my take after fiddling around for a few hours  hopefully this will be generic enough for most people to find it useful  Some thoughts before any coding examples  Since queue Empty or queue qsize   or any other similar method is unreliable for flow control  any code of the like while True      try          task   pending queue get nowait       except queue Empty          break  is bogus  This will kill the worker even if milliseconds later another task turns up in the queue  The worker will not recover and after a while ALL the workers will disappear as they randomly find the queue momentarily empty  The end result will be that the main multiprocessing function  the one with the join   on the processes  will return without all the tasks having completed  Nice  Good luck debugging through that if you have thousands of tasks and a few are missing  The other issue is the use of sentinel values  Many people have suggested adding a sentinel value in the queue to flag the end of the queue  But to flag it to whom exactly  If there is N workers  assuming N is the number of cores available give or take  then a single sentinel value will only flag the end of the queue to one worker  All the other workers will sit waiting for more work when there is none left  Typical examples I ve seen are while True      task   pending queue get       if task    SOME SENTINEL VALUE          break  One worker will get the sentinel value while the rest will wait indefinitely  No post I came across mentioned that you need to submit the sentinel value to the queue AT LEAST as many times as you have workers so that ALL of them get it  The other issue is the handling of exceptions during task execution  Again these should be caught and managed  Moreover  if you have a completed tasks queue you should independently count in a deterministic way how many items are in the queue before you decide that the job is done  Again relying on queue sizes is bound to fail and returns unexpected results  In the example below  the par proc   function will receive a list of tasks including the functions with which these tasks should be executed alongside any named arguments and values  import multiprocessing as mp import dill as pickle import queue import time import psutil  SENTINEL   None   def do work tasks pending  tasks completed         Get the current worker s name     worker name   mp current process   name      while True          try              task   tasks pending get nowait           except queue Empty              print worker name     found an empty queue  Sleeping for a while before checking again                  time sleep 0 01          else              try                  if task    SENTINEL                      print worker name     no more work left to be done  Exiting                          break                  print worker name     received some work                       time start   time perf counter                   work func   pickle loads task  func                    result   work func   task  task                    tasks completed put  work func   name    result                   time end   time perf counter   - time start                 print worker name     done in    seconds  format round time end  5                except Exception as e                  print worker name     task failed      str e                   tasks completed put  work func   name    None     def par proc job list  num cpus None          Get the number of cores     if not num cpus          num cpus   psutil cpu count logical False       print    Parallel processing       print    Running on    cores  format num cpus          Set-up the queues for sending and receiving data to from the workers     tasks pending   mp Queue       tasks completed   mp Queue          Gather processes and results here     processes          results             Count tasks     num tasks   0        Add the tasks to the queue     for job in job list          for task in job  tasks                expanded job                  num tasks   num tasks   1             expanded job update   func   pickle dumps job  func                  expanded job update   task   task               tasks pending put expanded job         Use as many workers as there are cores  usually chokes the system so better use less      num workers   num cpus        We need as many sentinels as there are worker processes so that ALL processes exit when there is no more       work left to be done      for c in range num workers           tasks pending put SENTINEL       print    Number of tasks      format num tasks          Set-up and start the workers     for c in range num workers           p   mp Process target do work  args  tasks pending  tasks completed           p name    worker    str c          processes append p          p start          Gather the results     completed tasks counter   0     while completed tasks counter  lt  num tasks          results append tasks completed get            completed tasks counter   completed tasks counter   1      for p in processes          p join        return results  And here is a test to run the above code against def test parallel processing        def heavy duty1 arg1  arg2  arg3           return arg1   arg2   arg3      def heavy duty2 arg1  arg2  arg3           return arg1   arg2   arg3      task list               func   heavy duty1   tasks      arg1   1   arg2   2   arg3   3     arg1   1   arg2   3   arg3   5               func   heavy duty2   tasks      arg1   1   arg2   2   arg3   3     arg1   1   arg2   3   arg3   5                results   par proc task list       job1   sum  y for x in results if  heavy duty1  in x keys   for y in list x values          job2   sum  y for x in results if  heavy duty2  in x keys   for y in list x values           assert job1    15     assert job2    21  plus another one with some exceptions def test parallel processing exceptions        def heavy duty1 raises arg1  arg2  arg3           raise ValueError  Exception raised           return arg1   arg2   arg3      def heavy duty2 arg1  arg2  arg3           return arg1   arg2   arg3      task list               func   heavy duty1 raises   tasks      arg1   1   arg2   2   arg3   3     arg1   1   arg2   3   arg3   5               func   heavy duty2   tasks      arg1   1   arg2   2   arg3   3     arg1   1   arg2   3   arg3   5                results   par proc task list       job1   sum  y for x in results if  heavy duty1  in x keys   for y in list x values          job2   sum  y for x in results if  heavy duty2  in x keys   for y in list x values           assert not job1     assert job2    21  Hope that is helpful

User · Answer

Here s a dead simple usage of multiprocessing Queue and multiprocessing Process that allows callers to send an  event  plus arguments to a separate process that dispatches the event to a  do   method on the process   Python 3 4    import multiprocessing as mp import collections  Msg   collections namedtuple  Msg     event    args     class BaseProcess mp Process          A process backed by an internal queue for simple one-way message passing              def   init   self   args    kwargs           super     init    args    kwargs          self queue   mp Queue        def send self  event   args              Puts the event and args as a  Msg  on the queue                    msg   Msg event  args         self queue put msg       def dispatch self  msg           event  args   msg          handler   getattr self   do  s    event  None          if not handler              raise NotImplementedError  Process has no handler for   s     event           handler  args       def run self           while True              msg   self queue get               self dispatch msg    Usage   class MyProcess BaseProcess       def do helloworld self  arg1  arg2           print arg1  arg2   if   name         main         process   MyProcess       process start       process send  helloworld    hello    world     The send happens in the parent process  the do   happens in the child process    I left out any exception handling that would obviously interrupt the run loop and exit the child process  You can also customize it by overriding run to control blocking or whatever else    This is really only useful in situations where you have a single worker process  but I think it s a relevant answer to this question to demonstrate a common scenario with a little more object-orientation

User · Answer

Just made a simple and general example for demonstrating passing a message over a Queue between 2 standalone programs  It doesn t directly answer the OP s question but should be clear enough indicating the concept  Server  multiprocessing-queue-manager-server py  import asyncio import concurrent futures import multiprocessing import multiprocessing managers import queue import sys import threading from typing import Any  AnyStr  Dict  Union   class QueueManager multiprocessing managers BaseManager        def get queue self  ident  Union AnyStr  int  type None     None  - gt  multiprocessing Queue          pass   def get queue ident  Union AnyStr  int  type None     None  - gt  multiprocessing Queue      global q      if not ident in q          q ident    multiprocessing Queue        return q ident    q  Dict Union AnyStr  int  type None    multiprocessing Queue    dict   delattr QueueManager   get queue     def init queue manager server        if not hasattr QueueManager   get queue            QueueManager register  get queue   get queue    def serve no  int  term ev  threading Event       manager  QueueManager     with QueueManager authkey QueueManager   name   encode    as manager          print f quot Server address  no    manager address  quot            while not term ev is set                try                  item  Any   manager get queue   get timeout 0 1                  print f quot Client  no    item  from  manager address  quot               except queue Empty                  continue   async def main n  int       init queue manager server       term ev  threading Event   threading Event       executor  concurrent futures ThreadPoolExecutor   concurrent futures ThreadPoolExecutor        i  int     for i in range n           asyncio ensure future asyncio get running loop   run in executor executor  serve  i  term ev          Gracefully shut down     try          await asyncio get running loop   create future       except asyncio CancelledError          term ev set           executor shutdown           raise   if   name         main         asyncio run main int sys argv 1      Client  multiprocessing-queue-manager-client py  import multiprocessing import multiprocessing managers import os import sys from typing import AnyStr  Union   class QueueManager multiprocessing managers BaseManager        def get queue self  ident  Union AnyStr  int  type None     None  - gt  multiprocessing Queue          pass   delattr QueueManager   get queue     def init queue manager client        if not hasattr QueueManager   get queue            QueueManager register  get queue     def main        init queue manager client        manager  QueueManager   QueueManager sys argv 1   authkey QueueManager   name   encode        manager connect        message   f quot A message from  os getpid    quot      print f quot Message to send   message  quot       manager get queue   put message    if   name         main         main    Usage Server    python3 multiprocessing-queue-manager-server py N  N is a integer indicating how many servers should be created  Copy one of the  lt server-address-N gt  output by the server and make it the first argument of each multiprocessing-queue-manager-client py  Client  python3 multiprocessing-queue-manager-client py  lt server-address-1 gt   Result Server  Client 1   lt item gt  from  lt server-address-1 gt   Gist  https   gist github com 89062d639e40110c61c2f88018a8b0e5  UPD  Created a package here  Server  import ipcq   with ipcq QueueManagerServer address ipcq Address AUTO  authkey ipcq AuthKey AUTO  as server      server get queue   get    Client  import ipcq   client   ipcq QueueManagerClient address ipcq Address AUTO  authkey ipcq AuthKey AUTO  client get queue   put  a message

User · Answer

in  from queue import Queue  there is no module called queue  instead multiprocessing should be used  Therefore  it should look like  from multiprocessing import Queue

User · Answer

We implemented two versions of this  one a simple multi thread pool that can execute many types of callables  making our lives much easier and the second version that uses processes  which is less flexible in terms of callables and requires and extra call to dill    Setting frozen pool to true will freeze execution until finish pool queue is called in either class   Thread Version       Created on Nov 4  2019   author  Kevin     from threading import Lock  Thread from Queue import Queue import traceback from helium loaders loader retailers import print info from time import sleep import signal import os  class ThreadPool object       def   init   self  queue threads   args    kwargs           self frozen pool   kwargs get  frozen pool   False          self print queue   kwargs get  print queue   True          self pool results              self lock   Lock           self queue threads   queue threads         self queue   Queue           self threads               for i in range self queue threads               t   Thread target self make pool call              t daemon   True             t start               self threads append t       def make pool call self           while True              if self frozen pool                   print  -- gt  Queue is frozen                  sleep 1                  continue              item   self queue get               if item is None                  break              call   item get  call   None              args   item get  args                   kwargs   item get  kwargs                   keep results   item get  keep results   False               try                  result   call  args    kwargs                   if keep results                      self lock acquire                       self pool results append  item  result                       self lock release                except Exception as e                  self lock acquire                   print e                 traceback print exc                   self lock release                   os kill os getpid    signal SIGUSR1               self queue task done        def finish pool queue self           self frozen pool   False          while self queue unfinished tasks  gt  0              if self print queue                  print info  -- gt  Thread pool     s    self queue unfinished tasks              sleep 5           self queue join            for i in range self queue threads               self queue put None           for t in self threads              t join            del self threads         def get pool results self           return self pool results      def clear pool results self           del self pool results      Process Version         Created on Nov 4  2019   author  Kevin     import traceback from helium loaders loader retailers import print info from time import sleep import signal import os from multiprocessing import Queue  Process  Value  Array  JoinableQueue  Lock       RawArray  Manager from dill import dill import ctypes from helium misc utils import ignore exception from mem top import mem top import gc  class ProcessPool object       def   init   self  queue processes   args    kwargs           self frozen pool   Value ctypes c bool  kwargs get  frozen pool   False           self print queue   kwargs get  print queue   True          self manager   Manager           self pool results   self manager list           self queue processes   queue processes         self queue   JoinableQueue           self processes               for i in range self queue processes               p   Process target self make pool call              p start               self processes append p           print  Processes   self queue processes      def make pool call self           while True              if self frozen pool value                  sleep 1                  continue              item pickled   self queue get                if item pickled is None                   print  -- gt  Ending                  self queue task done                   break              item   dill loads item pickled               call   item get  call   None              args   item get  args                   kwargs   item get  kwargs                   keep results   item get  keep results   False               try                  result   call  args    kwargs                   if keep results                      self pool results append dill dumps  item  result                    else                      del call  args  kwargs  keep results  item  result              except Exception as e                  print e                 traceback print exc                   os kill os getpid    signal SIGUSR1               self queue task done        def finish pool queue self  callable None           self frozen pool value   False          while self queue  unfinished tasks get value    gt  0              if self print queue                  print info  -- gt  Process pool     s     self queue  unfinished tasks get value                  if callable                  callable                sleep 5           for i in range self queue processes               self queue put None           self queue join           self queue close            for p in self processes              with ignore exception  p join 10              with ignore exception  p terminate            with ignore exception  del self processes         def get pool results self           return self pool results      def clear pool results self           del self pool results       def test eg           print  EG   eg    Call with either   tp   ThreadPool queue threads 2  tp queue put   call   test   args    random randint 0  100     tp finish pool queue     or  pp   ProcessPool queue processes 2  pp queue put dill dumps   call   test   args    random randint 0  100      pp queue put dill dumps   call   test   args    random randint 0  100      pp finish pool queue

[python] How to use multiprocessing queue in Python?

Examples related to python

Examples related to multithreading

Examples related to multiprocessing