python concurrent.futures.ProcessPoolExecutor 在满内存时崩溃

Question

Python concurrent.futures.ProcessPoolExecutor 因 RAM 满而崩溃

程序说明

嗨，我有一个计算量大的函数，我想并行运行。该函数是一个接受输入的测试：

• 要测试的 DataFrame
• 运行计算所基于的参数。

返回值是一个简短的计算结果列表。

我想在具有不同参数和相同输入 DataFrame 的 for 循环中运行相同的函数，基本上运行蛮力来为我的问题找到最佳参数。

我写的代码

我目前正在从模块 concurrent.futures 中与 ProcessPoolExecutor 同时运行代码。

import concurrent.futures
from itertools import repeat
import pandas as pd

from my_tests import func


parameters = [
    (arg1, arg2, arg3),
    (arg1, arg2, arg3),
    ...
]
large_df = pd.read_csv(csv_path)

with concurrent.futures.ProcessPoolExecutor() as executor:
    for future in executor.map(func, repeat(large_df.copy()), parameters):
        test_result = future.result()
        ...

问题

我面临的问题是我需要运行大量迭代，但我的程序几乎立即崩溃。

为了不让它崩溃，我需要将它限制为最多 4 个 worker，这是我 CPU 资源的 1/4。

with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
    ...

我发现我的程序因满 RAM (16 GB) 而崩溃。我发现奇怪的是，当我在更多的 worker 上运行它时，它逐渐消耗越来越多的 RAM，它从未释放，直到它崩溃。

我没有传递 DataFrame 的副本，而是尝试传递文件路径，但除了减慢我的程序之外，它没有改变任何东西。

您是否知道为什么会出现该问题以及如何解决？

Answer 1

请参阅我对 map 实际返回内容的评论。

这个答案与你的parameters列表有多大有关，即在多处理池的任务队列中放置了多少任务：

每次提交新任务（parameters 的每个元素一个任务）时，您当前正在创建和传递数据帧的副本（使用 large_df.copy()）。您可以做的一件事是初始化池进程一次每个池进程都有一个副本，池进程提交和执行的每个任务都将使用该副本。假设数据框本身没有被my_tests.func修改。如果它被修改并且您需要为每个新任务复制原始large_df，函数worker（见下文）可以制作副本。在这种情况下，您需要同时存在 2 * N 个副本（而不仅仅是 N 个副本），其中 N 是池中的进程数。如果 parameters 的长度大于该长度，这将节省您的内存，因为在您的代码中，数据帧的副本将存在于任务队列或池进程的地址空间中。

如果您在使用 Linux 的平台下运行叉子创建新进程的方法，那么每个子进程都会自动继承一个副本作为全局变量：

import concurrent.futures
import pandas as pd

from my_tests import func


parameters = [
    (arg1, arg2, arg3),
    (arg1, arg2, arg3),
    ...
]

large_df = pd.read_csv(csv_path) # will be inherited

def worker(parameter):
    return func(large_df, parameter)
    """
    # or:
    return func(large_df.copy(), parameter)
    """

with concurrent.futures.ProcessPoolExecutor() as executor:
    for result in executor.map(worker, parameters):
        ...

my_tests.func 期望将数据帧作为其第一个参数，但通过上述更改，不再传递数据帧；数据帧现在作为全局变量访问。因此，在不修改 func 的情况下，我们需要一个适配器函数，worker，它将传递给 func 它所期望的。当然，如果你能修改func，那么你可以不用适配器。

如果您在使用 Windows 的平台上运行产卵方法来创建新进程，然后：

import concurrent.futures
import pandas as pd

from my_tests import func

def init_pool_processes(df):
    global large_df
    large_df = df


def worker(parameter):
    return func(large_df, parameter)
    """
    # or:
    return func(large_df.copy(), parameter)
    """

if __name__ == '__main__':
    
    parameters = [
        (arg1, arg2, arg3),
        (arg1, arg2, arg3),
        ...
    ]
    
    large_df = pd.read_csv(csv_path) # will be inherited
    
    with concurrent.futures.ProcessPoolExecutor(initializer=init_pool_processes, initargs=(large_df,)) as executor:
        for result in executor.map(worker, parameters):
            ...