Python Ошибка обработки нескольких данных

Question

Я стараюсь многопроцессорство в Python. Я написал код, который добавляет вектор, но не смог получить выход из функции. Что значит, выход Z выводит 0, а не 2.

from multiprocessing import Process 
import numpy as np 

numThreads = 16 
num = 16 

numIter = num/numThreads 

X = np.ones((num, 1)) 
Y = np.ones((num, 1)) 
Z = np.zeros((num, 1)) 

def add(X,Y,Z,j): 
    Z[j] = X[j] + Y[j] 

if __name__ == '__main__': 
    jobs = [] 
    for i in range(numThreads): 
    p = Process(target=add, args=(X, Y, Z, i,)) 
    jobs.append(p) 

    for i in range(numThreads): 
    jobs[i].start() 

    for i in range(numThreads): 
    jobs[i].join() 

    print Z[0] 

Edit: Принял совет Clocker и изменил код для этого:

import multiprocessing 
import numpy as np 

numThreads = 16 
numRows = 32000 
numCols = 2 
numOut = 3 

stride = numRows/numThreads 

X = np.ones((numRows, numCols)) 
W = np.ones((numCols, numOut)) 
B = np.ones((numRows, numOut)) 
Y = np.ones((numRows, numOut)) 

def conv(idx): 
    Y[idx*stride:idx*stride+stride] = X[idx*stride:idx*stride+stride].dot(W) + B[idx*stride:idx*stride+stride] 

if __name__=='__main__': 
    pool = multiprocessing.Pool(numThreads) 
    pool.map(conv, range(numThreads)) 
    print Y 

И выход Y вместо Saxp ,

Answer 1

Причина, по которой ваша последняя строка print Z[0] возвращает [0] вместо [2], заключается в том, что каждый из процессов делает независимую копию Z (или может быть Z[j] - не совсем уверен в этом) перед ее модификацией. В любом случае, отдельный процесс будет гарантировать, что исходная версия не изменится.

Если бы вы использовали threading module, вместо этого последняя строка действительно вернула бы [2], как ожидалось, но это не многопроцессорная обработка.

Итак, вы, скорее всего, захотите использовать multiprocessing.Pool. Проводя эксперимент исключительно для иллюстрации, можно было бы сделать следующее:

In [40]: pool = multiprocessing.Pool() 
In [41]: def add_func(j): 
    ....:  return X[j] + Y[j] 
In [42]: pool = multiprocessing.Pool(numThreads) 
In [43]: pool.map(add_func, range(numThreads)) 
Out[43]: 
[array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.]), 
array([ 2.])] 

Удачи!

Для вашей второй части на ваш вопрос проблема в том, что функция conv() не возвращает никакого значения. Хотя пул процессов получает копию X, B и W для вытягивания значений, Y внутри conv() является локальным для каждого запущенного процесса. Для того, чтобы получить новое вычисленное значение Y, вы бы использовать что-то вроде этого:

def conv(idx): 
    Ylocal_section = X[idx*stride:idx*stride+stride].dot(W) + B[idx*stride:idx*stride+stride] 
    return Ylocal_section 

results = pool.map(conv, range(numThreads)) # then apply each result to Y 
for idx in range(numThreads): 
    Y[idx*stride:idx*stride+stride] = results[idx] 

параллелизм может усложниться очень быстро, и в этот момент я хотел бы оценить существующие библиотеки, которые могут выполнять быстрый 2D свертка. numpy и scipy библиотеки могут быть суперэффективными и могут лучше удовлетворять ваши потребности.