Как накапливать результаты от вызова pool.apply_async? - PullRequest
Новая
       29

Как накапливать результаты от вызова pool.apply_async?

1 голос
/ 16 октября 2019

Я хочу сделать вызовы pool.apply_async (func) и накапливать результаты, как только они станут доступны, не ожидая друг друга. 


import multiprocessing
import numpy as np

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):
    print('%s %d' %(chrName,simNum))

    result=[]
    signal_array=np.full((10000,), simNum, dtype=float)
    count_array = np.full((10000,), simNum, dtype=int)
    result.append(signal_array)
    result.append(count_array)

    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)

    for chrName in chrNames:
        for simNum in sims:
            result= pool.apply_async(func, (chrName,simNum,))
            accumulate_chrBased_simBased_result(result.get(),accumulatedSignalArray,accumulatedCountArray)

    pool.close()
    pool.join()

    print(accumulatedSignalArray)
    print(accumulatedCountArray)

Таким образом, каждый вызов pool.apply_async ожидаетдля другого звонка, чтобы закончить. Есть ли способ избавиться от этого ожидания друг друга?

1 Ответ

1 голос
/ 16 октября 2019

Вы используете result.get() на каждой итерации и заставляете основной процесс ждать готовности функции к этому.

Ниже приведена рабочая версия с распечатками, показывающими, что накопление выполнено, когдаФункция «func» готова и добавляет случайные сны для обеспечения значительных различий во времени выполнения. 

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)
...