0

我很纠结这个问题。

我有一大串列表，我想用并行代码访问这些列表以执行 CPU 密集型操作。为了做到这一点，我正在尝试使用multiprocessing.Pool，问题是我还需要在我的子进程中查看这个庞大的列表列表。

由于“列表列表”不是常规的（例如：），因此我[[1, 2], [1, 2, 3]]无法将它们存储为mp.Array，并且如前所述，我没有使用mp.Process，所以我没有想出mp.Manager在此任务中使用的方法。保留这个列表列表对我来说很重要，因为我正在应用一个基于索引进行查询的函数，使用from operator import itemgetter.

这是我想要实现的一个虚构示例：

import multiprocessing as mp
from operator import itemgetter
import numpy as np

def foo(indexes):
    # here I must guarantee read acess for big_list_of_lists on every child process somehow
    # as this code would work with only with one child process using global variables but would fail
    # with larger data.
    store_tuples = itemgetter(*indexes)(big_list_of_lists)
    return np.mean([item for sublista in store_tuples for item in sublista])

def main():
    # big_list_of_lists is the varible that I want to share across my child process
    big_list_of_lists = [[1, 3], [3, 1, 3], [1, 2], [2, 0]]

    ctx = mp.get_context('spawn')
    # big_list_of_lists elements are also passed as args
    pool = mp.Pool(ctx.Semaphore(mp.cpu_count()).get_value())
    res=list(pool.map(foo, big_list_of_lists))
    pool.close()
    pool.join()

    return res

if __name__ is '__main__':
    print(main())
# desired output is equivalente to:
# a = []
# for i in big_list_of_lists:
#     store_tuples = itemgetter(*i)(big_list_of_lists)
#     a.append(np.mean([item for sublista in store_tuples for item in sublista]))
# 'a' would be equal to [1.8, 1.5714285714285714, 2.0, 1.75]

其他细节：最好使用 python 3.6 实现解决方案，并且必须在 Windows 上工作

非常感谢你！

4

1 回答 1

0

mp.Manager使用, 和mp.Manager.listof mp.Manager.lists对我来说似乎工作正常。我相信这不会将列表复制到每个进程。

重要的一行是：

big_list_of_lists_proxy = manager.list([manager.list(sublist) for sublist in big_list_of_lists])

您可能想要使用，具体取决于您的用例：

big_list_of_lists_proxy = manager.list(big_list_of_lists)

每个子列表是否应该是代理取决于每个子列表是否很大，以及是否完整读取。如果子列表很大，那么将列表对象传输到需要它的每个进程（O(n)复杂性）会很昂贵，因此应该使用来自管理器的代理列表，但是如果无论如何都需要每个元素，则没有使用代理的优势。

import multiprocessing as mp
from operator import itemgetter
import numpy as np
from functools import partial


def foo(indexes, big_list_of_lists):
    # here I must guarantee read acess for big_list_of_lists on every child process somehow
    # as this code would work with only with one child process using global variables but would fail
    # with larger data.
    store_tuples = itemgetter(*indexes)(big_list_of_lists)
    return np.mean([item for sublista in store_tuples for item in sublista])


def main():
    # big_list_of_lists is the varible that I want to share across my child process
    big_list_of_lists = [[1, 3], [3, 1, 3], [1, 2], [2, 0]]
    ctx = mp.get_context('spawn')
    with ctx.Manager() as manager:
        big_list_of_lists_proxy = manager.list([manager.list(sublist) for sublist in big_list_of_lists])
        # big_list_of_lists elements are also passed as args
        pool = ctx.Pool(ctx.Semaphore(mp.cpu_count()).get_value())
        res = list(pool.map(partial(foo, big_list_of_lists=big_list_of_lists_proxy), big_list_of_lists))
        pool.close()
        pool.join()

    return res


if __name__ == '__main__':
    print(main())
# desired output is equivalente to:
# a = []
# for i in big_list_of_lists:
#     store_tuples = itemgetter(*i)(big_list_of_lists)
#     a.append(np.mean([item for sublista in store_tuples for item in sublista]))
# 'a' would be equal to [1.8, 1.5714285714285714, 2.0, 1.75]
于 2021-12-22T23:54:50.687 回答