asyncio + multiprocessing + unix

Question

我有一个宠物项目,具有以下逻辑:

import asyncio, multiprocessing

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    asyncio.get_event_loop().run_until_complete(sub_main())

def start():
    multiprocessing.Process(target=sub_loop).start()

start()

如果你运行它,你会看到:

Hello from subprocess

那很好.但我要做的就是制作start()协程:

async def start():
    multiprocessing.Process(target=sub_loop).start()

要运行它,我必须做这样的事情:

asyncio.get_event_loop().run_until_complete(start())

这是问题:当创建子进程时,它会克隆整个Python环境,因此事件循环已经在那里运行:

Process Process-1:
Traceback (most recent call last):
  File "/usr/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/usr/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "test.py", line 7, in sub_loop
    asyncio.get_event_loop().run_until_complete(sub_main())
  File "/usr/lib/python3.5/asyncio/base_events.py", line 361, in run_until_complete
    self.run_forever()
  File "/usr/lib/python3.5/asyncio/base_events.py", line 326, in run_forever
    raise RuntimeError('Event loop is running.')
RuntimeError: Event loop is running.

我试图在没有运气的情况下在子进程端销毁它,但我认为正确的方法是防止它与子进程共享.有可能吗？

更新:这是完整的失败代码:

import asyncio, multiprocessing

import asyncio.unix_events

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    asyncio.get_event_loop().run_until_complete(sub_main())


async def start():
    multiprocessing.Process(target=sub_loop).start()

asyncio.get_event_loop().run_until_complete(start())

Answer 1

首先,如果计划从循环内运行python子进程,则应考虑将loop.run_in_executor与ProcessPoolExecutor一起使用.至于您的问题,您可以使用事件循环策略函数来设置新循环:

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(sub_main())

async def start(executor):
    await asyncio.get_event_loop().run_in_executor(executor, sub_loop)

if __name__ == '__main__':
    executor = ProcessPoolExecutor()
    asyncio.get_event_loop().run_until_complete(start(executor))