Asyncio 函数在从脚本而不是从 Flask 路由调用时起作用

Question

我是 Python 和这些库/模块的新手。我正在编写一个简单的 ping 测试网络扫描仪作为学习项目。

我首先使用 asyncio 开发了一个脚本来 ping 网络上的地址

#ip_test.py
import asyncio
import ipaddress

async def ping(addr):
    proc = await asyncio.create_subprocess_exec(
        'ping','-W','1','-c','3',addr,
        stdout=asyncio.subprocess.PIPE
    )
    await proc.wait()
    return proc.returncode

async def pingMain(net):
    #hosts() returns list of Ipv4Address objects
    result = await asyncio.gather(*(ping(str(addr)) for addr in net.hosts()))
    return result

def getHosts(net_): #net_ is an Ipv4Network object
    return asyncio.run(pingMain(net_))
    #Returns list of response codes which I then zip with the list of searched ips

当我打开 python 并运行以下命令时，它按预期工作：

import ip_test as iptest
import ipaddress
print(iptest.getHosts(ipaddress.ip_network('192.168.1.0/29')))
#prints: [0, 0, 0, 1, 1, 1] as expected on this network

然而，最终目标是通过表单输入从用户那里获取输入（结果被记录到数据库中，这是一个用于说明目的的简化示例）。我通过烧瓶路线收集输入：

@app.route("/newscan",methods=['POST'])
def newScan():
    form = request.form
    networkstring = form.get('network') + "/" + form.get('mask')
    result = iptest.getHosts(ipaddress.ip_network(networkstring))
    return result

当我以这种方式调用模块时，出现错误： Runtime Error: Cannot add child handler, the child watcher does not have a loop attached.

为什么当我导入模块并从命令行运行该函数时会起作用，但当我使用来自 Flask 路由的相同输入调用它时却不起作用？

编辑：回溯：

Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 2463, in __call__
  return self.wsgi_app(environ, start_response)
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 2449, in wsgi_app
  response = self.handle_exception(e)
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1866, in handle_exception
  reraise(exc_type, exc_value, tb)
File "/usr/local/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise
  raise value
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 2446, in wsgi_app
  response = self.full_dispatch_request()
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1951, in full_dispatch_request
  rv = self.handle_user_exception(e)
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1820, in handle_user_exception
  reraise(exc_type, exc_value, tb)
File "/usr/local/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise
  raise value
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1949, in full_dispatch_request
  rv = self.dispatch_request()
File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1935, in dispatch_request
  return self.view_functions[rule.endpoint](**req.view_args)
File "/app/app.py", line 41, in newScan
  result = iptest.getHosts(ipaddress.ip_network(networkstring))
File "/app/ip_test.py", line 22, in getHosts
  res = asyncio.run(pingMain(net_))
File "/usr/local/lib/python3.7/asyncio/runners.py", line 43, in run
  return loop.run_until_complete(main)
File "/usr/local/lib/python3.7/asyncio/base_events.py", line 579, in run_until_complete
  return future.result()
File "/app/ip_test.py", line 15, in pingMain
  result = await asyncio.gather(*(ping(str(addr)) for addr in net.hosts()))
File "/app/ip_test.py", line 7, in ping
  stdout=asyncio.subprocess.PIPE
File "/usr/local/lib/python3.7/asyncio/subprocess.py", line 217, in create_subprocess_exec
  stderr=stderr, **kwds)
File "/usr/local/lib/python3.7/asyncio/base_events.py", line 1529, in subprocess_exec
  bufsize, **kwargs)
File "/usr/local/lib/python3.7/asyncio/unix_events.py", line 193, in _make_subprocess_transport
  self._child_watcher_callback, transp)
File "/usr/local/lib/python3.7/asyncio/unix_events.py", line 930, in add_child_handler
  "Cannot add child handler, "
RuntimeError: Cannot add child handler, the child watcher does not have a loop attached

Answer 1

您正在尝试从主线程以外的线程运行异步子进程。这需要从主线程一些初始设置，见子进程和线程部分中的asyncio 子流程文档：

标准的 asyncio 事件循环支持从不同线程运行子进程，但有一些限制：

• 事件循环必须在主线程中运行。
• 在从其他线程执行子进程之前，必须在主线程中实例化子观察者。get_child_watcher()在主线程中调用该函数来实例化子观察者。

这里发生的事情是您的 WSGI 服务器正在使用多个线程来处理传入的请求，因此请求处理程序没有在主线程上运行。但是您的代码用于asyncio.run()启动一个新的事件循环，因此您的asyncio.create_subprocess_exec()调用将失败，因为主线程上没有子观察者。

您必须从主线程启动一个循环（而不是停止它），并调用asyncio.get_child_watcher()该线程，以使您的代码不会失败：

# to be run on the main thread, set up a subprocess child watcher
assert threading.current_thread() is threading.main_thread()
asyncio.get_event_loop()
asyncio.get_child_watcher()

注意：此限制仅适用于 Python 3.7 以下的 Python 版本，该限制已在 Python 3.8 中解除。

然而，仅仅运行一堆子进程并等待它们完成，使用asyncio是矫枉过正的；您的操作系统可以并行运行子进程就好了。只需subprocess.Popen()通过以下Popen.poll()方法使用 和检查每个进程：

import subprocess

def ping_proc(addr):
    return subprocess.Popen(
        ['ping', '-W', '1', '-c', '3', addr],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )

def get_hosts(net):
    # hosts() returns list of Ipv4Address objects
    procs = [ping_proc(str(addr)) for addr in net.hosts()]
    while any(p.poll() is None for p in procs):
        time.sleep(0.1)
    return [p.returncode for p in procs]

Popen.poll()不阻塞；如果Popen.returncode尚未设置，它会检查操作系统的进程状态，waitpid([pid], WNOHANG)并返回None进程是否仍在运行或现在可用的returncode值。以上只是在循环中检查这些状态，中间有一个短暂的睡眠以避免颠簸。

的asyncio子进程封套（POSIX上至少）或者使用SIGCHLD被通知退出或（在Python 3.8）使用每个子进程在单独的线程以使用阻断的子进程的信号处理程序waitpid()上创建的每个子进程调用。您可以实现相同的信号处理程序，但考虑到信号处理程序只能在主线程上注册，因此您必须跳过几个环节才能将传入的SIGCHLD信号信息传递给正确的线程。