跨多进程共享基于异步等待协程的复杂对象

Question

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

我有一个复杂的对象，里面有很多异步的协同进程。 在一个单独的进程中对这个对象运行一个长时间运行的进程的函数。现在，我想在主进程中运行ipythonshell，并在这个复杂对象上操作，而这个长时间运行的进程正在另一个进程中运行。在

为了跨进程共享这个复杂的对象，我尝试了多处理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,),
)

但这会产生错误：

^{pr2}$

它不工作，因为对象中有协同程序。我想不出一个更好的解决方案来让它发挥作用，我只能坚持下去。在

若不是Python，我会为长时间运行的进程生成一个线程，并且仍然能够对其进行操作。在

如果我没说错的话，这应该是多进程应用程序运行后台进程和只对其执行只读操作的主进程的常见模式，就像我的例子一样，而不是修改它。我想知道一般是怎么做的？在

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

Answer 1

我使用多处理模块和异步模块有一段时间了。在

进程之间不共享对象。在一个进程中创建一个对象（referent），返回一个代理对象并与另一个进程共享。另一个进程使用代理对象来调用referent的方法。在

在您的代码中，referent是基于复杂的\u asynio_类实例。在

这是你可以参考的愚蠢代码。主线程是一个运行UDP服务器和其他异步操作服务器的异步循环。长时间运行的进程简单地检查循环状态。在

import multiprocessing
import multiprocessing.managers as m
import asyncio 
import logging
import time 

logging.basicConfig(filename="main.log", level=logging.DEBUG) 

class MyManager(m.BaseManager):
    pass

class sinkServer(asyncio.Protocol):


    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        message = data.decode()
        logging.info('Data received: {!r}'.format(message))


class complex_asynio_based_class:

    def __init__(self, addr=('127.0.0.1', '8080')):
        self.loop = asyncio.new_event_loop() 
        listen = self.loop.create_datagram_endpoint(sinkServer, local_addr=addr,
                    reuse_address=True, reuse_port=True)
        self.loop.run_until_complete(listen)
        for name, delay in zip("abcdef", (1,2,3,4,5,6)):
            self.loop.run_until_complete(self.slow_op(name, delay))

    def run(self):
        self.loop.run_forever() 

    def stop(self):
        self.loop.stop() 

    def is_running(self):
        return self.loop.is_running() 

    async def slow_op(self, name, delay):
        logging.info("my name: {}".format(name))
        asyncio.sleep(delay)

def long_running_process(co):
    logging.debug('address: {!r}'.format(co))
    logging.debug("status: {}".format(co.is_running()))
    time.sleep(6)
    logging.debug("status: {}".format(co.is_running()))

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,),
)
process.start()

c.run()  #run the loop

Answer 2

无法在进程之间自动共享正在运行的协程，因为协程在拥有异步类的进程中的特定事件循环中运行。协同例程的状态不能被pickle，即使可以，它在事件循环的上下文之外也没有意义。在

您可以为您的异步类创建一个基于回调的适配器，每个协程方法都由一个基于回调的方法表示，该方法的语义是“开始做X，完成后调用这个函数”。如果回调是支持多处理的，则可以从其他进程调用这些操作。然后，您可以在每个进程中启动一个事件循环，并在基于代理的回调调用上创建一个协程facade。在

例如，考虑一个普通的异步类：

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“callback hell”样式的经典异步函数：

^{pr2}$

（转换可以自动进行，上面手动编写的代码只是展示了概念。）

CallbackAdapter可以注册为多处理，因此不同的进程可以通过多处理提供的代理来启动适配器的方法（因此也可以启动原始的异步协同程序）。这只要求作为on_success传递的回调是多处理友好的。在

最后一步，我们可以绕一圈，为基于回调的API（）创建一个异步适配器，同时在另一个进程中启动一个事件循环，并使用asyncio和async def。适配器类的适配器将运行一个功能齐全的repeat协程，该协程有效地代理原始的Async.repeat协程，而不必尝试pickle协程状态。在

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

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()

代码在python3.5上正确运行，但在3.6和3.7上由于a bug in multiprocessing而失败。在