python - 上下文管理器和多处理池

假设您正在使用multiprocessing.Pool对象，并且正在使用构造函数的initializer设置传递初始值设定项函数，然后在全局命名空间中创建资源。假设资源具有上下文管理器。如果上下文管理的资源必须在流程的整个生命周期中都存在，但是在最后要进行适当的清理，您将如何处理它的生命周期？

到目前为止，我有点像这样:

resource_cm = None
resource = None


def _worker_init(args):
    global resource
    resource_cm = open_resource(args)
    resource = resource_cm.__enter__()

从这里开始，池进程可以使用资源。到目前为止，一切都很好。但是，由于multiprocessing.Pool类未提供destructor或deinitializer参数，因此处理清理工作有些棘手。

我的想法之一是使用atexit模块，并在初始化程序中注册清除程序。像这样:

def _worker_init(args):
    global resource
    resource_cm = open_resource(args)
    resource = resource_cm.__enter__()

    def _clean_up():
        resource_cm.__exit__()

    import atexit
    atexit.register(_clean_up)

这是一个好方法吗？有更简单的方法吗？

编辑:atexit似乎不起作用。至少在上面我没有用它的方式，所以到目前为止，我仍然没有解决这个问题的方法。

最佳答案

首先，这是一个非常好的问题!在研究了multiprocessing代码后，我想我找到了一种方法:

当您启动multiprocessing.Pool时，在内部Pool对象为池的每个成员创建一个multiprocessing.Process对象。这些子流程启动时，它们会调用_bootstrap函数，如下所示:

def _bootstrap(self):
    from . import util
    global _current_process
    try:
        # ... (stuff we don't care about)
        util._finalizer_registry.clear()
        util._run_after_forkers()
        util.info('child process calling self.run()')
        try:
            self.run()
            exitcode = 0
        finally:
            util._exit_function()
        # ... (more stuff we don't care about)

run方法是实际运行您为target对象提供的Process的方法。对于Pool进程，它是一个具有长时间运行的while循环的方法，该循环等待工作项通过内部队列进入。对我们来说真正有趣的是self.run之后的情况:util._exit_function()被调用。

事实证明，该函数进行了一些清理，听起来很像您要寻找的内容:

def _exit_function(info=info, debug=debug, _run_finalizers=_run_finalizers,
                   active_children=active_children,
                   current_process=current_process):
    # NB: we hold on to references to functions in the arglist due to the
    # situation described below, where this function is called after this
    # module's globals are destroyed.

    global _exiting

    info('process shutting down')
    debug('running all "atexit" finalizers with priority >= 0')  # Very interesting!
    _run_finalizers(0)

这是_run_finalizers的文档字符串:

def _run_finalizers(minpriority=None):
    '''
    Run all finalizers whose exit priority is not None and at least minpriority

    Finalizers with highest priority are called first; finalizers with
    the same priority will be called in reverse order of creation.
    '''

该方法实际上遍历终结器回调列表并执行它们:

items = [x for x in _finalizer_registry.items() if f(x)]
items.sort(reverse=True)

for key, finalizer in items:
    sub_debug('calling %s', finalizer)
    try:
        finalizer()
    except Exception:
        import traceback
        traceback.print_exc()

完美的。那么我们如何进入_finalizer_registry呢？ Finalize中有一个未记录的对象multiprocessing.util，它负责向注册表添加回调:

class Finalize(object):
    '''
    Class which supports object finalization using weakrefs
    '''
    def __init__(self, obj, callback, args=(), kwargs=None, exitpriority=None):
        assert exitpriority is None or type(exitpriority) is int

        if obj is not None:
            self._weakref = weakref.ref(obj, self)
        else:
            assert exitpriority is not None

        self._callback = callback
        self._args = args
        self._kwargs = kwargs or {}
        self._key = (exitpriority, _finalizer_counter.next())
        self._pid = os.getpid()

        _finalizer_registry[self._key] = self  # That's what we're looking for!

好的，因此将它们放到一个示例中:

import multiprocessing
from multiprocessing.util import Finalize

resource_cm = None
resource = None

class Resource(object):
    def __init__(self, args):
        self.args = args

    def __enter__(self):
        print("in __enter__ of %s" % multiprocessing.current_process())
        return self

    def __exit__(self, *args, **kwargs):
        print("in __exit__ of %s" % multiprocessing.current_process())

def open_resource(args):
    return Resource(args)

def _worker_init(args):
    global resource
    print("calling init")
    resource_cm = open_resource(args)
    resource = resource_cm.__enter__()
    # Register a finalizer
    Finalize(resource, resource.__exit__, exitpriority=16)

def hi(*args):
    print("we're in the worker")

if __name__ == "__main__":
    pool = multiprocessing.Pool(initializer=_worker_init, initargs=("abc",))
    pool.map(hi, range(pool._processes))
    pool.close()
    pool.join()

输出:

calling init
in __enter__ of <Process(PoolWorker-1, started daemon)>
calling init
calling init
in __enter__ of <Process(PoolWorker-2, started daemon)>
in __enter__ of <Process(PoolWorker-3, started daemon)>
calling init
in __enter__ of <Process(PoolWorker-4, started daemon)>
we're in the worker
we're in the worker
we're in the worker
we're in the worker
in __exit__ of <Process(PoolWorker-1, started daemon)>
in __exit__ of <Process(PoolWorker-2, started daemon)>
in __exit__ of <Process(PoolWorker-3, started daemon)>
in __exit__ of <Process(PoolWorker-4, started daemon)>

如您所见，当我们对池进行__exit__时，我们所有工作程序中都会调用join()。

关于python - 上下文管理器和多处理池，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/24717468/