Многопроцессорная обработка для вычисления собственного значения

Question

Я генерирую 100 случайных матриц с размером 1000x1000. Я использую модуль многопроцессорности для вычисления собственных значений 100 матриц.

код приведен ниже:

import timeit 
import numpy as np 
import multiprocessing as mp 

def calEigen(): 

S, U = np.linalg.eigh(a) 

def multiprocess(processes): 
pool = mp.Pool(processes=processes) 
#Start timing here as I don't want to include time taken to initialize the processes 
start = timeit.default_timer() 
results = [pool.apply_async(calEigen, args=())] 
stop = timeit.default_timer() 
print (processes":", stop - start) 

results = [p.get() for p in results] 
results.sort() # to sort the results 


if __name__ == "__main__": 

global a 
a=[] 

for i in range(0,100): 
    a.append(np.random.randint(1,100,size=(1000,1000))) 

#Print execution time without multiprocessing 
start = timeit.default_timer() 
calEigen() 
stop = timeit.default_timer() 
print stop - start 

#With 1 process 
multiprocess(1) 

#With 2 processes 
multiprocess(2) 

#With 3 processes 
multiprocess(3) 

#With 4 processes 
multiprocess(4) 

Выход

0.510247945786 
('Process:', 1, 5.1021575927734375e-05) 
('Process:', 2, 5.698204040527344e-05) 
('Process:', 3, 8.320808410644531e-05) 
('Process:', 4, 7.200241088867188e-05) 

Другой итерация показал этот результат:

69.7296020985 
('Process:', 1, 0.0009050369262695312) 
('Process:', 2, 0.023727893829345703) 
('Process:', 3, 0.0003509521484375) 
('Process:', 4, 0.057518959045410156) 

Мои вопросы таковы:

1. Почему время выполнения не уменьшается по мере увеличения числа процессов ? Я правильно использую модуль многопроцессорности?
2. Правильно ли я рассчитываю время выполнения?

Я отредактировал код, указанный в комментариях ниже. Я хочу, чтобы последовательные и многопроцессорные функции находили собственные значения для того же списка из 100 матриц. Код редакции:

import numpy as np 
import time 
from multiprocessing import Pool 

a=[] 

for i in range(0,100): 
a.append(np.random.randint(1,100,size=(1000,1000))) 

def serial(z): 
result = [] 
start_time = time.time() 
for i in range(0,100):  
    result.append(np.linalg.eigh(z[i])) #calculate eigen values and append to result list 
end_time = time.time() 
print("Single process took :", end_time - start_time, "seconds") 


def caleigen(c): 
result = []   
result.append(np.linalg.eigh(c)) #calculate eigenvalues and append to result list 
return result 

def mp(x): 
start_time = time.time() 
with Pool(processes=x) as pool: # start a pool of 4 workers 
    result = pool.map_async(caleigen,a) # distribute work to workers 
    result = result.get() # collect result from MapResult object 
end_time = time.time() 
print("Mutltiprocessing took:", end_time - start_time, "seconds") 

if __name__ == "__main__": 

serial(a) 
mp(1,a) 
mp(2,a) 
mp(3,a) 
mp(4,a) 

Существует не сокращение времени по мере увеличения количества процессов. Где я иду не так? Многопроцессорность разделяет список на куски для процессов или мне нужно делать разделение?

Answer 1

Вы не используете модуль многопроцессорности правильно. Как отметил @dopstar, вы не делите свою задачу. Для пула процессов есть только одна задача, поэтому неважно, сколько рабочих вы назначили, только один получит работу. Что касается вашего второго вопроса, я не использовал timeit для точного измерения времени процесса. Я просто использую модуль time, чтобы получить грубое представление о том, насколько быстрыми являются вещи. Тем не менее, он служит цели большую часть времени. Если я понимаю, что вы пытаетесь сделать правильно, то это должно быть единой версией процесса кода

import numpy as np 
import time 

result = [] 
start_time = time.time() 
for i in range(100): 
    a = np.random.randint(1, 100, size=(1000,1000)) #generate random matrix 
    result.append(np.linalg.eigh(a))     #calculate eigen values and append to result list 
end_time = time.time() 
print("Single process took :", end_time - start_time, "seconds") 

Одиночная версия процесса занимает 15,27 секунды на моем компьютере. Ниже приведена версия многопроцессорности, которая заняла всего 0,46 секунды на моем компьютере. Я также включил версию для одного процесса для сравнения. (Одна версия процесса должна быть заключена в блок if, а также размещена после многопроцессорной версии.) Поскольку вы хотели бы повторить свой расчет в течение 100 раз, было бы намного проще создать пул работников и позволить они автоматически выполняют незавершенную задачу, чем вручную запускают каждый процесс и определяют, что должен делать каждый процесс. Здесь, в моих кодах, аргумент для вызова caleigen - это просто отслеживать, сколько раз выполнение задачи выполнялось. Наконец, , как правило, быстрее, чем apply_async, причем его недостаток потребляет немного больше памяти и принимает только один аргумент для вызова функции. Причина использования map_async, но не map, заключается в том, что в этом случае порядок, в котором возвращается результат, не имеет значения, а map_async намного быстрее, чем map.

from multiprocessing import Pool 
import numpy as np 
import time 

def caleigen(x):  # define work for each worker 
    a = np.random.randint(1,100,size=(1000,1000)) 
    S, U = np.linalg.eigh(a)       
    return S, U 


if __name__ == "main": 
    start_time = time.time() 
    with Pool(processes=4) as pool:  # start a pool of 4 workers 
     result = pool.map_async(caleigen, range(100)) # distribute work to workers 
     result = result.get()  # collect result from MapResult object 
    end_time = time.time() 
    print("Mutltiprocessing took:", end_time - start_time, "seconds") 

    # Run the single process version for comparison. This has to be within the if block as well. 
    result = [] 
    start_time = time.time() 
    for i in range(100): 
     a = np.random.randint(1, 100, size=(1000,1000)) #generate random matrix 
     result.append(np.linalg.eigh(a))     #calculate eigen values and append to result list 
    end_time = time.time() 
    print("Single process took :", end_time - start_time, "seconds")