2

问题

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

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

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

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)
4

1 回答 1

1

问题是你试图np.memmap在进程之间传递一个，这不起作用。

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

当您通过 将参数传递给子进程或池方法multiprocessing，或从其中返回一个值（包括通过 a 间接这样做ProcessPoolExecutor）时，它通过调用pickle.dumps该值，跨进程传递泡菜来工作（细节有所不同，但它不'不管它是 aPipe还是 aQueue或其他东西），然后在另一边解开结果。

Amemmap基本上只是一个在ped 内存中分配的mmap对象。ndarraymmap

而且 Python 不知道如何腌制一个mmap对象。（如果你尝试，你会得到一个PicklingError或一个BrokenProcessPool错误，这取决于你的 Python 版本。）

Anp.memmap 可以被腌制，因为它只是一个子类np.ndarray——但是腌制和取消腌制它实际上会复制数据并为您提供一个普通的内存数组。（如果您查看data._mmap，它是。）如果它给您一个错误而不是默默地复制您的所有数据（pickle-replacement 库正是这样做None的：），它可能会更好，但事实并非如此。dillTypeError: can't pickle mmap.mmap objects

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

另一种选择是显式使用 的共享内存功能multiprocessing，并在共享内存中分配数组而不是mmap.

但最简单的解决方案是，正如我在顶部所说，只传递文件名而不是对象，并让每一侧memmap都使用相同的文件。不幸的是，这确实意味着您不能只使用 delete-on-close NamedTemporaryFile（尽管您使用它的方式已经是不可移植的，并且不会像在 Unix 上那样在 Windows 上工作），而是改变这可能仍然比其他替代方案少。

于 2018-08-30T17:38:59.240 回答