问题

我的应用程序正在提取内存中的 zip 文件列表并将数据写入临时文件。然后我内存映射临时文件中的数据以用于另一个函数。当我在单个进程中执行此操作时，它工作正常，读取数据不会影响内存，最大 RAM 约为 40MB。但是，当我使用 concurrent.futures 执行此操作时，RAM 会上升到 500MB。

我看过this例如，我知道我可以以更好的方式提交作业以在处理过程中节省内存。但我不认为我的问题是相关的，因为我在处理过程中没有用完内存。我不明白的问题是为什么即使在返回内存映射后它仍然保留在内存中。我也不了解内存中的内容，因为在单个进程中执行此操作不会将数据加载到内存中。

谁能解释一下内存中实际有什么，以及为什么单一处理和并行处理之间存在差异？

PS 我使用memory_profiler 来测量内存使用情况

代码

主要代码:

def main(): 
    datadir = './testdata' 
    files = os.listdir('./testdata') 
    files = [os.path.join(datadir, f) for f in files] 
    datalist = download_files(files, multiprocess=False) 
    print(len(datalist)) 
    time.sleep(15) 
    del datalist # See here that memory is freed up 
    time.sleep(15) 

其他功能:

def download_files(filelist, multiprocess=False): 
    datalist = [] 
    if multiprocess: 
        with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor: 
            returned_future = [executor.submit(extract_file, f) for f in filelist] 
        for future in returned_future: 
            datalist.append(future.result()) 
    else: 
        for f in filelist: 
            datalist.append(extract_file(f)) 
    return datalist 
 
def extract_file(input_zip): 
    buffer = next(iter(extract_zip(input_zip).values())) 
    with tempfile.NamedTemporaryFile() as temp_logfile: 
        temp_logfile.write(buffer) 
        del buffer 
        data = memmap(temp_logfile, dtype='float32', shape=(2000000, 4), mode='r') 
    return data 
 
def extract_zip(input_zip): 
    with ZipFile(input_zip, 'r') as input_zip: 
        return {name: input_zip.read(name) for name in input_zip.namelist()} 

数据辅助代码

我不能分享我的实际数据，但这里有一些简单的代码来创建演示问题的文件:

for i in range(1, 16): 
    outdir = './testdata' 
    outfile = 'file_{}.dat'.format(i) 
    fp = np.memmap(os.path.join(outdir, outfile), dtype='float32', mode='w+', shape=(2000000, 4)) 
    fp[:] = np.random.rand(*fp.shape) 
    del fp 
    with ZipFile(outdir + '/' + outfile[:-4] + '.zip', mode='w', compression=ZIP_DEFLATED) as z: 
        z.write(outdir + '/' + outfile, outfile) 

请您参考如下方法:

问题是您要传递 np.memmap在进程之间，这是行不通的。

最简单的解决方案是传递文件名，并让子进程 memmap 使用相同的文件。

当你pass an argument to a child process or pool method via multiprocessing ，或从一个返回一个值(包括通过 ProcessPoolExecutor 间接这样做)，它通过调用 pickle.dumps 来工作在值上，跨进程传递 pickle(细节各不相同，但无论是 Pipe 还是 Queue 或其他什么都不重要)，然后 unpickling结果在另一边。

memmap 基本上只是一个 mmap在 mmapped 内存中分配了一个 ndarray 的对象。

而且 Python 不知道如何 pickle mmap 对象。 (如果您尝试，您将收到 PicklingError 或 BrokenProcessPool 错误，具体取决于您的 Python 版本。)

A np.memmap 可以被 pickle，因为它只是 np.ndarray 的子类——但是 pickling 和 unpickling 它实际上复制了数据并为您提供一个普通的内存数组。 (如果你查看 data._mmap，它是 None。)如果它给你一个错误而不是默默地复制你的所有数据(pickle-替换库 dill 正是这样做的:TypeError: can't pickle mmap.mmap objects)，但它没有。

在进程之间传递底层文件描述符并非不可能——每个平台的细节都不同，但所有主要平台都有办法做到这一点。然后您可以使用传递的 fd 在接收端构建一个 mmap，然后从中构建一个 memmap。您甚至可以将其包装在 np.memmap 的子类中。但我怀疑如果这不是有点困难，那么有人已经做到了，事实上它可能已经成为 dill 的一部分，如果不是 numpy 本身。

另一种选择是显式使用 shared memory features of multiprocessing ，并在共享内存中分配数组而不是 mmap。

但最简单的解决方案是，正如我在顶部所说的那样，只传递文件名而不是对象，并让每一方 memmap 相同的文件。不幸的是，这确实意味着你不能只使用关闭时删除 NamedTemporaryFile (尽管你使用它的方式已经不可移植并且不会以相同的方式在 Windows 上工作它在 Unix 上确实如此)，但改变它仍然可能比其他替代方案更有效。