python - 多重处理不适用于计算矩阵叉积的函数

Question

我正在尝试比较函数的非并行版本和并行版本的运行时间。问题是，虽然并行函数与线程包配合得很好，但在我切换到多处理包后，进程永远不会启动。我想知道这是由我的编译器还是其他原因引起的。任何人都可以运行我的代码来看看它在其他环境中是否有效吗？如果没有，我的代码中存在什么问题？

import numpy as np
from multiprocessing import Process

def single_row(a,b,output):
    for j in range(len(b[0])):
        for k in range(len(a)):
            output[j]=output[j]+a[k]*b[k][j]

#Parallel Matrix Cross Multiplication
def cross_parallel(a,b):
    if len(a[0])==len(b):
        tasks=[None]*len(a)
        T=np.array([[0]*len(b[0])]*len(a))
        for i in range(len(a)):
            tasks[i]=Process(target=single_row,args=(a[i],b,T[i]))
        for task in tasks:
            task.start()
        for task in tasks:
            task.join()
        return T
    else:
        print 'Error: Invalid Matrices'

#Non-parallel Matrix Cross Multiplication
def cross_basic(a,b):
    if len(a[0])==len(b):
        T=np.array([[0]*len(b[0])]*len(a))
        for i in range(len(a)):
            for j in range(len(b[0])):
                for k in range(len(a[0])):
                    T[i][j]=T[i][j]+a[i][k]*b[k][j]
        return T
    else:
        print 'Error: Invalid Matrices'

if __name__ == '__main__':     
    x=[[1,2,3,4],[5,6,7,8],[9,10,11,12],[13,14,15,16]]
    y=[[1,2,3,4],[5,6,7,8],[9,10,11,12],[13,14,15,16]]
    print cross_basic(x,y)
    print cross_parallel(x,y)

结果:

[[ 90 100 110 120]
 [202 228 254 280]
 [314 356 398 440]
 [426 484 542 600]]
[[0 0 0 0]
 [0 0 0 0]
 [0 0 0 0]
 [0 0 0 0]]

使用线程包的版本有效(仅第15行不同):

import numpy as np
from threading import Thread

def single_row(a,b,output):
    for j in range(len(b[0])):
        for k in range(len(a)):
            output[j]=output[j]+a[k]*b[k][j]

#Parallel Matrix Cross Multiplication
def cross_parallel(a,b):
    if len(a[0])==len(b):
        tasks=[None]*len(a)
        T=np.array([[0]*len(b[0])]*len(a))
        for i in range(len(a)):
            tasks[i]=Thread(target=single_row,args=(a[i],b,T[i]))
        for task in tasks:
            task.start()
        for task in tasks:
            task.join()
        return T
    else:
        print 'Error: Invalid Matrices'

#Non-parallel Matrix Cross Multiplication
def cross_basic(a,b):
    if len(a[0])==len(b):
        T=np.array([[0]*len(b[0])]*len(a))
        for i in range(len(a)):
            for j in range(len(b[0])):
                for k in range(len(a[0])):
                    T[i][j]=T[i][j]+a[i][k]*b[k][j]
        return T
    else:
        print 'Error: Invalid Matrices'

if __name__ == '__main__':     
    x=[[1,2,3,4],[5,6,7,8],[9,10,11,12],[13,14,15,16]]
    y=[[1,2,3,4],[5,6,7,8],[9,10,11,12],[13,14,15,16]]
    print cross_basic(x,y)
    print cross_parallel(x,y)

结果:

[[ 90 100 110 120]
 [202 228 254 280]
 [314 356 398 440]
 [426 484 542 600]]
[[ 90 100 110 120]
 [202 228 254 280]
 [314 356 398 440]
 [426 484 542 600]]

Answer 1

当您使用线程时，结果矩阵T在线程中共享(这意味着它基本上是相同的对象，使用相同的内存插槽)。因此，在子 Thread 中修改 T 将修改 T 的本地版本，并且您会得到正确的结果。

对于子流程，每个子Process都会获得T的新副本。因此，子进程中T的修改不会修改其本地版本。为了获得正确的结果，您需要发送回计算结果，例如使用Queue。但您必须小心，因为返回结果的顺序是不确定的。

import numpy as np
from multiprocessing import Process, Queue


def single_row(a, b, idx, q):
    N = len(b[0])
    output = np.zeros(N)
    for j in range(len(b[0])):
        for k in range(len(a)):
            output[j] = output[j]+a[k]*b[k][j]
    q.put((idx, output))

# Parallel Matrix Cross Multiplication


def cross_parallel(a, b):
    M = len(a)
    q = Queue()
    if len(a[0]) == len(b):
        tasks = [None]*M
        T = np.array([[0]*len(b[0])]*len(a))
        for i in range(M):
            tasks[i] = Process(target=single_row, args=(a[i], b, i, q))
        for task in tasks:
            task.start()
        T = []
        for i in range(M):
            T += [q.get()]
        for task in tasks:
            task.join()
        T.sort()
        T = np.array([v[1] for v in T])
        return T
    else:
        print('Error: Invalid Matrices')

# Non-parallel Matrix Cross Multiplication


def cross_basic(a, b):
    if len(a[0]) == len(b):
        T = np.array([[0]*len(b[0])]*len(a))
        for i in range(len(a)):
            for j in range(len(b[0])):
                for k in range(len(a[0])):
                    T[i][j] = T[i][j]+a[i][k]*b[k][j]
        return T
    else:
        print('Error: Invalid Matrices')

if __name__ == '__main__':
    x = [[1, 2, 3, 4], [5, 6, 7, 8], [9, 10, 11, 12], [13, 14, 15, 16]]
    y = [[1, 2, 3, 4], [5, 6, 7, 8], [9, 10, 11, 12], [13, 14, 15, 16]]
    print(cross_basic(x, y))
    print(cross_parallel(x, y))

这种方法可能不太好用，例如，如果在加入进程之前不清空队列，可能会导致程序死锁。我建议查看 multiprocessing.Pool 或 concurrents.futures.ProcessPoolExecutor 以获得以更好的方式管理通信/进程数量的高级协议(protocol)。

def single_row2(a, b):
    N = len(b[0])
    output = np.zeros(N)
    for j in range(len(b[0])):
        for k in range(len(a)):
            output[j] = output[j]+a[k]*b[k][j]
    return output

def cross_parallel2(a, b):
    import itertools
    from concurrent.futures import ProcessPoolExecutor
    executor = ProcessPoolExecutor(max_workers=4)
    M = len(a)
    if len(a[0]) == len(b):
        res = executor.map(single_row2, a, itertools.repeat(b))

        return np.array([row for row in res])
    else:
        print('Error: Invalid Matrices')