我一般都知道，不应该在多进程之间共享对象以及可能由此产生的问题。但我的要求是必须这样做。
我有一个复杂的对象，其中包含所有漂亮的协同程序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，我会为长时间运行的进程生成一个线程，并且仍然可以对它进行操作。
如果我没有错，这应该是多进程应用程序运行后台进程的常用模式，以及只对它执行一些只读操作的主进程，就像我的情况一样，而不是修改它。我想知道它是如何完成的？
在多进程中如何共享无法拾取的复杂对象？
无法在进程之间自动共享运行协同程序，因为协程在拥有异步类的进程中的特定事件循环内运行。协程具有无法腌制的状态，即使可能，它在事件循环的上下文之外也没有意义。
你可以做的是为异步类创建一个基于回调的适配器，每个协程方法由一个基于回调的方法表示，语义为“开始执行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
基于回调的适配器可以使用公共asyncio API将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上正确运行，但由于a bug in multiprocessing而在3.6和3.7上失败。
我一直在使用多处理模块和asyncio模块。
您不在进程之间共享对象。您在一个进程中创建一个对象（指示对象），返回一个代理对象并与其他进程共享它。其他进程使用代理对象来调用referent的方法。
在您的代码中，referent是complex_asynio_based_class实例。
这是您可以参考的愚蠢代码。主线程是运行UDP服务器的单个asyncio循环和其他异步操作。长时间运行的过程简单地检查循环状态。
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