在多进程中共享异步 - 等待基于协同程序的复杂对象

Question

我一般都知道,不应该在多进程之间共享对象以及可能由此产生的问题.但我的要求是必须这样做.

我有一个复杂的对象,其中包含所有漂亮的协同程序async-await.一个函数,它在自己的单独进程中对此对象运行长时间运行的进程.现在,我想在主进程中运行一个IPython shell,并对这个复杂对象进行操作,而长时间运行的进程正在另一个进程中运行.

为了跨进程共享这个复杂的对象,我尝试了在SO上遇到的多处理BaseManager方法:

import multiprocessing
import multiprocessing.managers as m


class MyManager(m.BaseManager):
    pass

MyManager.register('complex_asynio_based_class', complex_asynio_based_class)
manager = MyManager()
manager.start()
c = manager.complex_asynio_based_class()

process = multiprocessing.Process(
     target=long_running_process,
     args=(c,),
)

但这会给出错误:

Unserializable message: Traceback (most recent call last):
  File "/usr/3.6/lib/python3.6/multiprocessing/managers.py", line 283, in serve_client
    send(msg)
  File "/usr/3.6/lib/python3.6/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/usr/3.6/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: can't pickle coroutine objects

它不起作用,因为对象中有协同程序.我无法想出一个更好的解决方案来让它工作,我坚持下去.

如果它不是Python,我会为长时间运行的进程生成一个线程,并且仍然可以对它进行操作.

如果我没有错,这应该是多进程应用程序运行后台进程的常用模式,以及只对它执行一些只读操作的主进程,就像我的情况一样,而不是修改它.我想知道它是如何完成的？

在多进程中如何共享无法拾取的复杂对象？

Answer 1

无法在进程之间自动共享运行协程,因为协程在拥有异步类的进程中的特定事件循环内运行.协程具有无法腌制的状态,即使可能,它在事件循环的上下文之外也没有意义.

你可以做的是为异步类创建一个基于回调的适配器,每个协程方法由一个基于回调的方法表示,语义为"开始执行X并在完成后调用此函数".如果回调是多处理感知的,则可以从其他进程调用这些操作.然后,您可以在每个进程中启动一个事件循环,并在代理的基于回调的调用上创建一个协程外观.

例如,考虑一个简单的异步类:

class Async:
    async def repeat(self, n, s):
        for i in range(n):
            print(s, i, os.getpid())
            await asyncio.sleep(.2)
        return s

基于回调的适配器可以使用公共asyncioAPI将repeat协同程序转换为JavaScript"回调地狱"样式的经典异步函数:

class CallbackAdapter:
    def repeat_start(self, n, s, on_success):
        fut = asyncio.run_coroutine_threadsafe(
            self._async.repeat(n, s), self._loop)
        # Once the coroutine is done, notify the caller.
        fut.add_done_callback(lambda _f: on_success(fut.result()))

(转换可以自动化,上面手动编写的代码只显示了这个概念.)

CallbackAdapter可以使用多处理注册,因此不同的进程可以通过多处理提供的代理启动适配器的方法(以及原始的异步协程).这只需要将回调作为on_success多处理友好的方式传递.

作为最后一步,可以完全循环并为基于回调的API(!)创建异步适配器,在其他进程中启动事件循环,并且还使用asyncio和async def.这个适配器适配器类将运行一个功能齐全的repeat协程,它可以有效地代理原始Async.repeat协程,而无需尝试腌制协程状态.

以下是上述方法的示例实现:

import asyncio, multiprocessing.managers, threading, os

class Async:
    # The async class we are bridging.  This class is unaware of multiprocessing
    # or of any of the code that follows.
    async def repeat(self, n, s):
        for i in range(n):
            print(s, i, 'pid', os.getpid())
            await asyncio.sleep(.2)
        return s


def start_asyncio_thread():
    # Since the manager controls the main thread, we have to spin up the event
    # loop in a dedicated thread and use asyncio.run_coroutine_threadsafe to
    # submit stuff to the loop.
    setup_done = threading.Event()
    loop = None
    def loop_thread():
        nonlocal loop
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        setup_done.set()
        loop.run_forever()
    threading.Thread(target=loop_thread).start()
    setup_done.wait()
    return loop

class CallbackAdapter:
    _loop = None

    # the callback adapter to the async class, also running in the
    # worker process
    def __init__(self, obj):
        self._async = obj
        if CallbackAdapter._loop is None:
            CallbackAdapter._loop = start_asyncio_thread()

    def repeat_start(self, n, s, on_success):
        # Submit a coroutine to the event loop and obtain a Task/Future.  This
        # is normally done with loop.create_task, but repeat_start will be
        # called from the main thread, owned by the multiprocessng manager,
        # while the event loop will run in a separate thread.
        future = asyncio.run_coroutine_threadsafe(
            self._async.repeat(n, s), self._loop)
        # Once the coroutine is done, notify the caller.
        # We could propagate exceptions by accepting an additional on_error
        # callback, and nesting fut.result() in a try/except that decides
        # whether to call on_success or on_error.
        future.add_done_callback(lambda _f: on_success(future.result()))


def remote_event_future(manager):
    # Return a function/future pair that can be used to locally monitor an
    # event in another process.
    #
    # The returned function and future have the following property: when the
    # function is invoked, possibly in another process, the future completes.
    # The function can be passed as a callback argument to a multiprocessing
    # proxy object and therefore invoked by a different process.
    loop = asyncio.get_event_loop()
    result_pipe = manager.Queue()
    future = loop.create_future()
    def _wait_for_remote():
        result = result_pipe.get()
        loop.call_soon_threadsafe(future.set_result, result)
    t = threading.Thread(target=_wait_for_remote)
    t.start()
    return result_pipe.put, future


class AsyncAdapter:
    # The async adapter for a callback-based API, e.g. the CallbackAdapter.
    # Designed to run in a different process and communicate to the callback
    # adapter via a multiprocessing proxy.
    def __init__(self, cb_proxy, manager):
        self._cb = cb_proxy
        self._manager = manager

    async def repeat(self, n, s):
        set_result, future = remote_event_future(self._manager)
        self._cb.repeat_start(n, s, set_result)
        return await future


class CommManager(multiprocessing.managers.SyncManager):
    pass

CommManager.register('Async', Async)
CommManager.register('CallbackAdapter', CallbackAdapter)


def get_manager():
    manager = CommManager()
    manager.start()
    return manager

def other_process(manager, cb_proxy):
    print('other_process (pid %d)' % os.getpid())
    aadapt = AsyncAdapter(cb_proxy, manager)
    loop = asyncio.get_event_loop()
    # Create two coroutines printing different messages, and gather their
    # results.
    results = loop.run_until_complete(asyncio.gather(
        aadapt.repeat(3, 'message A'),
        aadapt.repeat(2, 'message B')))
    print('coroutine results (pid %d): %s' % (os.getpid(), results))
    print('other_process (pid %d) done' % os.getpid())

def start_other_process(loop, manager, async_proxy):
    cb_proxy = manager.CallbackAdapter(async_proxy)
    other = multiprocessing.Process(target=other_process,
                                    args=(manager, cb_proxy,))
    other.start()
    return other

def main():
    loop = asyncio.get_event_loop()
    manager = get_manager()
    async_proxy = manager.Async()
    # Create two external processes that drive coroutines in our event loop.
    # Note that all messages are printed with the same PID.
    start_other_process(loop, manager, async_proxy)
    start_other_process(loop, manager, async_proxy)
    loop.run_forever()

if __name__ == '__main__':
    main()

代码在Python 3.5上正确运行,但由于多处理中的错误而在3.6和3.7上失败.