多处理——线程池内存泄漏？

Question

我正在观察无法向自己解释的内存使用情况。下面我提供了我的实际代码的精简版本，它仍然表现出这种行为。该代码旨在完成以下操作：

以 1000 行的块读取文本文件。每一行都是一个句子。将这 1000 个句子拆分为 4 个生成器。将这些生成器传递到线程池并在 250 个句子上并行运行特征提取。在我的实际代码中，我从整个文件的所有句子中积累了特征和标签。现在出现了奇怪的事情：即使没有累积这些值，内存也已分配但不会再次释放！而且我认为它与线程池有关。总共占用的内存量取决于为任何给定单词提取的特征数量。我在这里用range(100). 看一看：

from sys import argv
from itertools import chain, islice
from multiprocessing import Pool
from math import ceil


# dummyfied feature extraction function
# the lengt of the range determines howmuch mamory is used up in total,
# eventhough the objects are never stored
def features_from_sentence(sentence):
    return [{'some feature'  'some value'} for i in range(100)], ['some label' for i in range(100)]


# split iterable into generator of generators of length `size`
def chunks(iterable, size=10):
    iterator = iter(iterable)
    for first in iterator:
        yield chain([first], islice(iterator, size - 1))


def features_from_sentence_meta(l):
    return list(map (features_from_sentence, l))


def make_X_and_Y_sets(sentences, i):
    print(f'start: {i}')
    pool = Pool()
    # split sentences into a generator of 4 generators
    sentence_chunks = chunks(sentences, ceil(50000/4))
    # results is a list containing the lists of pairs of X and Y of all chunks
    results = map(lambda x : x[0], pool.map(features_from_sentence_meta, sentence_chunks))
    X, Y = zip(*results)
    print(f'end: {i}')
    return X, Y


# reads file in chunks of `lines_per_chunk` lines
def line_chunks(textfile, lines_per_chunk=1000):
    chunk = []
    i = 0
    with open(textfile, 'r') as textfile:
        for line in textfile:
            if not line.split(): continue
            i+=1
            chunk.append(line.strip())
            if i == lines_per_chunk:
                yield chunk
                i = 0
                chunk = []
        yield chunk

textfile = argv[1]

for i, line_chunk in enumerate(line_chunks(textfile)):
    # stop processing file after 10 chunks to demonstrate
    # that memory stays occupied (check your system monitor)
    if i == 10:
        while True:
            pass
    X_chunk, Y_chunk = make_X_and_Y_sets(line_chunk, i)

我用来调试的文件有 50000 行非空行，这就是我在一个地方使用硬编码 50000 的原因。如果你想使用同一个文件，为了你方便，他是一个链接：

https://www.dropbox.com/s/v7nxb7vrrjim349/de_wiki_50000_lines?dl=0

现在，当您运行此脚本并打开系统监视器时，您将观察到内存已用完，并且使用量一直持续到第 10 个块，在那里我人为地进入一个无限循环以证明内存仍在使用中，即使我从未存储任何东西。

你能向我解释为什么会发生这种情况吗？我似乎缺少关于应该如何使用多处理池的一些信息。

Answer 1

首先，让我们澄清一些误解——尽管事实证明，这实际上并不是探索的正确途径。

当您在 Python 中分配内存时，当然必须从操作系统获取该内存。

但是，当您释放内存时，它很少会返回到操作系统，直到您最终退出。相反，它进入一个“空闲列表”——或者，实际上，用于不同目的的多个级别的空闲列表。这意味着下次您需要内存时，Python 已经拥有它，并且可以立即找到它，而无需与操作系统交谈以分配更多内存。这通常会使内存密集型程序更快。

但这也意味着——尤其是在现代 64 位操作系统上——试图通过查看你的活动监视器/任务管理器等来了解你是否真的有任何内存压力问题。旁边是没用的。

tracemalloc标准库中的模块提供了低级工具来查看内存使用情况。在更高级别上，您可以使用类似 的东西memory_profiler，它（如果您启用tracemalloc支持——这很重要）可以将该信息与来自诸如psutil了解事情进展的来源的操作系统级信息放在一起。

然而，如果你没有看到任何实际问题——你的系统不会进入交换地狱，你不会得到任何MemoryError异常，你的性能不会达到某种奇怪的悬崖，它线性扩展到 N 然后突然下降在 N+1 等情况下全都见鬼去吧——你通常一开始就不需要为这些而烦恼。

如果您确实发现了问题，那么幸运的是，您已经解决了一半。正如我在顶部提到的，您分配的大部分内存在您最终退出之前都不会返回给操作系统。但是如果你所有的内存使用都发生在子进程中，而这些子进程没有状态，你可以让它们随时退出并重新启动。

当然，这样做会带来性能成本——进程拆卸和启动时间，页面映射和缓存必须重新开始，并要求操作系统再次分配内存，等等。还有一个复杂的成本——你不能只是运行一个池，让它做它的事情；你必须参与它的事情，让它为你回收流程。

multiprocessing.Pool类中没有内置支持来执行此操作。

当然，您可以构建自己的Pool. 如果你想变得花哨，你可以查看源代码multiprocessing并执行它的操作。或者您可以从Process对象列表和一对Queues 中构建一个简单的池。或者您可以直接使用Process对象而不需要池的抽象。

您可能会遇到内存问题的另一个原因是您的各个进程都很好，但是您的进程太多了。

而且，事实上，这里似乎就是这种情况。

您Pool在此函数中创建了4 个工人：

def make_X_and_Y_sets(sentences, i):
    print(f'start: {i}')
    pool = Pool()
    # ...

......你为每个块调用这个函数：

for i, line_chunk in enumerate(line_chunks(textfile)):
    # ...
    X_chunk, Y_chunk = make_X_and_Y_sets(line_chunk, i)

因此，您最终会为每个块创建 4 个新进程。即使每个内存使用量都非常低，同时拥有数百个内存也会增加。

更不用说您可能会因数百个进程在 4 个内核上竞争而严重损害您的时间性能，因此您将时间浪费在上下文切换和操作系统调度上，而不是进行实际工作。

正如您在评论中指出的那样，解决这个问题很简单：只需pool为每个调用创建一个全局而不是一个新的全局。

很抱歉把所有 Columbo 都放在这里，但是……还有一件事……这段代码运行在模块的顶层：

for i, line_chunk in enumerate(line_chunks(textfile)):
    # ...
    X_chunk, Y_chunk = make_X_and_Y_sets(line_chunk, i)

……这就是试图启动池和所有子任务的代码。但是该池中的每个子进程都需要import这个模块，这意味着它们最终都将运行相同的代码，并启动另一个池和一组额外的子任务。

你大概是在 Linux 或 macOS 上运行它，默认startmethod是fork，这意味着multiprocessing可以避免这种情况import，所以你没有问题。但是对于其他 startmethods，这段代码基本上是一个会吃掉所有系统资源的 forkbomb。这包括spawn，这是 Windows 上的默认启动方法。因此，如果有任何人可能在 Windows 上运行此代码，您应该将所有顶级代码置于if __name__ == '__main__':保护中。