如何实现具有实时日志记录和超时选项的 subprocess.Popen ？

Question

我的目标是实现一个 Python 3 方法，该方法将支持运行系统命令（使用子进程），并满足以下几个要求：

• 运行持久命令
• stdout 和 stderr 的实时日志记录
• 如果命令未能按时完成，则强制超时以停止该命令

为了支持实时日志记录，我使用了 2 个线程来处理 stdout 和 stderr 输出。

我的挑战是强制线程和子进程进程超时。

我尝试使用信号处理程序实现超时，似乎在调用处理程序后立即冻结解释器。

我的实施有什么问题吗？

还有其他方法可以实现我的要求吗？

这是我当前的实施尝试：

def run_live_output(cmd, timeout=900, **kwargs):
    full_output = StringIO()

    def log_popen_pipe(p, log_errors=False):
        while p.poll() is None:
            output = ''
            if log_errors:
                output = p.stderr.readline()
                log.warning(f"{output}")
            else:
                output = p.stdout.readline()
                log.info(f"{output}")
            full_output.write(output)

        if p.poll():
            log.error(f"{cmd}\n{p.stderr.readline()}")

    class MyTimeout(Exception):
        pass

    def handler(signum, frame):
        log.info(f"Signal handler called with signal {signum}")
        raise MyTimeout

    with subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        stdin=subprocess.PIPE,
        universal_newlines=True, 
        **kwargs
    ) as sp:
        with ThreadPoolExecutor(2) as pool:
                try:
                    signal.signal(signal.SIGALRM, handler)
                    signal.alarm(timeout)
                    r1 = pool.submit(log_popen_pipe, sp)
                    r2 = pool.submit(log_popen_pipe, sp, log_errors=True)
                    r1.result()
                    r2.result()
                except MyTimeout:
                    log.info(f"Timed out - Killing the threads and process")
                    pool.shutdown(wait=True)
                    sp.kill()
                except Exception as e:
                    log.info(f"{e}")

    return full_output.getvalue()

Answer 1

Q-1）我尝试使用信号处理程序实现超时，似乎在调用处理程序后立即冻结解释器，我的实现有什么问题？

A-1）不，你的信号处理程序没有冻结，有冻结但不在信号处理程序中，信号处理程序很好。当您调用 时，您的主线程被阻塞（冻结）pool.shutdown(wait=True)。因为你的子进程仍在运行并且你while p.poll() is None:在log_popen_pipefunc 中执行。这就是为什么你的主线程在完成之前不会继续log_popen_pipe。

为了解决这个问题，我们需要删除pool.shutdown(wait=True)然后调用sp.terminate(). 我建议您使用sp.terminate()它，sp.kill()因为sp.kill()会发送SIGKILL信号，直到您真正需要它时才首选它。此外，with ThreadPoolExecutor(2) as pool:语句 结束时将被调用，如果func 结束，pool.shutdown(wait=True)这不会阻止您。log_popen_pipe

在您的情况下，log_popen_pipe如果子进程在我们完成时完成，则 func 将完成sp.terminate()。

Q-2）还有其他方法可以实现我的要求吗？

A-2）是的，您可以使用库Timer中的类threading。Timer 类将创建 1 个线程，该线程将等待timeout几秒并在超时秒数结束时，该创建的线程将调用sp.terminatefunc

这是代码：

from io import StringIO
import signal,subprocess
from concurrent.futures import ThreadPoolExecutor
import logging as log
from threading import Timer
log.root.setLevel(log.INFO)

def run_live_output(cmd, timeout=900, **kwargs):
    full_output = StringIO()

    def log_popen_pipe(p, log_errors=False):
        while p.poll() is None:
            output = ''
            if log_errors:
                output = p.stderr.readline()
                log.warning(f"{output}")
            else:
                output = p.stdout.readline()
                log.info(f"{output}")
            full_output.write(output)

        if p.poll()!=None:
            log.error(f"subprocess finished, {cmd}\n{p.stdout.readline()}")


    with subprocess.Popen(
         cmd,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         stdin=subprocess.PIPE,
         universal_newlines=True, 
         **kwargs
    ) as sp:

        Timer(timeout,sp.terminate).start()

        with ThreadPoolExecutor(2) as pool:
            try:
                r1 = pool.submit(log_popen_pipe, sp)
                r2 = pool.submit(log_popen_pipe, sp, log_errors=True)
                r1.result()
                r2.result()

            except Exception as e:
                log.info(f"{e}")

    return full_output.getvalue()


run_live_output(["python3","...."],timeout=4)

顺便说一下，p.poll()将返回returncode已终止的子进程。如果你想获得成功终止子进程的输出，你需要使用if p.poll()==00 通常表示子进程成功终止