Multiprocessing queues receiving from wrong process

[MRAB]

On 2016-12-23 21:56, Charles Hixson wrote:
> I was looking to avoid using a upd connection to transfer messages
> between processes, so I thought I'd use multiprocessing (which I expect
> would be faster), but...I sure would appreciate an explanation of this
> problem.
>
> When I run the code (below) instead of messages receiving messages from
> the correct process I get:
> (where the first number of a line identifies the index assigned to the
> process.)
>
> waiting for completion
> 1=Process-2 received: at time 0.001243: Process-1 says Hi to 0
> 0=Process-1 received: at time 0.001211: Process-2 says Hi to 1
> 0=Process-1 received: at time 0.001144: Process-3 says Hi to 2
> 4=Process-5 received: at time 0.002324: Process-1 says Hi to 0
> 0=Process-1 received: at time 0.000953: Process-4 says Hi to 3
> 0=Process-1 received: at time 0.000674: Process-5 says Hi to 4
> 3=Process-4 received: at time 0.002114: Process-1 says Hi to 0
> 3=Process-4 received: at time 0.001864: Process-2 says Hi to 1
> 4=Process-5 received: at time 0.002094: Process-2 says Hi to 1
> 2=Process-3 received: at time 0.001711: Process-1 says Hi to 0
> 4=Process-5 received: at time 0.001885: Process-3 says Hi to 2
> 4=Process-5 received: at time 0.001586: Process-4 says Hi to 3
> 1=Process-2 received: at time 0.001456: Process-3 says Hi to 2
> 3=Process-4 received: at time 0.001734: Process-3 says Hi to 2
> 2=Process-3 received: at time 0.00158: Process-2 says Hi to 1
> 2=Process-3 received: at time 0.001444: Process-4 says Hi to 3
> 2=Process-3 received: at time 0.001088: Process-5 says Hi to 4
> 3=Process-4 received: at time 0.001221: Process-5 says Hi to 4
> 1=Process-2 received: at time 0.001212: Process-4 says Hi to 3
> 1=Process-2 received: at time 0.000885: Process-5 says Hi to 4
>
I don't see a bug, but perhaps that's because it's not clear to me what 
you expected to see.

> ##    Test multiprocessing queues
> import    multiprocessing    as    mp
> import    time
>
> from    multiprocessing    import    Process
> from    multiprocessing    import    Queue
> from    queue                    import    Empty
> from    queue                    import    Full
> from    random                import    random
>
>
> class TestMPQ:
>      """ Class doc """
>
>      def __init__ (self, ndx):
>          """ Class initialiser """
>          self.name    =    mp.current_process().name
>          self.ndx    =    ndx
>
>      def    sendHi (self):
>          for i in range(5):
>              if i != self.ndx:
>                  qs[i].put ("{} says Hi to {}".format(self.name, self.ndx))
>
>      def    processHi (self):
>          while (True):
>              time.sleep(random() + 0.001)
>              try:
>                  msg    =    qs[self.ndx].get_nowait()
>                  print ("{}={} received: {}".format(self.ndx, self.name,
> msg) )
>              except    Empty:
>                  break
>              except Exception as ex:
>                  print ("Exception: ", repr(ex))
>                  break
>
> def    procHandler (ndx, qs):
>      p    =    TestMPQ(ndx)
>      p.sendHi()
>      p.processHi()
>
> if "__main__" == __name__:
>      qs    =    []
>      for i in range(5):
>          qs.append(Queue())
>      ps    =    []
>      for i in range(5):
>          ps.append(Process(target = procHandler, args = (i, qs) ) )
>          ps[i].start()
>      print ("waiting for completion")
>      for i in range(5):
>          ps[i].join()
>
>