为什么线程中的 python asyncio 进程在 Linux 上似乎不稳定？

Question

我尝试从 Qt 应用程序运行 python3 异步外部命令。在我使用多处理线程来执行此操作而不冻结 Qt 应用程序之前。但是现在，我想用QThread 来做这件事，以便能够腌制并给出QtWindows 作为其他一些函数的参数（这里没有介绍）。我做到了，并在我的Windows 操作系统上成功测试了它，但我在Linux 操作系统上尝试了该应用程序，我收到以下错误：RuntimeError: Cannot add child handler, the child watcher does not have a loop attached

从那时起，我尝试隔离问题，并获得了下面复制问题的最小（尽可能）示例。 当然，正如我之前提到的，如果我将QThreadPool 替换为multiprocessing.thread 的列表，则此示例运行良好。我还意识到了令我惊讶的事情：如果我在示例的最后部分取消注释 rc = subp([sys.executable,"./HelloWorld.py"]) 行，它也可以工作。我无法解释为什么。

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

## IMPORTS ##
from functools import partial
from PyQt5 import QtCore
from PyQt5.QtCore import QThreadPool, QRunnable, QCoreApplication
import sys
import asyncio.subprocess

# Global variables
Qpool = QtCore.QThreadPool()


def subp(cmd_list):
    """ """

    if sys.platform.startswith('linux'):
        new_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(new_loop)
    elif sys.platform.startswith('win'):
        new_loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
        asyncio.set_event_loop(new_loop)
    else :
        print('[ERROR]     OS not available for encodage... EXIT')
        sys.exit(2)

    rc, stdout, stderr= new_loop.run_until_complete(get_subp(cmd_list) )
    new_loop.close()
    if rc!=0 :
        print('Exit not zero ({}): {}'.format(rc, sys.exc_info()[0]) )#, exc_info=True)
    return rc, stdout, stderr

async def get_subp(cmd_list):
    """ """

    print('subp: '+' '.join(cmd_list) )
    # Create the subprocess, redirect the standard output into a pipe
    create = asyncio.create_subprocess_exec(*cmd_list, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) #
    proc = await create

    # read child's stdout/stderr concurrently (capture and display)
    try:
        stdout, stderr = await asyncio.gather(
            read_stream_and_display(proc.stdout),
            read_stream_and_display(proc.stderr))
    except Exception:
        proc.kill()
        raise
    finally:
        rc = await proc.wait()
        print(" [Exit {}] ".format(rc)+' '.join(cmd_list))
    return rc, stdout, stderr

async def read_stream_and_display(stream):
    """ """
    async for line in stream:
        print(line, flush=True)

class Qrun_from_job(QtCore.QRunnable):
    def __init__(self, job, arg):
        super(Qrun_from_job, self).__init__()
        self.job=job
        self.arg=arg

    def run(self):
        code = partial(self.job)
        code()

def ThdSomething(job,arg):
    testRunnable = Qrun_from_job(job,arg)
    Qpool.start(testRunnable)

def testThatThing():
    rc = subp([sys.executable,"./HelloWorld.py"])


if __name__=='__main__':
    app = QCoreApplication([])
    # rc = subp([sys.executable,"./HelloWorld.py"])
    ThdSomething(testThatThing,'tests')
    sys.exit(app.exec_())

使用 HelloWorld.py 文件：

#!/usr/bin/env python3
import sys
if __name__=='__main__':
   print('HelloWorld')
   sys.exit(0)

因此我有两个问题：如何使这个示例与 QThread 正常工作？为什么之前调用异步任务（调用subp 函数）会改变Linux 上示例的稳定性？

编辑

根据@user4815162342 的建议，我尝试使用run_coroutine_threadsafe 使用下面的代码。但它不起作用并返回相同的错误，即RuntimeError: Cannot add child handler, the child watcher does not have a loop attached。我还尝试通过模块 mutliprocessing 中的等效命令来更改 threading 命令；最后一个，命令subp 永远不会启动。

代码：

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

## IMPORTS ##
import sys
import asyncio.subprocess
import threading
import multiprocessing

# at top-level
loop = asyncio.new_event_loop()

def spin_loop():
    asyncio.set_event_loop(loop)
    loop.run_forever()

def subp(cmd_list):
    # submit the task to asyncio
    fut = asyncio.run_coroutine_threadsafe(get_subp(cmd_list), loop)
    # wait for the task to finish
    rc, stdout, stderr = fut.result()
    return rc, stdout, stderr

async def get_subp(cmd_list):
    """ """
    print('subp: '+' '.join(cmd_list) )
    # Create the subprocess, redirect the standard output into a pipe
    proc = await asyncio.create_subprocess_exec(*cmd_list, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) #

    # read child's stdout/stderr concurrently (capture and display)
    try:
        stdout, stderr = await asyncio.gather(
            read_stream_and_display(proc.stdout),
            read_stream_and_display(proc.stderr))
    except Exception:
        proc.kill()
        raise
    finally:
        rc = await proc.wait()
        print(" [Exit {}] ".format(rc)+' '.join(cmd_list))
    return rc, stdout, stderr

async def read_stream_and_display(stream):
    """ """
    async for line in stream:
        print(line, flush=True)

if __name__=='__main__':
    threading.Thread(target=spin_loop, daemon=True).start()
    # multiprocessing.Process(target=spin_loop, daemon=True).start()
    print('thread passed')
    rc = subp([sys.executable,"./HelloWorld.py"])
    print('end')
    sys.exit(0)

Answer 1

作为一般设计原则，创建新的事件循环只是为了运行单个子例程是不必要且浪费的。相反，创建一个事件循环，在单独的线程中运行它，并通过使用asyncio.run_coroutine_threadsafe 向它提交任务来满足您所有的 asyncio 需求。

例如：

# at top-level
loop = asyncio.new_event_loop()
def spin_loop():
    asyncio.set_event_loop(loop)
    loop.run_forever()

asyncio.get_child_watcher().attach_loop(loop)
threading.Thread(target=spin_loop, daemon=True).start()
# ... the rest of your code ...

有了这个，您可以使用以下命令轻松地从任何线程执行任何 asyncio 代码：

def subp(cmd_list):
    # submit the task to asyncio
    fut = asyncio.run_coroutine_threadsafe(get_subp(cmd_list), loop)
    # wait for the task to finish
    rc, stdout, stderr = fut.result()
    return rc, stdout, stderr

请注意，您可以使用add_done_callback 在asyncio.run_coroutine_threadsafe 返回的未来完成时收到通知，因此您可能一开始就不需要线程。

请注意，与事件循环的所有交互都应通过上述run_coroutine_threadsafe（提交协程时）或当您需要事件循环调用普通函数时通过loop.call_soon_threadsafe。例如，要停止事件循环，您可以调用 loop.call_soon_threadsafe(loop.stop)。

Answer 2

我怀疑你正在做的事情根本不受支持 - 根据documentation：

要处理信号和执行子进程，事件循环必须在主线程中运行。

当您尝试执行子进程时，我认为在另一个线程中运行新的事件循环不起作用。

问题是，Qt 已经有一个事件循环，你真正需要的是说服asyncio 使用它。这意味着您需要一个事件循环实现，它提供在“Qt 的事件循环”之上实现的“异步事件循环接口”。

我相信asyncqt 提供了这样的实现。您可能想尝试使用QEventLoop(app) 代替asyncio.new_event_loop()。