python - 共享列表的多处理

Question

我写了这样一个程序:

from multiprocessing import Process, Manager

def worker(i):
    x[i].append(i)

if __name__ == '__main__':
    manager = Manager()
    x = manager.list()
    for i in range(5):
        x.append([])
    p = []
    for i in range(5):
        p.append(Process(target=worker, args=(i,)))
        p[i].start()

    for i in range(5):
        p[i].join()

    print x

我想在进程之间创建一个共享列表列表，每个进程修改其中的一个列表。但是这个程序的结果是一个空列表的列表:[[],[],[],[],[]]。

出了什么问题？

Answer 1

我认为这是因为管理器的实现方式有些怪癖。

如果您创建两个 Manager.list 对象，然后将其中一个列表附加到另一个列表，则您附加的列表类型会在父列表中发生变化:

>>> type(l)
<class 'multiprocessing.managers.ListProxy'>
>>> type(z)
<class 'multiprocessing.managers.ListProxy'>
>>> l.append(z)
>>> type(l[0])
<class 'list'>   # Not a ListProxy anymore

l[0] 和 z 不是同一个对象，因此不会像您期望的那样表现:

>>> l[0].append("hi")
>>> print(z)
[]
>>> z.append("hi again")
>>> print(l[0])
['hi again']

如您所见，更改嵌套列表对 ListProxy 对象没有任何影响，但更改 ListProxy 对象会更改嵌套列表。文档实际上 explicitly notes this :

Note

Modifications to mutable values or items in dict and list proxies will not be propagated through the manager, because the proxy has no way of knowing when its values or items are modified. To modify such an item, you can re-assign the modified object to the container proxy:

翻开源码可以看到，当你在一个ListProxy上调用append时，append调用实际上是通过IPC发送给一个manager对象，然后manager在shared上调用append列表。这意味着 append 的参数需要被 pickled/unpickled。在 unpickling 过程中，ListProxy 对象变成了一个常规的 Python 列表，它是 ListProxy 指向的对象(又名其所指对象)的副本。这也是noted in the documentation :

An important feature of proxy objects is that they are picklable so they can be passed between processes. Note, however, that if a proxy is sent to the corresponding manager’s process then unpickling it will produce the referent itself. This means, for example, that one shared object can contain a second

所以，回到上面的例子，如果 l[0] 是 z 的副本，为什么更新 z 也会更新 l[0] ？因为副本也注册到 Proxy 对象，所以，当您更改 ListProxy(上例中的 z)时，它也会更新列表的所有已注册副本(l[ 0] 在上面的例子中)。但是，副本对代理一无所知，因此当您更改副本时，代理不会更改。

因此，为了使您的示例正常工作，您需要在每次要修改子列表时创建一个新的 manager.list() 对象，并且只直接更新该代理对象，而不是而不是通过父列表的索引更新它:

#!/usr/bin/python

from multiprocessing import Process, Manager

def worker(x, i, *args):
    sub_l = manager.list(x[i])
    sub_l.append(i)
    x[i] = sub_l


if __name__ == '__main__':
    manager = Manager()
    x = manager.list([[]]*5)
    print x
    p = []
    for i in range(5):
        p.append(Process(target=worker, args=(x, i)))
        p[i].start()

    for i in range(5):
        p[i].join()

    print x

这是输出:

dan@dantop2:~$ ./multi_weirdness.py 
[[0], [1], [2], [3], [4]]