Python 并行矩阵向量乘法

Question

我正在尝试将矩阵的每一列乘以向量元素。

我有一个可以正常工作的串行解决方案。

    for i in range(0,temp.shape[0]):
        for j in range(0,temp.shape[1]):
             temp[i,j] = temp[i,j] * h[i,0]

以下是适用于我正在尝试做的事情的并行解决方案，但不会返回与上述代码相同的矩阵。

def mv_mult(start_col,end_col,M,V):
for i in range(0,M.shape[0]):
    for j in range(start_col,end_col):
        M[i,j] = M[i,j] * V[i,0]

num_threads = multiprocessing.cpu_count()
threads = []
extra = temp.shape[1] % num_threads
start_col = 0
jump = temp.shape[1] / num_threads
for i in range(0,num_threads):
    print 'starting thread ', i
    if (i <  extra)
        args = start_col, start_col+jump+1,temp,h
        p = multiprocessing.Process(target=mv_mult,args=args)
        p.start()
        threads.append(p)
        start_col += jump+1
    else:
        args = start_col, start_col+jump,temp,h
        p = multiprocessing.Process(target=mv_mult,args=args)
        p.start()
        threads.append(p)
        start_col += jump

for i in threads:
    i.join()

我对 Python 比较陌生，但据我所知，一切都是通过引用传递的，所以传递给每个新进程的临时矩阵是同一个对象，所以它应该与串行解决方案一样工作，减去事实它是按列拆分的。

关于它为什么不起作用的任何想法？

Answer 1

假设你有一个数组，首先尝试使用 numpy 的向量化计算：

M[:,startcol:endcol] *= V[:,0]

您可能会发现这比您可以执行的任何并行化技巧都快，如果不是，您最好使用类似的行并将矩阵的块发送到每个进程，而不是单个元素。

Answer 2

这只是 numpy multiply 操作：

import numpy

a = numpy.matrix(numpy.ones((5,5)))
b = numpy.arange(5).reshape(5,1)
print a
print b    
#[[ 1.  1.  1.  1.  1.]
# [ 1.  1.  1.  1.  1.]
# [ 1.  1.  1.  1.  1.]
# [ 1.  1.  1.  1.  1.]
# [ 1.  1.  1.  1.  1.]]
#[[0]
# [1]
# [2]
# [3]
# [4]]

c = numpy.multiply(a,b)
print c
#[[ 0.  0.  0.  0.  0.]
# [ 1.  1.  1.  1.  1.]
# [ 2.  2.  2.  2.  2.]
# [ 3.  3.  3.  3.  3.]
# [ 4.  4.  4.  4.  4.]]

（请注意，在您的串行版本中，您是在乘以行，而不是列）。

将此时间与您的序列版本进行比较：

#!/usr/bin/env python

import numpy
import numpy.random
import numpy.linalg
import sys
import time

def init(n):
    a = numpy.matrix(numpy.random.rand(n,n))
    b = numpy.random.rand(n,1)
    b = b.reshape((n,1))
    return a,b

def np_mult(a, b):
    c = numpy.multiply(a,b)
    return c

def manual_mult(a,b):
    c = numpy.matrix(numpy.zeros_like(a))
    for i in range(0,a.shape[0]):
        for j in range(0,a.shape[1]):
             c[i,j] = a[i,j] * b[i]
    return c

def main(argv=None):
    if argv is None:
        argv = sys.argv

    n = int(argv[1])
    niters = int(argv[2])
    a,b = init(n)

    start = time.time()
    for i in xrange(niters):
        cnp = np_mult(a,b)
    end = time.time()

    nptime = (end-start)/niters
    print 'Numpy: ', nptime

    start = time.time()
    for i in xrange(niters):
        cm = manual_mult(a,b)
    end = time.time()

    manualtime = (end-start)/niters
    print 'Manual: ', manualtime
    print 'Speed difference = ', manualtime/nptime
    print 'Diff = ', numpy.linalg.norm(cm - cnp)/numpy.linalg.norm(cnp)

    return 0

if __name__ == "__main__":
    sys.exit(main())

使用一些小尺寸跑步可以：

$ ./mult.py 50 5
Numpy:  3.48091125488e-05
mManual:  0.021101808548
Speed difference =  606.215068493
Diff =  0.0

$ ./mult.py 100 5
Numpy:  6.59942626953e-05
Manual:  0.0830503940582
Speed difference =  1258.44869942
Diff =  0.0

$ ./mult.py 500 5
Numpy:  0.000931406021118
Manual:  2.08903641701
Speed difference =  2242.88481032
Diff =  0.0

速度差异使您可以通过并行性弥补的任何东西都相形见绌。

Answer 3

其他人已经提出了解决您问题的正确方法，我只是想回应一下：

所以传递给每个新进程的临时矩阵是同一个对象，所以 它应该与串行解决方案一样工作，

当您尝试进行多处理时，这会发生故障——并行进程不能有任何共享内存，因此它们不能同时访问同一个数组。我相信您正在将数组的 N_WORKERS 副本传递给工作人员。每个工人在他自己的副本上工作，然后在他停下来时扔掉结果。我希望您在最后得到与开始时相同的矩阵。

如果您想采用这种方法，您需要让他们在最后返回自己的副本，然后将它们重新合并在一起（这会扼杀您可能获得的任何效率提升）。