python - 如何在Python中并行化嵌套for循环？

Question

好吧，这是我的问题:我的程序中有一个嵌套的 for 循环，它在单核上运行。由于程序 99% 以上的运行时间都花在这个嵌套的 for 循环中，我想对其进行并行化。现在我必须等待 9 天才能完成计算。我尝试使用 multiprocessing 库实现并行 for 循环。但我只找到非常基本的例子，无法将它们转移到我的问题上。以下是带有随机数据的嵌套循环:

import numpy as np

dist_n = 100
nrm = np.linspace(1,10,dist_n)

data_Y = 11000
data_I = 90000
I = np.random.randn(data_I, 1000)
Y = np.random.randn(data_Y, 1000)
dist = np.zeros((data_I, dist_n)

for t in range(data_Y):
    for i in range(data_I):
        d = np.abs(I[i] - Y[t])
        for p in range(dist_n):
            dist[i,p] = np.sum(d**nrm[p])/nrm[p]

    print(dist)

请给我一些如何使其并行的建议。

Answer 1

启动进程的开销很小(50ms+，具体取决于数据大小)，因此通常最好 MP 尽可能大的代码块。从您的评论看来，t 的每个循环都是独立的，因此我们应该可以自由地并行化它。

当 python 创建一个新进程时，您会获得主进程的副本，因此您可以使用所有全局数据，但是当每个进程写入数据时，它会写入自己的本地副本。这意味着 dist[i,p] 将不可用于主进程，除非您显式地将其返回并返回(这将产生一些开销)。在您的情况下，如果每个进程都将 dist[i,p] 写入文件，那么您应该没问题，只是不要尝试写入同一个文件，除非您实现某种类型的互斥访问控制。

#!/usr/bin/python
import time
import multiprocessing as mp
import numpy as np

data_Y = 11 #11000
data_I = 90 #90000
dist_n = 100
nrm = np.linspace(1,10,dist_n)
I = np.random.randn(data_I, 1000)
Y = np.random.randn(data_Y, 1000)
dist = np.zeros((data_I, dist_n))

def worker(t):
    st = time.time()
    for i in range(data_I):
        d = np.abs(I[i] - Y[t])
        for p in range(dist_n):
            dist[i,p] = np.sum(d**nrm[p])/nrm[p]
    # Here - each worker opens a different file and writes to it
    print 'Worker time %4.3f mS' % (1000.*(time.time()-st))


if 1:   # single threaded
    st = time.time()
    for x in map(worker, range(data_Y)):
        pass
    print 'Single-process total time is %4.3f seconds' % (time.time()-st)
    print

if 1:   # multi-threaded
    pool = mp.Pool(28) # try 2X num procs and inc/dec until cpu maxed
    st = time.time()
    for x in pool.imap_unordered(worker, range(data_Y)):
        pass
    print 'Multiprocess total time is %4.3f seconds' % (time.time()-st)
    print

如果再次增加 data_Y/data_I 的大小，加速比应该会增加到理论极限。