Pass variables between Pool workers?

Question

I have a pool of 2 processes. Process #1 with infinite loop. I need to stop infinite loop in process #1 when something happens in process #2. How to pass info from process #1 to process #2?

def do_smth(value):
  a = 0
  if value == "1":
    while 1:
      time.sleep(0.5)
      print("process_1", a)
      if a == 10: break
  if value == "2":
    while a < 10:
      time.sleep(0.5)
      print("process_2", a)
      a +=1

def make_a_pool(all):
  with multiprocessing.Pool(processes=2) as pool:
      pool.map(do_smth, all)

if __name__ == "__main__":
    all = ["1", "2"]
    make_a_pool(all)

Off-topic: all is the name of a built-in, so you should refrain from defining it to be something else. — martineau
– martineau, Commented May 25, 2022 at 0:00

Bharel · Accepted Answer · 2022-05-25 00:26:48Z

2

If what you wish is to share a full variable, and not just a stop condition for an infinite loop, you may use multiprocessing.Value(). Keep in mind you have to initialize the value differently, as multiprocessing.Pool cannot pass around synchronization primitives that can't be pickled:

import functools
import multiprocessing
import time

def initialize_a(a_):
    global a
    a = a_

def do_smth(value):
  if value == "1":
    while True:
      time.sleep(0.5)
      print("process_1", a.value)
      if a.value >= 10: break
  if value == "2":
    while a.value < 10:
      time.sleep(0.5)
      print("process_2", a.value)
      a.value +=1

def make_a_pool(all):
  a = multiprocessing.Value("i")
  a.value = 0
  with multiprocessing.Pool(processes=2,
                            initializer=initialize_a, initargs=(a,)) as pool:
    pool.map(do_smth, all)

if __name__ == "__main__":
    all = ["1", "2"]
    make_a_pool(all)

Output:

process_2 0
process_1 0
process_1 1
process_2 1
process_2 2
process_1 2
process_1 3
process_2 3
process_1 4
process_2 4
process_2 5
process_1 5
process_1 6
process_2 6
process_1 7
process_2 7
process_1 8
process_2 8
process_2 9
process_1 9

I do not need to use any lock, as only one process changes the value, otherwise, you need to use Value.lock().

edited May 25, 2022 at 0:26

answered May 24, 2022 at 23:36

Bharel

27.5k8 gold badges52 silver badges100 bronze badges

Sign up to request clarification or add additional context in comments.

6 Comments

martineau Over a year ago

Very generally speaking I think you may be wrong about not needing a Lock (because there are two processes reading the shared variable's value). The documentation specifically mentions that "Operations like += which involve a read and write are not atomic" and then goes on to describe a proper way to do it.

Bharel Over a year ago

@mart it's not atomic, but it doesn't need to be. In case of a collision you'll either see the previous state or the new state, and it doesn't matter which. The problem arises when you have two writers using the non atomic operation (which by itself includes a read and a write). The single write operation is atomic and will not cause memory corruption or show wrong values in the reader.

martineau Over a year ago

The point I was trying to make was merely that even though it doesn't matter much in this toy example, it might in real world code.

Bharel Over a year ago

@mart on any real world single writer multiple readers - it will not make any difference. If there are multiple writer processes, the lock will be needed of course, as I started at the end :-)

martineau Over a year ago

I disagree. Imagine a reader that needs to stop doing something after a certain value has been reached, but it happens to read the value at the same time that it's in the middle of non-atomically being incremented, so keeps on going.

|

Bharel · Accepted Answer · 2022-05-24 23:24:38Z

Simplest way is to use an Event. Keep in mind you have to initialize the event differently, as multiprocessing.Pool cannot pass around synchronization primitives that can't be pickled:

import multiprocessing
import time

def initialize_event(e):
    global event
    event = e

def do_smth(value):
  a = 0
  if value == "1":
    while not event.is_set():
      time.sleep(0.5)
      print("process_1", a)
      if a == 10: break
  if value == "2":
    while a < 10:
      time.sleep(0.5)
      print("process_2", a)
      a +=1
      if a == 5: event.set()

def make_a_pool(all):
  event = multiprocessing.Event()
  with multiprocessing.Pool(processes=2,
                            initializer=initialize_event, initargs=(event,)
                            ) as pool:
    pool.map(do_smth, all)

if __name__ == "__main__":
    all = ["1", "2"]
    make_a_pool(all)

Output (you didn't advance a on value=="1"):

process_2 0
process_1 0
process_1 0
process_2 1
process_1 0
process_2 2
process_2 3
process_1 0
process_2 4
process_1 0
process_2 5
process_2 6
process_2 7
process_2 8
process_2 9

Collectives™ on Stack Overflow

Pass variables between Pool workers?

2 Answers 2

6 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

6 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related