Python 多处理块中的数组并将结果添加到共享数组

Question

我处理 2 个大型一维数组（假设 A 和 B）：我对 A 和 B 元素对进行操作，并将结果写入共享数组 C（将 C 视为直方图）。我想使用 multiprocessing 来并行化这个过程。我认为最佳方法可能是将数组 A 分割成多个独特的块，这些块等于我选择执行的并行进程的数量，并使用 for 循环对 B 的所有元素进行数学运算。

我正在阅读许多问题/答案。我以Multiprocessing a loop of a function that writes to an array in python 为例，它使用了Process。我试图适应我的问题，但我正在获得串行执行的性能。我正在测试的代码：

from multiprocessing import Process, Lock
from multiprocessing.sharedctypes import Array
import numpy as np
import time


def ProcessData(sub_data1, data2, freq):
    for dat1 in sub_data1:
        for dat2 in data2:
            d = int( np.sqrt( (dat1 - dat2)**2 ) )
            #d = int(dat1 - dat2)
            if (d < len(freq)):
                freq[d] += 1

def SplitList(data, n):
    sub_len = divmod(len(data),n)[0]
    print(sub_len)
    slices = []
    for i in range(n):
        slices.append( data[i*sub_len:i*sub_len+sub_len] )
    return slices
    
def main(nproc):
    print("Number of cpu : ", mp.cpu_count())
    lock = Lock()
    N = 30
    chip = [1,1,1,1,1,2,2,2,2,2,3,3,3,3,4,4,4,4,4,5,5,5,5,5,6,6,6,6,6,7,7,7,7,7,8,8,8,8,8,9,9,9,9]
    data1 = np.array( chip * N )
    data2 = np.array( chip * N )
    freq = Array('i', 100, lock=lock)
    dat1_subs = SplitList(data1,nproc)
    print('Number of data1 slices {:d}'.format(len(dat1_subs)))
    t_start = time.time()
    
    if __name__ == '__main__':
        for i in range(0, nproc):
            print('LEN {:d}: {:d}'.format(i, len(dat1_subs[i] )) )
            p = Process(target=ProcessData, args=(dat1_subs[i], data2, freq))
            p.start()
            p.join()

    t_end =  time.time()
    print('Total time (s)= ' + str(t_end - t_start))
    print(str(list(freq)))
    #new_array = np.frombuffer(freq.get_obj())
    Sum = sum( list(freq) )
    print('Total {:d}'.format(Sum))
    
NProc = 4

main(NProc)

我将不胜感激任何输入或提示我做错了什么。或者也许有更简单的方法我只是不知道。 谢谢。

Answer 1

这样试试

from concurrent.futures import ProcessPoolExecutor
def pool_factorizer_map(nums, nprocs):
    # Let the executor divide the work among processes by using 'map'.
    with ProcessPoolExecutor(max_workers=nprocs) as executor:
        return {num:factors for num, factors inmething like this
                                    executor.map(factorize_naive, nums))}

Answer 2

来自您的评论：

共享数组（累积计数的“freq”）在使用不同数量的 cpu 处理后在某些绑定中的计数略有不同的原因可能是什么？

即使您没有更改 CPU 的数量，如果每次运行都没有得到不同的结果，我会感到惊讶。如果您确实更改了它们，那么“显而易见”的第一个原因是您的 SplitLists() 可以丢弃不同数量的尾随数据，具体取决于传递给它的 n。

但即使不改变进程数，

freq[d] += 1

不是确定性的。是的，Array 类型是跨进程同步的，但这仅适用于它们自己的加载和存储。该操作很复杂，并且在幕后工作就像

with lock:
    temp = freq[d]
temp += 1
with lock:
    freq[d] = temp

没有什么可以阻止多个进程读取相同的当前 freq[d] 值，每个进程自行将其加 1，然后多次存储相同的新值。

要使增量就地原子作为一个整体，您需要传递一个不同的锁（在这种情况下，您也可以使用RawArray），然后执行以下操作：

with lock:
    freq[d] += 1

但是，对该锁的极高争用会降低性能。前提是没有 “太多”可能的d 值，传递len(freq) 不同锁定对象的数组会更好（为了速度），然后执行以下操作：

with locks[d]:
    freq[d] += 1

只有当多个进程碰巧更改相同的d计数时，才会发生锁争用。

但如果len(freq) 不是那么大，我会完全避免共享内存。让每个进程用自己的freq列表全速运行，然后让主程序总结它们。这是一个示例，但它与您发布的代码完全不同。没有 numpy（它与这里的问题无关），也没有你从链接到的帖子中继承的坦率地说奇怪的东西：

def work(raw, freqlen):
    freq = [0] * freqlen
    for x in raw:
        if x < freqlen:
            freq[x] += 1
    return freq

def main(nproc, nfreq, numperchunk=100000):
    import multiprocessing as mp
    base = list(range(200)) * 1000000
    with mp.Pool(processes=nproc) as pool:
        i = 0
        ps = []
        while i < len(base):
            ps.append(pool.apply_async(work,
                                       (base[i : i + numperchunk],
                                        nfreq)))
            i += numperchunk
        result = [0] * nfreq
        for p in ps:
            for i, x in enumerate(p.get()):
                result[i] += x
        print(result)

if __name__ == "__main__":
    main(4, 10)

共享内存实际上可能有帮助的地方：分割一个巨大的向量并将切片传递给工作进程是昂贵的（大量的进程间通信）。最好将只读的巨型向量放在共享内存中，然后将切片索引传递给工作人员。或者，在 Linux 系统上，让工作人员通过写时复制fork() 语义在模块级别继承巨型向量。

权衡

为了使其中一些想法具体化，这里有一个更类似于您的原始想法的变体。但是：

• 它也将“巨型向量”放入共享内存中。
• 由于向量是共享的，因此只需要将切片边界传递给工作程序。需要更少的进程间通信。
• freq 向量也是共享的，但工作人员首先使用自己的本地版本，以实现最高无锁速度。它只在最后持有一个锁，将其本地结果折叠到共享结果中。
• 因为我们正在做自己的锁定RawArrays 被使用。访问这些内容要快得多。

这尖叫。事实上，程序创建测试用例所花费的时间比计算直方图要多得多；-)

def work(base, lo, hi, freq, L):
    freqlen = len(freq)
    myfreq = [0] * freqlen
    for i in range(lo, min(hi, len(base))):
        x = base[i]
        if x < freqlen:
            myfreq[x] += 1
    with L:
        for i, x in enumerate(myfreq):
            freq[i] += x

def main(nproc, nfreq):
    import multiprocessing as mp
    import math
    base = mp.RawArray('h', list(range(201)) * 1000003)
    freq = mp.RawArray('i', nfreq)
    L = mp.Lock()
    numperchunk = math.ceil(len(base) / nproc)
    print(f"{len(base)=:,} {numperchunk=:,}")
    ps = []
    a = 0
    for i in range(nproc):
        p = mp.Process(target=work,
                       args=(base, a, a + numperchunk, freq, L))
        p.start()
        a += numperchunk
        ps.append(p)
    for p in ps:
        p.join()
    print(list(freq))

if __name__ == "__main__":
    main(4, 10)