多处理一个写入python数组的函数循环

Question

我正在尝试为此循环实现多处理。它无法修改数组或似乎没有正确排序作业（在最后一个函数完成之前返回数组）。

import multiprocessing
import numpy


def func(i, array):
    array[i] = i**2
    print(i**2)

def main(n):
    array = numpy.zeros(n)

    if __name__ == '__main__':
        jobs = []
        for i in range(0, n):
            p = multiprocessing.Process(target=func, args=(i, array))
            jobs.append(p)
            p.start()

    return array

print(main(10))

Answer 1

进程不共享内存，您的程序最初会创建一个全为零的数组，然后启动 10 个进程，这些进程将在第一次创建数组的副本时调用 func 函数，但不会调用原始数组。

看来您真正想要完成的是：

from multiprocessing import Process, Lock
from multiprocessing.sharedctypes import Array


def modify_array(index, sharedarray):
    sharedarray[index] = index ** 2
    print([x for x in sharedarray])


def main(n):
    lock = Lock()
    array = Array('i', 10, lock=lock)
    if __name__ == '__main__':
        for i in range(0, n):
            p = Process(target=modify_array, args=(i, array))
            p.start()
            p.join()
    return list(array)

main(10)

输出：

[0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
[0, 1, 0, 0, 0, 0, 0, 0, 0, 0]
[0, 1, 4, 0, 0, 0, 0, 0, 0, 0]
[0, 1, 4, 9, 0, 0, 0, 0, 0, 0]
[0, 1, 4, 9, 16, 0, 0, 0, 0, 0]
[0, 1, 4, 9, 16, 25, 0, 0, 0, 0]
[0, 1, 4, 9, 16, 25, 36, 0, 0, 0]
[0, 1, 4, 9, 16, 25, 36, 49, 0, 0]
[0, 1, 4, 9, 16, 25, 36, 49, 64, 0]
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

但问题是，使用多处理是错误的。与新线程相比，产生额外进程或什至只是保持单线程并利用事件循环来触发操作会产生很多开销。

在 Python 的单线程、单个进程中使用并发的示例可能如下所示：

import numpy as np
from asyncio import get_event_loop, wait, ensure_future


def modify_array(index, array):
    array[index] = index ** 2
    print([x for x in array])


async def task(loop, function, index, array):
    await loop.run_in_executor(None, function, index, array)


def main(n):
    loop = get_event_loop()
    jobs = list()
    array = np.zeros(10)
    for i in range(0, n):
        jobs.append(
            ensure_future(
                task(loop, modify_array, i, array)
            )
        )
    loop.run_until_complete(wait(jobs))
    loop.close()

main(10)

这是当今流行的模式，使用 asyncio 事件循环来并行完成任务。但是，由于您使用的是 Numpy 之类的库，我怀疑这种模式对您有多大价值。

Answer 2

我以前没有使用过multiprocessing，所以我也是新手，但是在做了一些研究（主要来自thesetwo 帖子）之后，我想我已经部​​分设法解决了你的问题使用此代码：

import multiprocessing
import numpy


def func(i, array, connection):
    squared_value = i ** 2
    array[i] = squared_value
    print(squared_value)

    connection.send(array)


def main(n):
    array = numpy.zeros(n)

    for i in range(0, n):
        recv_end, send_end = multiprocessing.Pipe(False)
        p = multiprocessing.Process(target=func, args=(i, array, send_end))
        p.start()
        p.join()
        array = recv_end.recv()

    return array


if __name__ == '__main__':
    print(main(10))

输出

0
1
4
9
16
25
36
49
64
81
[  0.   1.   4.   9.  16.  25.  36.  49.  64.  81.]

this answer that I referenced in the comments 中解释了这种方法修改数组而您的方法不修改数组的原因：

问题在于，当对象被传递给工作进程时，它们会被泡菜打包，然后运送到另一个进程，在那里它们被解包并处理。您的对象并没有像克隆的那样传递给其他进程。您不返回对象，因此克隆的对象被愉快地修改，然后被丢弃。

关于我的（部分）解决方案，我应该指出几点：

• 此实现的运行速度很多比仅以常规方式（通过单个线程）生成此列表要慢。这很可能是由于创建新进程和在它们之间编组数据的额外开销。

• 由于问题的性质（每个作业都修改数组），每个作业都必须将前一个作业的输出作为输入。由于这个限制，我认为不可能让作业同时运行，这有点违背了多处理的意义。

至于这两点中的后者，我确实尝试了一个变体，其中func 返回一个函数，它接受一个数组并返回它的修改版本。这将允许作业同时运行，但不幸的是，函数似乎不能被腌制。