Python 多处理性能

这应该是我的第三个也是最后一个问题，涉及我尝试提高使用 python 进行的一些统计分析的性能。我的代码有 2 个版本（单核与多处理），我希望通过使用多个核心来获得性能，因为我希望我的代码能够解压/解压相当多的二进制字符串，遗憾的是我注意到使用多个核心实际上会降低性能核心。

我想知道是否有人对我观察到的现象有可能的解释（向下滚动至 4 月 16 日更新以了解更多信息)?

程序的关键部分是函数 numpy_array （+多处理中的解码），代码片段如下（完整代码可通过 Pastebin 访问，如下）：

def numpy_array(data, peaks):
    rt_counter=0
    for x in peaks:
        if rt_counter %(len(peaks)/20) == 0:
            update_progress()
        peak_counter=0
        data_buff=base64.b64decode(x)
        buff_size=len(data_buff)/4
        unpack_format=">%dL" % buff_size
        index=0
        for y in struct.unpack(unpack_format,data_buff):
            buff1=struct.pack("I",y)
            buff2=struct.unpack("f",buff1)[0]
            if (index % 2 == 0):
                data[rt_counter][1][peak_counter][0]=float(buff2)
            else:
                data[rt_counter][1][peak_counter][1]=float(buff2)
                peak_counter+=1
            index+=1
        rt_counter+=1

多处理版本通过一组函数执行此操作，我将在下面显示键 2：

def tonumpyarray(mp_arr):
    return np.frombuffer(mp_arr.get_obj())

def numpy_array(shared_arr,peaks):
    processors=mp.cpu_count()
    with contextlib.closing(mp.Pool(processes=processors,
                                    initializer=pool_init,
                                    initargs=(shared_arr, ))) as pool:
        chunk_size=int(len(peaks)/processors)
        map_parameters=[]
        for i in range(processors):
            counter = i*chunk_size
            chunk=peaks[i*chunk_size:(i+1)*chunk_size]
            map_parameters.append((chunk, counter))
        pool.map(decode,map_parameters)

def decode ((chunk, counter)):
    data=tonumpyarray(shared_arr).view(
        [('f0','<f4'), ('f1','<f4',(250000,2))])
    for x in chunk:
        peak_counter=0
        data_buff=base64.b64decode(x)
        buff_size=len(data_buff)/4
        unpack_format=">%dL" % buff_size
        index=0
        for y in struct.unpack(unpack_format,data_buff):
            buff1=struct.pack("I",y)
            buff2=struct.unpack("f",buff1)[0]
            #with shared_arr.get_lock():
            if (index % 2 == 0):
                data[counter][1][peak_counter][0]=float(buff2)
            else:
                data[counter][1][peak_counter][1]=float(buff2)
                peak_counter+=1
            index+=1
        counter+=1

可以通过这些pastebin链接访问完整的程序代码

Pastebin 单核版本 http://pastebin.com/tSqc7Ntj

Pastebin 多处理版本 http://pastebin.com/FWD2SMWm

我使用包含 239 个时间点和每个时间点约 180k 测量对的文件观察到的性能对于单核而言约为 2.5m，对于多处理而言约为 3.5m。

PS：前两个问题（我第一次尝试并行化）：

1. Python 多处理 https://stackoverflow.com/q/15966157/1093485
2. 使我的 NumPy 数组跨进程共享 https://stackoverflow.com/q/15976937/1093485

——4月16日——

我一直在使用 cProfile 库分析我的程序（具有cProfile.run('main()')在里面__main__，这表明有 1 个步骤会减慢一切：

ncalls  tottime  percall  cumtime  percall filename:lineno(function)
23   85.859    3.733   85.859    3.733 {method 'acquire' of 'thread.lock' objects}

我在这里不明白的是thread.lock对象用于threading（据我的理解）但不应该在多处理中使用，因为每个核心应该运行一个线程（除了有自己的锁定机制），那么这是如何发生的以及为什么单个调用需要 3.7 秒？

共享数据是一种已知的因同步而导致速度变慢的情况。

您可以在进程之间拆分数据，或者为每个进程提供独立的副本吗？然后，在所有计算完成之前，您的流程不需要同步任何内容。

然后，我让主进程将所有工作处理器的输出加入到一个连贯的集合中。

该方法可能需要额外的 RAM，但现在 RAM 很便宜。

如果你问，我也对每个线程锁获取 3700 毫秒感到困惑。对于这样的特殊调用，OTOH 分析可能会出现错误。