multiprocessing.Manager()如何在python中工作？

Question

我关心python中的multiprocessing.Manager(),这是一个例子,

import multiprocessing 

def f(ns):

    ns.x *=10
    ns.y *= 10

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    ns = manager.Namespace()
    ns.x = 1
    ns.y = 2

    print 'before', ns
    p = multiprocessing.Process(target=f, args=(ns,))
    p.start()
    p.join()
    print 'after', ns

而输出是,

before Namespace(x=1, y=2)
after Namespace(x=10, y=20)

直到现在,它按照我的预期工作,然后我修改了这样的代码,

import multiprocessing 

def f(ns):

    ns.x.append(10)
    ns.y.append(10)

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    ns = manager.Namespace()
    ns.x = []
    ns.y = []

    print 'before', ns
    p = multiprocessing.Process(target=f, args=(ns,))
    p.start()
    p.join()
    print 'after', ns

现在,输出是,

before Namespace(x=[], y=[])
after Namespace(x=[], y=[])

它让我感到困惑,为什么列表没有像我预期的那样改变？任何人都可以帮我弄清楚发生了什么？提前致谢!

Answer 1

Manager代理对象无法传播对容器内(非托管)可变对象所做的更改.换句话说,如果您有一个manager.list()对象,则对托管列表本身的任何更改都会传播到所有其他进程.但是如果在该列表中有一个普通的Python列表,则不会传播对内部列表的任何更改,因为管理器无法检测到更改.

为了传播的变化,你必须使用manager.list()的嵌套列表对象太(需要的Python 3.6或更新版本),或需要修改的manager.list()直接对象(见注释上manager.list的Python 3.5或以上).

例如,请考虑以下代码及其输出:

import multiprocessing
import time

def f(ns, ls, di):
    ns.x += 1
    ns.y[0] += 1
    ns_z = ns.z
    ns_z[0] += 1
    ns.z = ns_z

    ls[0] += 1
    ls[1][0] += 1 # unmanaged, not assigned back
    ls_2 = ls[2]  # unmanaged...
    ls_2[0] += 1
    ls[2] = ls_2  # ... but assigned back
    ls[3][0] += 1 # managed, direct manipulation

    di[0] += 1
    di[1][0] += 1 # unmanaged, not assigned back
    di_2 = di[2]  # unmanaged...
    di_2[0] += 1
    di[2] = di_2  # ... but assigned back
    di[3][0] += 1 # managed, direct manipulation

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    ns = manager.Namespace()
    ns.x = 1
    ns.y = [1]
    ns.z = [1]
    ls = manager.list([1, [1], [1], manager.list([1])])
    di = manager.dict({0: 1, 1: [1], 2: [1], 3: manager.list([1])})

    print('before', ns, ls, ls[2], di, di[2], sep='\n')
    p = multiprocessing.Process(target=f, args=(ns, ls, di))
    p.start()
    p.join()
    print('after', ns, ls, ls[2], di, di[2], sep='\n')

输出:

before
Namespace(x=1, y=[1], z=[1])
[1, [1], [1], <ListProxy object, typeid 'list' at 0x10b8c4630>]
[1]
{0: 1, 1: [1], 2: [1], 3: <ListProxy object, typeid 'list' at 0x10b8c4978>}
[1]
after
Namespace(x=2, y=[1], z=[2])
[2, [1], [2], <ListProxy object, typeid 'list' at 0x10b8c4630>]
[2]
{0: 2, 1: [1], 2: [2], 3: <ListProxy object, typeid 'list' at 0x10b8c4978>}
[2]

如您所见,当一个新值直接分配给托管容器时,它会发生变化; 当它被分配给托管容器中的可变容器时,它不会改变; 但是如果随后将可变容器重新分配给托管容器,则它会再次更改.使用嵌套的托管容器也可以工作,直接检测更改,而无需分配回父容器.

Answer 2

ns是NamespaceProxy实例.这些对象具有特殊的__getattr__,__setattr__和__delattr__允许跨进程共享值的方法.为了在更改值时利用此机制,必须触发__setattr__.

ns.x.append(10)

导致ns.__getattr__被调用以检索ns.x,但它不会ns.__setattr__被调用.

要解决这个问题,你必须使用ns.x = ....

def f(ns):   
    tmp = ns.x     # retrieve the shared value
    tmp.append(10)
    ns.x = tmp     # set the shared value