利用“抄写副本”将数据复制到多处理。pool（）工作流程

Question

我有一些多处理 python代码，看起来有点像这样：

import time
from multiprocessing import Pool
import numpy as np

class MyClass(object):
    def __init__(self):
        self.myAttribute = np.zeros(100000000) # basically a big memory struct

    def my_multithreaded_analysis(self):
        arg_lists = [(self, i) for i in range(10)]
        pool = Pool(processes=10)
        result = pool.map(call_method, arg_lists)
        print result

    def analyze(self, i):
        time.sleep(10)
        return i ** 2

def call_method(args):
    my_instance, i = args
    return my_instance.analyze(i)


if __name__ == '__main__':
    my_instance = MyClass()
    my_instance.my_multithreaded_analysis()

在阅读了有关记忆如何在其他stackoverflow中工作的答案之后，例如此

我的问题基本上是，我是否正确解释了我的myClass的实例，实际上是在整个池中重复的？如果是这样，我该如何防止这种情况；我应该不使用这样的结构吗？我的目标是减少记忆使用量以进行计算分析。

PID   COMMAND      %CPU  TIME     #TH    #WQ  #PORT MEM    PURG   CMPRS  PGRP PPID STATE
2494  Python       0.0   00:01.75 1      0    7     765M   0B     0B     2484 2484 sleeping
2493  Python       0.0   00:01.85 1      0    7     765M   0B     0B     2484 2484 sleeping
2492  Python       0.0   00:01.86 1      0    7     765M   0B     0B     2484 2484 sleeping
2491  Python       0.0   00:01.83 1      0    7     765M   0B     0B     2484 2484 sleeping
2490  Python       0.0   00:01.87 1      0    7     765M   0B     0B     2484 2484 sleeping
2489  Python       0.0   00:01.79 1      0    7     167M   0B     597M   2484 2484 sleeping
2488  Python       0.0   00:01.77 1      0    7     10M    0B     755M   2484 2484 sleeping
2487  Python       0.0   00:01.75 1      0    7     8724K  0B     756M   2484 2484 sleeping
2486  Python       0.0   00:01.78 1      0    7     9968K  0B     755M   2484 2484 sleeping
2485  Python       0.0   00:01.74 1      0    7     171M   0B     594M   2484 2484 sleeping
2484  Python       0.1   00:16.43 4      0    18    775M   0B     12K    2484 2235 sleeping

Answer 1

另外，要利用分叉的复制品福利，在保留封装的外观的同时，您可以利用纯的班级属性和@classmethods Globals 。

import time
from multiprocessing import Pool
import numpy as np

class MyClass(object):

    myAttribute = np.zeros(100000000) # basically a big memory struct
    # myAttribute is a class-attribute

    @classmethod
    def my_multithreaded_analysis(cls):
        arg_list = [i for i in range(10)]
        pool = Pool(processes=10)
        result = pool.map(analyze, arg_list)
        print result

    @classmethod
    def analyze(cls, i):
        time.sleep(10)
        # If you wanted, you could access cls.myAttribute w/o worry here.
        return i ** 2

""" We don't need this proxy step !
    def call_method(args):
        my_instance, i = args
        return my_instance.analyze(i)
"""

if __name__ == '__main__':
    my_instance = MyClass()
    # Note that now you can instantiate MyClass anywhere in your app,
    # While still taking advantage of copy-on-write forking
    my_instance.my_multithreaded_analysis()

注1：是的，我承认class-attributes和类方法是荣耀的全球范围。但是它购买了一些封装...

注2：，而不是明确创建您的arg_lists，您可以隐式传递实例（self）到每个任务由pool创建，通过将界限方法传递给分析（self） pool.map（），并在脚上拍摄自己更容易！

Answer 2

发送到pool.map（和相关方法）的任何内容实际上都不使用共享的复制在写入资源。这些值为“腌制”（Python的序列化机制）工程处理并在那里进行了未划分的处理，从而从头开始重建孩子中的对象。因此，在这种情况下，每个孩子最终都获得了原始数据的复制版本（从未使用过，因为它被告知使用通过IPC发送的复制），以及对原始数据的个人娱乐在孩子中重建，没有共享。

如果您想利用分叉的写入福利，则不能在管道上发送数据（或引用数据引用数据的对象）。您必须将它们存储在可以通过访问自己的全球群体从孩子那里找到的位置。因此：例如：

import os
import time
from multiprocessing import Pool
import numpy as np

class MyClass(object):
    def __init__(self):
        self.myAttribute = os.urandom(1024*1024*1024) # basically a big memory struct(~1GB size)

    def my_multithreaded_analysis(self):
        arg_lists = list(range(10))  # Don't pass self
        pool = Pool(processes=10)
        result = pool.map(call_method, arg_lists)
        print result

    def analyze(self, i):
        time.sleep(10)
        return i ** 2

def call_method(i):
    # Implicitly use global copy of my_instance, not one passed as an argument
    return my_instance.analyze(i)

# Constructed globally and unconditionally, so the instance exists
# prior to forking in commonly accessible location
my_instance = MyClass()


if __name__ == '__main__':
    my_instance.my_multithreaded_analysis()

通过不传递self，您避免制作副本，而只需使用被映射到子女的单个全局对象即可。如果您需要多个对象，则可以在创建池之前制作一个全局list dict 映射到对象的实例，然后传递可以的索引或键查找对象作为参数的一部分，pool.map。然后，工人功能使用索引/键（必须腌制并通过IPC发送给孩子）以查找全局dict（也是抄写映射的复制映射）的值（也是复制映射），因此，您将廉价的信息复制到不复制孩子的孩子中的昂贵数据。

如果对象很小，即使您不写信给它们，它们也会被复制。 CPYTHON是参考计数的，并且参考计数出现在公共对象标头中，并通过参考对象不断更新，即使它是逻辑上非变形引用。因此，将写入并复制的小对象（以及在内存同一页面中分配的所有其他对象）。对于大对象（您的亿个元素numpy阵列），只要您不写它，大部分将保持共享，因为标头仅占据了许多页面之一

在Python版本3.8中更改了：在MacOS上，Spawn Start方法现在为默认值。参见 mulitProcessing doc 。 Spawn不利用折叠式复印件。