Multiprocess a generator method only working once for each process

Question

I have a generator that looks kind of like this:

class GeneratorClass():
    def __init__(self, objClient):
        self.clienteGen = objClient

    def generatorDataClient(self):
        amount = 0
        while True:
            amount += random.randint(0, 2000)
            yield amount
            sleep = random.choice([1,2,3])
            print("sleep " + str(sleep))
            time.sleep(sleep)

Then I iterate through it, which works: it does the current_mean() method each time new data is generated.

def iterate_clients(pos):
    genobject4 = GeneratorClass(client_list[pos])
    generator4 = genobject4.generatorDataClient()    
    current_client = genobject1.default_client
    account1 = current_client.account
    cnt = 0
    acc_mean = 0

    for item in generator4:
        #We call a function previously defined
        acc_mean, cnt = account1.current_mean(acc_mean, item, cnt)
        print("media : " + str(acc_mean), str(cnt))

# iterate_clients(2)

And it works, you give it a valid client, it starts doing the generation operation, which is a moving average, and since it is defined with a While: True it does not stop.

Now I wanted to paralellize this and I managed to get it work, but only once:

names = ["James", "Anna"]
client_list = [Cliente(name) for name in names]
array_length = len(client_list)


import multiprocessing
if __name__ == '__main__':
  for i in range(array_length):
    p = multiprocessing.Process(target=iterate_clients, args=(i,))
    p.start()

But instead each process starts, iterates exactly once, then stops. The result is the following:

calling object with ID: 140199258213624
calling the generator
moving average : 4622.0 1
calling object with ID: 140199258211160
sleep 2
calling the generator
moving average : 8013.0 1
sleep 1

I am sure the code can be improved but could it be I am missing some information on how to parallelize this problem in particular?

Edit:

Thanks to this answer I tried changing the loop from for i in range(array_length): to while True:

And I got something new:

calling object 140199258211160
calling the generator
calling object 140199258211160
moving average : 7993.0 1
calling the generator
duerme 3
calling object 140199258211160
calling the generator
calling object 140199258211160
moving average : 8000.0 1
moving average : 7869.0 1
duerme 3
calling the generator

And it never stops. So from this I get that I am making a huge mistake, because only 1 process gets created, and it has what seems to be a race condition, since the moving average goes back and forth and it only goes up in a normal process.

score 1 · Accepted Answer · answered Nov 11 '19 at 09:19

The issue here is probably that the child processes are terminating once the main process finishes so they never get a chance to fully run.

Using .join() will make the main process wait for the child processes.

...

import multiprocessing

procs = []

if __name__ == '__main__':
  for i in range(array_length):
    p = multiprocessing.Process(target=iterate_clients, args=(i,))
    p.start() 
    procs.append(p)  # Hold a reference to each child process

for proc in procs:
  proc.join()  # Wait for each child process

This does seem to work!! I am going to triple check but it does exchange between processes and keeps the average from ever stopping, so thank you ver much! — monkey intern, Nov 11 '19 at 10:16

Multiprocess a generator method only working once for each process

1 Answers1