如何使用共享内存而不是在多个进程之间通过pickle传递对象

Question

我正在研究一个CPU密集型的ML问题，该问题的中心是一个加法模型。由于加法是主要的操作，所以我可以将输入数据分成几部分并生成多个模型，然后用覆盖的__add__方法合并这些模型。在

与多处理相关的代码如下所示：

def pool_worker(filename, doshuffle):
    print(f"Processing file: {filename}")
    with open(filename, 'r') as f:
        partial = FragmentModel(order=args.order, indata=f, shuffle=doshuffle)
        return partial

def generateModel(is_mock=False, save=True):
    model = None
    with ThreadPool(args.nthreads) as pool:
        from functools import partial
        partial_models = pool.imap_unordered(partial(pool_worker, doshuffle=is_mock), args.input)
        i = 0
        for m in partial_models:
            logger.info(f'Starting to merge model {i}')
            if model is None:
                import copy
                model = copy.deepcopy(m)
            else:
                model += m
            logger.info(f'Done merging...')
            i += 1

    return model

问题是，随着模型阶数的增加，内存消耗呈指数级增长，因此在4阶时，模型的每个实例大约为4-5gb，这会导致线程池崩溃，因为中间模型对象随后无法pickle。在

我读了一点这方面的文章，似乎即使酸洗不是问题，像这样传递数据仍然非常低效，正如对this answer所评论的那样。在

然而，对于如何使用共享内存来达到这个目的，几乎没有什么指导。有没有可能避免这个问题而不必改变模型对象的内部结构？在

Answer 1

查看ray项目，它是一个分布式执行框架，它利用apache arrow进行序列化。如果您正在使用numpy数组，那么它将是一个非常好的ML工作流工具。在

以下是object serialization上的文档片段

In Ray, we optimize for numpy arrays by using the Apache Arrow data format. When we deserialize a list of numpy arrays from the object store, we still create a Python list of numpy array objects. However, rather than copy each numpy array, each numpy array object holds a pointer to the relevant array held in shared memory. There are some advantages to this form of serialization.

• Deserialization can be very fast.
• Memory is shared between processes so worker processes can all read the same data without having to copy it.

在我看来，对于并行执行，它甚至比多处理库更容易使用，尤其是在希望使用共享内存时，这是tutorial中的用法介绍。在