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]

编辑:采纳了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]返回而不是2的原因是，每个进程在修改Z之前都会生成一个独立的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的副本以从中提取值，但是conv()中的Y对于启动的每个进程都是本地的。要获得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库可以非常高效，并且可能更好地满足您的需求。