Python 子进程 readline() 挂起；不能使用普通选项

Question

首先，我知道这看起来像是重复的。我一直在阅读：

Python subprocess readlines() hangs

Python Subprocess readline hangs() after reading all input

subprocess readline hangs waiting for EOF

但是这些选项要么直接不起作用，要么我无法使用。

问题

# Obviously, swap HOSTNAME1 and HOSTNAME2 with something real
cmd = "ssh -N -f -L 1111:<HOSTNAME1>:80 <HOSTNAME2>"

p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=os.environ)
while True:
    out = p.stdout.readline()
    # Hangs here ^^^^^^^ forever

    out = out.decode('utf-8')
    if out:
        print(out)
    if p.poll() is not None:
        break

我的困境是调用subprocess.Popen()的函数是一个运行bash命令的库函数，所以它需要非常通用并且有以下限制：

• 必须在输入时显示输出；不要阻止然后一次性向屏幕发送垃圾邮件
• 如果父调用者对库函数进行多处理，则不能使用多处理（Python 不允许子进程拥有子进程）
• 不能使用signal.SIGALRM，原因与多处理相同；父调用者可能正在尝试设置自己的超时
• 不能使用第三方非内置模块
• 直接向上穿线不起作用。当readline() 调用在线程中时，thread.join(timeout=1) 让程序继续运行，但 ctrl+c 根本不起作用，调用 sys.exit() 不会退出程序，因为线程仍处于打开状态。如您所知，您不能通过设计来杀死 Python 中的线程。
• 没有任何方式的 bufsize 或其他子进程 args 似乎有所作为；也不会将 readline() 放入迭代器中。

如果我可以杀死一个线程，我会有一个可行的解决方案，但这是超级禁忌，尽管这绝对是一个合法的用例。

我愿意接受任何想法。

Answer 1

一种选择是使用线程发布到队列。然后你可以在队列上阻塞超时。您可以使阅读器线程成为守护程序，这样它就不会阻止系统退出。这是一个草图：

import subprocess
from threading import Thread
from queue import Queue

def reader(stream, queue):
    while True:
        line = stream.readline()
        queue.put(line)
        if not line:
            break

p = subprocess.Popen(cmd, stdout=subprocess.PIPE, ...)
queue = Queue()
thread = Thread(target=reader, args=(p.stdout, queue))
thread.daemon = True
thread.start()
while True:
    out = queue.get(timeout=1)  # timeout is optional
    if not out:  # Reached end of stream
        break
    ...  # Do whatever with output

# Output stream was closed but process may still be running
p.wait()

请注意，您应该根据您的特定用例调整此答案。例如，您可能希望添加一种方法来向读取器线程发出信号以在到达流结束之前停止运行。

另一种选择是轮询输入流，例如这个问题：timeout on subprocess readline in python

Answer 2

我终于找到了一个可行的解决方案；我缺少的关键信息是 thread.daemon = True，@augurar 在他们的回答中指出了这一点。

设置thread.daemon = True允许线程在主进程终止时终止；因此解除阻止我使用子线程来监控readline()。

这是我的解决方案的示例实现；我使用Queue() 对象将字符串传递给主进程，并为我试图解决子进程已完成和终止的原始问题之类的情况实施了一个 3 秒计时器，但 readline() 已挂起一段时间原因。

这也有助于避免事物先完成之间的竞争条件。

这适用于 Python 2 和 3。

import sys
import threading
import subprocess
from datetime import datetime

try:
    import queue
except:
    import Queue as queue # Python 2 compatibility


def _monitor_readline(process, q):
    while True:
        bail = True
        if process.poll() is None:
            bail = False
        out = ""
        if sys.version_info[0] >= 3:
            out = process.stdout.readline().decode('utf-8')
        else:
            out = process.stdout.readline()
        q.put(out)
        if q.empty() and bail:
            break

def bash(cmd):
    # Kick off the command
    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True)

    # Create the queue instance
    q = queue.Queue()
    # Kick off the monitoring thread
    thread = threading.Thread(target=_monitor_readline, args=(process, q))
    thread.daemon = True
    thread.start()
    start = datetime.now()
    while True:
        bail = True
        if process.poll() is None:
            bail = False
            # Re-set the thread timer
            start = datetime.now()
        out = ""
        while not q.empty():
            out += q.get()
        if out:
            print(out)

        # In the case where the thread is still alive and reading, and
        # the process has exited and finished, give it up to 3 seconds
        # to finish reading
        if bail and thread.is_alive() and (datetime.now() - start).total_seconds() < 3:
            bail = False
        if bail:
            break

# To demonstrate output in realtime, sleep is called in between these echos
bash("echo lol;sleep 2;echo bbq")