为什么返回np.memmap时并发.futures保留在内存中？

Question

问题

我的应用程序正在提取内存中的zip文件列表，并将数据写入临时文件。然后，我将数据映射到临时文件中，以供其他功能使用。当我在单个过程中执行此操作时，它可以正常工作，读取数据不会影响内存，最大RAM约为40MB。但是，当我使用current.futures执行此操作时，RAM高达500MB。

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

任何人都可以解释一下内存中的实际内容，以及为什么单处理和并行处理之间的区别吗？

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

代码

主要代码：

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)

Answer 1

问题在于您正在尝试np.memmap在流程之间传递a ，而这是行不通的。最简单的解决方案是改为传递文件名，并让子进程处理memmap相同的文件。当您通过通过参数将参数传递给子进程或池方法multiprocessing，或者从一个参数返回值（包括通过间接返回ProcessPoolExecutor）时，它通过调用pickle.dumps该值，在各个进程之间传递泡菜来工作（细节有所不同，但是不会）不管是aPipe还是aQueue或其他），然后从另一侧解开结果。Amemmap基本上只是在ped内存中分配了一个mmap对象。ndarraymmap而且Python不知道如何腌制mmap对象。（如果尝试这样做，则将得到aPicklingError或BrokenProcessPool错误，具体取决于您的Python版本。）np.memmap 可以腌制A ，因为它只是的子类，np.ndarray但是腌制和去腌制实际上会复制数据并为您提供一个简单的内存数组。（如果您看的话data._mmap，它是None。）如果它给您一个错误而不是静默地复制所有数据（pickle-replacement库dill完全是这样：），则可能会更好TypeError: can't pickle mmap.mmap objects，但事实并非如此。在进程之间传递底层文件描述符不是不可能的-每个平台上的细节都不同，但是所有主要平台都可以做到这一点。然后，您可以使用传递的fdmmap在接收端构建一个，然后再构建一个memmap。您甚至可以将其包装到的子类中np.memmap。但是我怀疑这是否有点困难，有人会做到这一点，实际上dill，如果不是numpy它本身，那可能已经成为它的一部分。另一种选择是显式使用的共享内存功能multiprocessing，并在共享内存而不是中分配数组mmap。但是，最简单的解决方案是，就像我在顶部所说的那样，只是传递文件名而不是对象，并让每一侧memmap使用相同的文件。不幸的是，这确实意味着您不能只使用关闭时删除功能NamedTemporaryFile（尽管您使用它的方式已经不可移植，并且在Windows上无法像在Unix上那样工作），但是改变了与其他替代方案相比，这项工作的工作量可能仍然更少。