池处理使用异步内存泄漏

Question

在我的代码方面需要一些帮助，我正在尝试使用 pool 在我的列表中运行一些函数。

它是一个包含大量数据的文件。比如 71GB 的文本。

我正在尝试尽可能快地越线。 但是由于某种原因在运行过程中，我被内存耗尽了

我认为这是因为我的 pool.close 和 pool.join 在我的代码末尾。

但我不确定如何修复它以避免内存泄漏。我认为我的进程永远不会死并保存数据。

注意：api.delete_object(item) 返回大量数据。也许它会以某种方式卡在内存中？

这是我的代码：

import pandas as pd
import boto3
from multiprocessing.pool import ThreadPool as Pool

pool_size = 8  # my "parallelness"


def worker(item):
    try:
        result = api.delete_object(item)
    except:
        print('error with item')

pool = Pool(pool_size)        
i=0
for chunk in pd.read_csv("/folder/files_to_delete",chunksize=1000,header=None):
            i+=1000
            for ind in chunk.index:
              if "something.txt" not in chunk[0][ind]:
                    pool.apply_async(worker, (chunk[0][ind],))
                    print("{}".format(i), end='\r')
pool.close()
pool.join()

Answer 1

虽然您已使用multiprocessing 标记您的问题，但您使用的是多线程。但在我看来，使用多处理可能会更好，因为我不清楚使用线程执行 api.delete_object(item) 可以实现多少“并行性”。

无论如何，您可以尝试重新排列代码以使用函数imap_unordered，然后迭代它返回的iterable。虽然您的工作函数worker 的返回值不是特别有趣（发布的工作函数只返回None，但也许这是一种简化），检索这些值应该释放用于保存它们的内存.

import pandas as pd
import boto3
from multiprocessing.pool import ThreadPool as Pool

def generate_arguments():
    i = 0
    with pd.read_csv("/folder/files_to_delete",chunksize=1000,header=None) as rdr:
        for chunk in rdr:
            i += 1000
            for ind in chunk.index:
                if "something.txt" not in chunk[0][ind]:
                    yield chunk[0][ind]
                    print("{}".format(i), end='\r')
        

def worker(item):
    try:
        result = api.delete_object(item)
    except:
        print('error with item')

def main():
    pool_size = 8  # my "parallelness"
    pool = Pool(pool_size)
    results = pool.imap_unordered(worker, generate_arguments())
    # You can iterate results to get return values, which are None, from worker function:
    for result in results:
        # result is return value from worker
        pass
    pool.close()
    pool.join()

if __name__ == '__main__':
    main()

如果您确实切换到多处理，您应该在调用imap_unordered 时指定一个chunksize 参数。我建议您估计将提交的任务数除以 4 * 池大小并将结果用作该值。例如，如果您估计要提交 200,000 个任务并且池大小为 8，则使用 chunksize 值 200_000 / (4 * 8) = 6250。

请注意，imap_unordered 可以按任意顺序返回结果，但比imap 更有效。如果您实际上是从worker 返回的不是None 并且需要按任务提交顺序排列结果，则使用方法imap 或让imap_unordered 额外返回其传递的参数。