如何使用共享可写内存并行化代码

Question

我有一个目前非常慢且 CPU 很重的程序，我对如何并行化它感到困惑。

我的问题：

算法是这样工作的（使用随机数）：

import numpy as np
import scipy.spatial as spatial
from collections import Counter

# Particle dictionary
num_parts = int(1e4)
particles = {'coords': np.random.rand(num_parts, 3),
             'flags': np.random.randint(0, 4, size=num_parts)}

# Object dictionary
num_objects = int(1e3)
objects_dict = {'x': [np.random.rand(1)
                      for i in range(0, num_objects)],
                'y': [np.random.rand(1)
                      for i in range(0, num_objects)],
                'z': [np.random.rand(1)
                      for i in range(0, num_objects)],
                'prop_to_calc': [np.array([])
                                 for i in range(0, num_objects)]}

# Build KDTree for particles
tree = spatial.cKDTree(particles['coords'])

# Loop through objects to calculate 'prop_to_calc' based
# on nearby particles flags within radius, r.
r = 0.1
for i in range(0, num_objects):

    # Find nearby indices of particles.
    indices = tree.query_ball_point([objects_dict['x'][i][0],
                                     objects_dict['y'][i][0],
                                     objects_dict['z'][i][0]],
                                    r)

    # Extract flags of nearby particles.
    flags_array = particles['flags'][indices]

    # Find most common flag and store that in property_to_calculate
    c = Counter(flags_array)
    objects_dict['prop_to_calc'] = np.append(objects_dict['prop_to_calc'],
                                             c.most_common(1)[0][0])

有两个数据集particles 和objects_dict。我想通过搜索半径内的附近粒子r 并找到它们最常见的标志来计算objects_dict['prop_to_dict']。这是通过cKDTree 和query_ball_point 完成的。

对于这些数字，时间是： 10 loops, best of 3: 55.8 ms per loop 在 Ipython 4.2.0 中

但是，我想要num_parts=1e6 和num_objects=1e5，这会导致速度严重下降。

我的问题：

由于 CPU 很重，我想尝试并行化它以加快速度。

我查看了multiprocessing 和multi threading。但是文档让我很困惑，我不确定如何将示例应用于此问题。

具体来说，我关心的是如何在进程之间共享这两个字典并在最后写入objects_dict。

提前感谢您的帮助。

Answer 1

尝试拆分代码并制作本地列表和字典。联合最后的字典和列表。只需import threading 并创建一个带有子类threading.Thread 的class CalculationThread 并在类的方法run 中编写并行代码，并通过thread = CalculationThread() 和thread.start() 启动线程

这个的最终代码是：

import threading

class CalculationThread(threading Tread):
    def __init__(self, range):
        self.range = range
        # use this copy instead for the thread
        self.objects_dict = dict(objects_dict)
    def run(self):
        for i in self.range:
            # your code

threads = []
threadCount = 8

lastThreadStart = num_objects
tLen = num_objects // threadCount
for i in reversed(range(threadCount)):
    threadStart = i*tLen
    t = CalculationThread(range(threadStart, lastThreadStart))
    lastThreadStart = threadStart

# wait for all to finish
for t in threads: t.join()

# code to union all objects_dicts comes here