Python多进程池映射，向生成的进程传递参数

Question

def content_generator(applications, dict):
    for app in applications:
        yield(app, dict[app])

with open('abc.pickle', 'r') as f:
    very_large_dict = pickle.load(f)
all_applications = set(very_large_dict.keys())

pool = multiprocessing.Pool()
for result in pool.imap_unordered(func_process_application, content_generator(all_applications, very_large_dict)):
    do some aggregation on result

我有一个非常庞大的字典，其键是字符串（应用程序名称），值是与应用程序相关的信息。由于应用程序是独立的，我想使用多进程并行处理它们。当字典不是很大时，并行化运作良好，但当字典太大时，所有Python进程都被杀死了。我使用 dmesg 检查了出了什么问题，并发现它们被杀死是因为机器没有足够的内存。在池进程运行时，我使用 top 发现它们都占用相同数量的驻留内存（RES），即全部为 3.4G。这让我感到困惑，因为似乎已经将整个字典复制到生成的进程中。我以为我已经打破了字典，并通过仅返回 dict[app] 而不是 dict 来传递与生成的进程相关的内容。您对我的错误有何想法？

- hch

---

你应该考虑以下几点：为了进行最小化处理而传递大量数据并不是使用多进程的好方法。如果你有小数据，但需要长时间任务，则这是更好的方式。除非你有序列化在磁盘上的数据，可以仅传递路径并在进程中取消pickle，否则你的示例似乎不是一个好的用例。其次，对于大文件，许多处理会成为一个问题。使用信号量来避免通过内存，限制活动进程的数量。 - Alex Huszagh

接下来，您需要考虑管道限制：管道仅对一定大小保证原子性，一旦超过该限制，您基本上会保证数据损坏（例如在使用大型字典时）。 - Alex Huszagh

在Linux系统上，新进程是通过fork()创建的，因此它们在创建时会获得整个父进程地址空间的副本。这是“写时复制”，因此更像是“虚拟”副本而不是“真实”副本，但仍然...;-)首先，请尝试在创建大型数据结构之前创建您的Pool。然后子进程将继承一个更小的地址空间。 - Tim Peters

另外，最后，是的：除非您可以传递指向底层C数据类型的指针（例如numpy数组，没有GIL，然后不修改缓冲区的内容），否则多进程无法使用共享内存。这意味着在使用多进程处理字典时，复制操作是不可避免的。 - Alex Huszagh

@AlexanderHuszagh，他并没有传递它们。他传递的是content_generator(all_applications, very_large_dict)，这是一个生成器，每次产生一个(key, value)对。该生成器在主进程中运行。 - Tim Peters

Answer 1

评论已经变得难以跟踪，所以我在此粘贴我的重要评论：

在Linux系统上，通过fork()创建新进程，因此在它们创建时获取整个父进程地址空间的副本。它是“写时复制”，因此更像是“虚拟”副本而不是“实际”副本，但仍然...;-)首先，尝试在创建巨大数据结构之前创建您的Pool。然后子进程将继承一个更小的地址空间。

然后回答一些问题：

那么在Python 2.7中，没有办法生成新进程吗？

在Linux系统上，没有。在Python 3.4中首次添加了在这些系统上使用“spawn”的功能。在Windows系统上，“spawn”一直是唯一的选择（Windows上没有fork()）。

大字典作为参数传递给函数，我只能在此函数内部创建池。我怎样才能在大字典之前创建池？

很简单：将这两行代码放在程序的前两行：

import multiprocessing
pool = multiprocessing.Pool()

你可以随时创建池（只要它在使用之前存在），工作进程将继承在调用Pool构造函数时的整个地址空间。

另一个建议：如果创建字典后不会变异，请尝试改用以下方法：

def content_generator(dict):
    for app in dict:
        yield app, dict[app]

那样你就不必实例化一个巨大的键集合。或者更好的方法是（如果可能的话），直接迭代项目：

for result in pool.imap_unordered(func_process_application, very_large_dict.iteritems()):

- Tim Peters