CODEWITHSUNDEEP
pythonmultiprocessing
burcak
burcak

Reputation: 1147

Is there a way to accumulate the result of a pool.apply_async calls whenever it is available without collecting them in a list like structure?

In for loops, I are sending jobs using python pool.apply_async() calls.

The python code below was initially written by me and then edited by @Santiago Magariños 

import multiprocessing
import numpy as np
from time import time, sleep
from random import random

chrNames=['chr1','chr2','chr3']
sims=[1,2,3]



def accumulate_chrBased_simBased_result(chrBased_simBased_result,accumulatedSignalArray,accumulatedCountArray):    
    signalArray = chrBased_simBased_result[0]
    countArray = chrBased_simBased_result[1]

    accumulatedSignalArray += signalArray
    accumulatedCountArray += countArray


def func(chrName,simNum):

    result=[]
    sleep(random()*5)
    signal_array=np.full((10000,), simNum, dtype=float)
    count_array = np.full((10000,), simNum, dtype=int)
    result.append(signal_array)
    result.append(count_array)
    print('%s %d' %(chrName,simNum))

    return result


if __name__ == '__main__':

    accumulatedSignalArray = np.zeros((10000,), dtype=float)
    accumulatedCountArray = np.zeros((10000,), dtype=int)

    numofProcesses = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(numofProcesses)

    results = []
    for chrName in chrNames:
        for simNum in sims:
            results.append(pool.apply_async(func, (chrName,simNum,)))

    for i in results:
        print(i)

    while results:
        for r in results[:]:
            if r.ready():
                print('{} is ready'.format(r))
                accumulate_chrBased_simBased_result(r.get(),accumulatedSignalArray,accumulatedCountArray)
                results.remove(r)

    pool.close()
    pool.join()

    print(accumulatedSignalArray)
    print(accumulatedCountArray)

Is there a way to accumulate the result of a pool.apply_async() call whenever it is available without collecting them in a list like structure?

Upvotes: 0

Views: 222

Answers (1)

Jose A. García
Jose A. García

Reputation: 888

Something like this could work. Pretty much copying your code and adding a callback, note that this works because the pool is joined before we access the accumulator values. If not we'd need some other type of synchronization mechanism

class Accumulator:
    def __init__(self):
        self.signal = np.zeros((10000,), dtype=float)
        self.count = np.zeros((10000,), dtype=int)

    def on_result(self, result):
        self.signal += result[0]
        self.count += result[1]

if __name__ == '__main__':

    num_proc = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(num_proc)

    accumulator = Accumulator()
    for chrName in chrNames:
        for simNum in sims:
            pool.apply_async(func, (chrName,simNum,), callback=accumulator.on_result)

    pool.close()
    pool.join()

    print(accumulator.signal)
    print(accumulator.count)

Upvotes: 2

Related Questions