蟒蛇asyncio从3.4迁移到3.5+

Question

大家晚上好，我试图创建互联网机器人，我遇到了问题，而我的脚本从python 3.4迁移到3.5或3.6 +。它采用ASYNCIO和3.4 Python的工作很好，但是当我python3.5启动+我得到了错误：RuntimeError: Cannot run the event loop while another loop is running

这里是代码方案：

import multiprocessing as mp 
import asyncio 
import concurrent.futures 
import aiohttp 

def create_proccesses(separate_loop_creator, coro): 
    proccesses = [] 
    for n in range(2): 
     proc = mp.Process(target=separate_loop_creator, args=(coro,)) 
     proc.start() 
     proccesses.append(proc) 
    for p in proccesses: 
     p.join() 

def separate_loop_creator(coro): 
    sep_loop = asyncio.new_event_loop() 
    asyncio.set_event_loop(sep_loop) 
    tasks = [asyncio.async(coro(sep_loop)) for _ in range(100)] 
    try: 
     sep_loop.run_until_complete(asyncio.wait(tasks)) 
     sep_loop.close() 
    except Exception as err: 
     print(err) 
     for task in tasks: 
      task.cancel() 
     sep_loop.close() 


@asyncio.coroutine 
def manager(exe, loop): 
    # some calculations and start coros in several processes 
    loop.run_in_executor(
     exe, 
     create_proccesses, 
     separate_loop_creator, 
     some_coro 
    ) 

@asyncio.coroutine 
def some_work_in_mainloop(): 
    while True: 
     print('Some server dealing with connections here...') 
     yield from asyncio.sleep(1) 

@asyncio.coroutine 
def some_coro(loop): 
    with aiohttp.ClientSession(loop=loop) as session: 
     response = yield from session.get('http://google.com') 
     yield from asyncio.sleep(2) 
     print(response.status) 

if __name__ == '__main__': 
    mainloop = asyncio.get_event_loop() 
    executor = concurrent.futures.ProcessPoolExecutor(5) 
    asyncio.async(some_work_in_mainloop()) 
    asyncio.async(manager(executor, mainloop)) 
    try: 
     mainloop.run_forever() 
    finally: 
     mainloop.close() 

的例外separate_loop_creator()协程提高，这是RuntimeError: Cannot run the event loop while another loop is running。我想这是因为改变了机器人，但我不明白我的代码有什么问题。

这是我想做的事：

     +--------------+ 
       +-------+other service | 
    +----------+  +--------------+ 
    | mainloop | 
    +----------+ 
       |  +------------+ 
       +-----+ executor | 
        +------+-----+ 
          | 
        +------+--------+ 
        |start proccess | 
        +---+-------+---+ 
+-----------------+  |  |  +---------------+ 
|start new loop +------+  +------+ start new loop| 
+--------+--------+      +-------+-------+ 
     |          | 
+-------+-------+      +------v-------+ 
| run coro |      | run coro  | 
+---------------+      +--------------+ 

下面是跟踪我得到python3.5.3：

Traceback (most recent call last): 
    File "tst.py", line 21, in separate_loop_creator 
    sep_loop.run_until_complete(asyncio.wait(tasks)) 
    File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 454, in run_until_complete 
    self.run_forever() 
    File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 411, in run_forever 
    'Cannot run the event loop while another loop is running') 
RuntimeError: Cannot run the event loop while another loop is running 
Cannot run the event loop while another loop is running 
Traceback (most recent call last): 
    File "tst.py", line 21, in separate_loop_creator 
    sep_loop.run_until_complete(asyncio.wait(tasks)) 
    File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 454, in run_until_complete 
    self.run_forever() 
    File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 411, in run_forever 
    'Cannot run the event loop while another loop is running') 
RuntimeError: Cannot run the event loop while another loop is running 

的Python 3.4.3结果：

... 
200 
Some server dealing with connections here... 
200 
200 
Some server dealing with connections here... 
200 
200 
Some server dealing with connections here... 
200 
... 

Answer 1

这实际上是CPython 3.6.0中的asyncio中的一个错误。有一个PR来解决这个问题，所以3.6.1按预期工作。

正如你可以在你的项目中添加下面的一段代码解决方法：

import sys 

if sys.version_info[:3] == (3, 6, 0): 
    import asyncio.events as _ae 
    import os as _os 

    _ae._RunningLoop._pid = None 

    def _get_running_loop(): 
     if _ae._running_loop._pid == _os.getpid(): 
      return _ae._running_loop._loop 

    def _set_running_loop(loop): 
     _ae._running_loop._pid = _os.getpid() 
     _ae._running_loop._loop = loop 

    _ae._get_running_loop = _get_running_loop 
    _ae._set_running_loop = _set_running_loop 

Answer 2

如果可能，最好的解决方法是尝试从您的程序中完全删除multiprocessing，并仅使用一个事件循环（Optiona对孤立的CPU密集型任务使用ProcessPoolExecutor）。

截至2017-03-02，这个问题存在一个开放的python bug，影响非windows平台：https://bugs.python.org/issue22087。

下面是一个较短的程序触发了同样的问题：（！小心在您自担风险使用！）

import asyncio 
import multiprocessing as mp 


def create_another_loop(): 
    loop = asyncio.new_event_loop() 
    loop.run_forever() 


async def create_process(): 
    proc = mp.Process(target=create_another_loop) 
    proc.start() 
    proc.join() 


if __name__ == '__main__': 
    main_loop = asyncio.get_event_loop() 
    main_loop.run_until_complete(create_process()) 
    main_loop.close() 

一个hackish的解决方法从修复灵感这里建议https://github.com/python/asyncio/pull/497是将此代码添加到新创建Process：

if asyncio.events._running_loop: 
    asyncio.events._running_loop._loop = None 

实施例：

import asyncio 
import multiprocessing as mp 
import time 
from concurrent.futures.process import ProcessPoolExecutor 


async def clock(label, n=5, sleep=1): 
    print(label, "start") 
    for i in range(n): 
     await asyncio.sleep(sleep) 
     print(label, i + 1) 
    print(label, "end") 
    return label 


def create_another_loop(): 
    # HACK START 
    if asyncio.events._running_loop: 
     asyncio.events._running_loop._loop = None 
    # HACK END 

    loop = asyncio.new_event_loop() 
    loop.run_until_complete(clock("sub")) 
    loop.close() 


def create_process(): 
    time.sleep(2) 
    proc = mp.Process(target=create_another_loop) 
    proc.start() 
    proc.join() 
    return "ok" 


async def create_process_in_pool(): 
    return await main_loop.run_in_executor(ProcessPoolExecutor(), create_process) 


if __name__ == '__main__': 
    main_loop = asyncio.get_event_loop() 
    tasks = (
     clock("main"), 
     create_process_in_pool(), 
    ) 
    print(main_loop.run_until_complete(asyncio.gather(*tasks))) 
    main_loop.close() 

其他可能的解决方法：在启动循环之前创建流程，或使用asyncio.create_subprocess_exec甚至允许communicate with the subprocess via a stream。