python map_async，开销来自哪里？

Question

我按预期使用map_async - 使用以下方法将可迭代映射到多个处理核心:

cores = mp.cpu_count()
pool = mp.Pool()

r = pool.map_async(func, offsets,callback=mycallback)
r.wait()

func 返回一个字典，因此回调使用以下方式“合并”字典:

ddict = defaultdict(set)
def mycallback(w):
    for l in w:
        for key, value in l.items():
            for v in value:
                ddict[key].add(v)  

Offsets 是一个可迭代对象，我已经使用 1,000 - 50,000 个元素进行了测试。

如果我删除 r.wait() ，则无法返回 map_async 调用的所有输出。

使用r.wait()，我发现处理时间既低于串行实现，又无法扩展，即并行实现在时间上呈指数级增长，而串行版本则线性增加.

我知道 func 非常昂贵，因为它在串行和并行中固定我的处理核心。

我在哪里使用map_async引入了开销？它不在回调函数中，因为删除并替换为 result.append 不会影响时间。

编辑评论:

1. 我正在移动大型字典，范围从 1,000 到 100,000 个元素。该值通常是 3-5 个元素的集合。所以，酸洗肯定是一个问题。在不转移到共享内存中的情况下，人们会建议使用哪些替代数据结构？

2. apply_async 具有类似的回调，保存 for l in w 行，返回大致相同的结果。对于某些问题集，速度比 map_async 稍好，而对于其他问题集，速度稍差。使用托管字典和可连接队列的情况要糟糕得多。

3. 一些时间测试。这是使用 2 个核心。当我添加额外的核心时，我看到指数增长，因此我只能假设该增长是由进程生成或酸洗返回数据引起的。

func 获取一个数据点并寻找邻居。对于所有情况，它都是相同的函数，除了需要传递偏移量来告诉并行代码要搜索哪些数据点之外。这本质上是一个 KDTree 搜索功能。

均匀分布

1,000 个数据点:序列 0.098659992218 | apply_async 0.120759010315 | map_async 0.080078125

10,000 个数据点<====== 仅通过并行进行改进 |序列号 0.507845163345 | apply_async 0.446543931961 | map_async 0.477811098099

随机分布

10,000 个数据点:序列 0.584854841232 | apply_async 1.03224301338 | map_async 0.948460817337

50,000 个数据点:序列 3.66075992584 | apply_async 4.95467185974 | map_async 5.37306404114

Answer 1

您可以更改 func() 以返回集合字典而不是列表字典吗？然后你的回调函数可以这样重写:

def mycallback(w):
    for l in w:
        for key, value in l.items():
            ddict[key].update(value)

这应该有助于缩短串行和并行处理时间。

不幸的是，我认为@Dougal 在线程之间传递数据时对所有数据进行酸洗/取消酸洗是正确的。由于酸洗的开销，将二进制数据写入磁盘并再次读回可能比在内存中传递它更快。您可以使用如下格式:

key value1 value2 value3 ...
key2 valueA valueB valueC ...
...

应该易于编写和阅读。