python 多处理管理器复合图案共享

Question

我试图通过多处理管理器共享复合结构，但在尝试仅使用其中一种复合类方法时，我遇到了“运行时错误：超出最大递归深度”的麻烦。

该类是来自 code.activestate 的令牌，并在纳入管理器之前经过我的测试。

当将类检索到进程中并调用其 addChild() 方法时，我保留了 RunTimeError，而在进程之外它仍然有效。

复合类继承自 SpecialDict 类，该类实现 ** ____getattr()____ ** 方法。

有可能在调用 addChild() 时，python 解释器会寻找不同的 ** ____getattr()____ ** 因为正确的一个没有被管理器代理？

如果是这样，我不清楚为该类/方法创建代理的正确方法

下面的代码准确地重现了这种情况：

1）这是manager.py：2

from multiprocessing.managers import BaseManager
from CompositeDict import *

class PlantPurchaser():

    def __init__(self):
        self.comp  = CompositeDict('Comp')

    def get_cp(self):
        return self.comp

class Manager():

    def __init__(self):

        self.comp  = QueuePurchaser().get_cp()

        BaseManager.register('get_comp', callable=lambda:self.comp)

        self.m = BaseManager(address=('127.0.0.1', 50000), authkey='abracadabra')
        self.s = self.m.get_server()

        self.s.serve_forever()

）我想在这个consumer.py中使用复合：

from multiprocessing.managers import BaseManager

class Consumer():

    def __init__(self):

        BaseManager.register('get_comp')

        self.m = BaseManager(address=('127.0.0.1', 50000), authkey='abracadabra')
        self.m.connect()

        self.comp = self.m.get_comp()
        ret = self.comp.addChild('consumer')

3）通过controller.py运行所有启动：

from multiprocessing import Process

class Controller():
    def __init__(self):
        for child in _run_children():
            child.join()

def _run_children():

    from manager import Manager
    from consumer import Consumer as Consumer

procs = (
         Process(target=Manager,  name='Manager' ),
         Process(target=Consumer, name='Consumer'),
        )

for proc in procs:
    proc.daemon = 1
    proc.start()
return procs

c = Controller()

看看这个相关问题，了解如何为 CompositeDict() 类做代理 正如阿尔伯特建议的那样。

tgray给出的解决方案有效，但无法避免竞争条件

Answer 1

类之间是否有可能存在循环引用？例如，外部类具有对复合类的引用，而复合类又具有对外部类的引用。

多处理管理器运行良好，但是当您有大型、复杂的类结构时，您可能会遇到类型/引用无法正确序列化的错误。另一个问题是来自多处理管理器的错误非常神秘。这使得调试故障条件变得更加困难。

Answer 2

我认为问题在于你必须指示 Manager 如何管理你的对象，这不是标准的 python 类型。

在其他世界中您必须为您的 CompositeDict 创建一个代理

您可以查看此文档的示例：http://ruffus.googlecode.com/svn/trunk/doc/html/sharing_data_across_jobs_example.html

Answer 3

Python 默认的最大递归深度为 1000（或者 999，我忘了……）。但您可以这样更改默认行为：

import sys
sys.setrecursionlimit(n)

其中 n 是您希望允许的递归次数。

编辑：

以上答案没有解决此问题的根本原因（如评论中指出的）。仅当您有意递归超过 1000 次时才需要使用它。如果您处于无限循环中（就像在这个问题中），您最终将达到您设置的任何限制。

为了解决您的实际问题，我从头开始重新编写了您的代码，尽可能简单地将其构建为我认为您想要的：

import sys
from multiprocessing import Process
from multiprocessing.managers import BaseManager
from CompositDict import *

class Shared():
    def __init__(self):
        self.comp = CompositeDict('Comp')

    def get_comp(self):
        return self.comp

    def set_comp(self, c):
        self.comp = c

class Manager():
    def __init__(self):
        shared = Shared()
        BaseManager.register('get_shared', callable=lambda:shared)
        mgr = BaseManager(address=('127.0.0.1', 50000), authkey='abracadabra')
        srv = mgr.get_server()
        srv.serve_forever()

class Consumer():
    def __init__(self, child_name):
        BaseManager.register('get_shared')
        mgr = BaseManager(address=('127.0.0.1', 50000), authkey='abracadabra')
        mgr.connect()

        shared = mgr.get_shared()
        comp = shared.get_comp()
        child = comp.addChild(child_name)
        shared.set_comp(comp)
        print comp

class Controller():
    def __init__(self):
        pass

    def main(self):
        m = Process(target=Manager, name='Manager')
        m.daemon = True
        m.start()

        consumers = []
        for i in xrange(3):
            p = Process(target=Consumer, name='Consumer', args=('Consumer_' + str(i),))
            p.daemon = True
            consumers.append(p)

        for c in consumers:
            c.start()
        for c in consumers:
            c.join()
        return 0


if __name__ == '__main__':
    con = Controller()
    sys.exit(con.main())

我在一个文件中完成了这一切，但您不应该有任何麻烦把它拆散。

我向您的消费者添加了一个 child_name 参数，以便我可以检查 CompositDict 是否正在更新。

请注意，CompositDict 对象有一个 getter 和 setter。当我只有一个 getter 时，每个 Consumer 在添加子项时都会覆盖 CompositDict 。

这就是为什么我还将您的注册方法更改为 get_shared 而不是 get_comp，因为您将需要访问 Consumer 类中的 setter 和 getter。

另外，我认为您不想尝试加入您的经理流程，因为它将“永远服务”。如果您查看 BaseManager 的源代码（./Lib/multiprocessing/managers.py:Line 144），您会注意到 serve_forever() 函数会让您陷入一个无限循环，该循环只是被破坏了通过KeyboardInterrupt 或SystemExit。

最重要的是，这段代码可以在没有任何递归循环的情况下工作（据我所知），但如果您仍然遇到错误，请告诉我。