为什么我的循环每次迭代都需要更多内存？

Question

我正在尝试减少我的 python 3 代码的内存需求。现在，for 循环的每次迭代都需要比上一次更多的内存。

我写了一小段与我的项目具有相同行为的代码：

import numpy as np
from multiprocessing import Pool
from itertools import repeat


def simulation(steps, y):  # the function that starts the parallel execution of f()
    pool = Pool(processes=8, maxtasksperchild=int(steps/8))
    results = pool.starmap(f, zip(range(steps), repeat(y)), chunksize=int(steps/8))
    pool.close()
    return results


def f(steps, y):  # steps is used as a counter. My code doesn't need it.
        a, b = np.random.random(2)
        return y*a, y*b

def main():
    steps = 2**20  # amount of times a random sample is taken
    y = np.ones(5)  # dummy variable to show that the next iteration of the code depends on the previous one
    total_results = np.zeros((0,2))
    for i in range(5):
        results = simulation(steps, y[i-1])
        y[i] = results[0][0]
        total_results = np.vstack((total_results, results))

    print(total_results, y)

if __name__ == "__main__":
    main()

对于 for 循环的每次迭代，simulation() 中的每个线程的内存使用量都等于我的代码使用的总内存量。

每次运行并行进程时，Python 是否会克隆我的整个环境，包括 f() 不需要的变量？如何防止这种行为？

理想情况下，我希望我的代码只复制执行 f() 所需的内存，而我可以将结果保存在内存中。

Answer 1

尽管脚本确实使用了相当多的内存，即使是“较小”的示例值，但答案

每次并行时，Python 是否会克隆我的整个环境？ 运行进程，包括 f() 不需要的变量？如何 我可以防止这种行为吗？

是它以某种方式克隆使用forking 一个新进程的环境，但是如果copy-on-write 语义可用，则在将其写入之前不需要复制实际的物理内存.例如在这个系统上

 % uname -a 
Linux mypc 4.2.0-27-generic #32-Ubuntu SMP Fri Jan 22 04:49:08 UTC 2016 x86_64 x86_64 x86_64 GNU/Linux

COW 似乎可用并正在使用中，但在其他系统上可能并非如此。在 Windows 上，这是完全不同的，因为新的 Python 解释器是从 .exe 执行的，而不是分叉。由于您提到使用 htop，因此您使用的是某种 UNIX 或类似 UNIX 的系统，并且您会得到 COW 语义。

对于 for 循环的每次迭代，simulation() 中的进程 每个 内存使用量等于我的代码使用的总内存。

生成的进程将显示几乎相同的RSS 值，但这可能会产生误导，因为如果不发生写入，它们大多占用映射到多个进程的相同实际物理内存。使用Pool.map 的情况有点复杂，因为它“将可迭代对象分割成多个块，作为单独的任务提交给进程池”。此提交发生在IPC 上，提交的数据将被复制。在您的示例中，IPC 和 2**20 函数调用也支配 CPU 使用率。用 simulation 中的单个向量化乘法替换映射使脚本在这台机器上的运行时间从大约 150 秒缩短到 0.66 秒。

我们可以观察COW 的一个（有点）简化的例子，它分配一个大数组并将它传递给一个衍生的进程进行只读处理：

import numpy as np
from multiprocessing import Process, Condition, Event
from time import sleep
import psutil


def read_arr(arr, done, stop):
    with done:
        S = np.sum(arr)
        print(S)
        done.notify()
    while not stop.is_set(): 
        sleep(1)


def main():
    # Create a large array
    print('Available before A (MiB):', psutil.virtual_memory().available / 1024 ** 2)
    input("Press Enter...")
    A = np.random.random(2**28)
    print('Available before Process (MiB):', psutil.virtual_memory().available / 1024 ** 2)
    input("Press Enter...")
    done = Condition()
    stop = Event()
    p = Process(target=read_arr, args=(A, done, stop))
    with done:
        p.start()
        done.wait()
    print('Available with Process (MiB):', psutil.virtual_memory().available / 1024 ** 2)
    input("Press Enter...")
    stop.set()
    p.join()

if __name__ == '__main__':
    main()

这台机器上的输出：

 % python3 test.py
Available before A (MiB): 7779.25
Press Enter...
Available before Process (MiB): 5726.125
Press Enter...
134221579.355
Available with Process (MiB): 5720.79296875
Press Enter...

现在，如果我们将函数 read_arr 替换为修改数组的函数：

def mutate_arr(arr, done, stop):
    with done:
        arr[::4096] = 1
        S = np.sum(arr)
        print(S)
        done.notify()
    while not stop.is_set(): 
        sleep(1)

结果完全不同：

Available before A (MiB): 7626.12109375
Press Enter...
Available before Process (MiB): 5571.82421875
Press Enter...
134247509.654
Available with Process (MiB): 3518.453125
Press Enter...

---

for 循环在每次迭代后确实需要更多内存，但这很明显：它从映射中堆叠 total_results，因此它必须为新数组分配空间来保存旧结果和新结果释放现在未使用的旧结果数组。

Answer 2

也许你应该知道Operating System 中thread 和process 之间的区别。看到这个What is the difference between a process and a thread。

在for循环中，有processes，而不是threads。线程共享创建它的进程的地址空间；进程有自己的地址空间。

可以打印进程id，输入os.getpid()。