在 popen.stdout.readline 上检测流的结尾答案

【问题标题】：Detecting the end of the stream on popen.stdout.readline在 popen.stdout.readline 上检测流的结尾
【发布时间】：2013-01-29 06:21:49
【问题描述】：

我有一个 python 程序，它使用Popen 启动子进程，并在生成时几乎实时地使用它们的输出。相关循环的代码为：

def run(self, output_consumer):
    self.prepare_to_run()
    popen_args = self.get_popen_args()
    logging.debug("Calling popen with arguments %s" % popen_args)
    self.popen = subprocess.Popen(**popen_args)
    while True:
        outdata = self.popen.stdout.readline()
        if not outdata and self.popen.returncode is not None:
            # Terminate when we've read all the output and the returncode is set
            break
        output_consumer.process_output(outdata)
        self.popen.poll()  # updates returncode so we can exit the loop
    output_consumer.finish(self.popen.returncode)
    self.post_run()

def get_popen_args(self):
    return {
        'args': self.command,
        'shell': False, # Just being explicit for security's sake
        'bufsize': 0,   # More likely to see what's being printed as it happens
                        # Not guarantted since the process itself might buffer its output
                        # run `python -u` to unbuffer output of a python processes
        'cwd': self.get_cwd(),
        'env': self.get_environment(),
        'stdout': subprocess.PIPE,
        'stderr': subprocess.STDOUT,
        'close_fds': True,  # Doesn't seem to matter
    }

这在我的生产机器上效果很好，但在我的开发机器上，当某些子进程完成时，对 .readline() 的调用会挂起。也就是说，它将成功处理所有输出，包括最后的输出行“处理完成”，但随后将再次轮询 readline 并且永远不会返回。对于我调用的大多数子进程，此方法在开发机器上正确退出，但对于一个本身调用许多子进程的复杂 bash 脚本始终无法退出。

值得注意的是，popen.returncode 在输出结束前多行被设置为非None（通常为0）值。所以我不能在设置循环时跳出循环，否则我会丢失在进程结束时吐出的所有内容，并且仍在缓冲等待读取。问题是当我在那个时候刷新缓冲区时，我不知道我什么时候结束，因为最后一次调用readline() 挂起。呼叫read() 也会挂起。调用read(1) 让我每一个字符都出来，但在最后一行之后也会挂起。 popen.stdout.closed 始终是 False。我怎么知道我什么时候结束了？

所有系统都在 Ubuntu 12.04LTS 上运行 python 2.7.3。 FWIW，stderr 正在使用 stderr=subprocess.STDOUT 与 stdout 合并。

为什么不一样？是否由于某种原因未能关闭stdout？子子流程能否以某种方式使其保持打开状态？可能是因为我是从我的开发盒上的终端启动进程，但在生产中它是通过supervisord 作为守护进程启动的？这会改变管道的处理方式吗？如果是，我该如何规范它们？

【问题讨论】：

问题不是你从一个不再存在的进程中读取一行吗？
我不这么认为。如果错误就这么简单，那么它会在任何时候都失败。
.poll() 返回.returncode 即可以直接测试。在末尾添加close_fds=True 和self.popen.stdout.close()（清理）。 bufsize 是什么？您是否尝试过expect_unbuffer、stdbuf 等来防止子进程侧的块缓冲？你可以试试select。
听上去，问题不在于 Python 代码，而在于这个“一个复杂的 bash 脚本”。您能否提供一个可用于重现故障的最小脚本示例？
您可以关闭子 bash 进程的管道 (stdout + stderr)，当它的输出被这些命令 exec 1>&- exec 2>&- 删除时，正如我在我的回答中描述的那样。我认为来自已接受答案的建议不如明确关闭管道那么强大。

标签： python python-2.7 popen

【解决方案1】：

主代码循环看起来不错。可能是管道没有关闭，因为另一个进程使其保持打开状态。例如，如果脚本启动一个写入stdout 的后台进程，那么管道将不会关闭。您确定没有其他子进程仍在运行吗？

一个想法是在看到.returncode 已设置时更改模式。一旦你知道主进程已经完成，从缓冲区读取它的所有输出，但不要等待。您可以使用 select 从管道中读取超时。设置几秒钟的超时，您可以清除缓冲区而不会卡在等待子进程中。

【讨论】：

【解决方案2】：

在不知道导致问题的“一个复杂的 bash 脚本”的内容的情况下，确定确切原因的可能性太多。

但是，如果您在supervisord 下运行您的 Python 脚本，则关注您声称它可以工作的事实，如果子进程试图从标准输入读取，它可能会卡住，或者如果标准输入是，它可能会表现不同一个 tty，（我认为）supervisord 将从 /dev/null 重定向。

这个最小的例子似乎可以更好地处理我的例子 test.sh 运行尝试从标准输入读取的子进程...

import os
import subprocess

f = subprocess.Popen(args='./test.sh',
                     shell=False,
                     bufsize=0,
                     stdin=open(os.devnull, 'rb'),
                     stdout=subprocess.PIPE,
                     stderr=subprocess.STDOUT,
                     close_fds=True)

while 1:
    s = f.stdout.readline()
    if not s and f.returncode is not None:
        break
    print s.strip()
    f.poll()
print "done %d" % f.returncode

否则，您始终可以退回到使用 non-blocking read，并在您的最终输出行显示“进程完成”时退出，尽管这有点小技巧。

【讨论】：

【解决方案3】：

如果你使用 readline() 或 read()，它不应该挂起。无需检查返回码或 poll()。如果在您知道进程完成时它挂起，则很可能是一个子进程保持您的管道打开，正如其他人之前所说的那样。

您可以做两件事来调试它： * 尝试使用最小的脚本而不是当前复杂的脚本来重现，或者 * 用strace -f -e clone,execve,exit_group 运行那个复杂的脚本，看看那个脚本是什么开始的，如果有任何进程在主脚本中存活下来（检查主脚本何时调用exit_group，如果strace 在那之后还在等待，你有一个孩子还活着） .

【讨论】：

【解决方案4】：

为什么要将 sdterr 设置为 STDOUT？

对子进程进行communicate() 调用的真正好处是，您可以检索包含stdout 响应和stderr 消息的元组。

如果逻辑取决于他们的成功或失败，这些可能会很有用。

此外，它还可以让您免于不得不遍历行的痛苦。 Communicate() 为您提供一切，并且没有关于是否收到完整消息的未解决问题

【讨论】：

communicate 等待进程终止。我的应用程序需要在生成输出时对其进行处理。

【解决方案5】：

我发现对read（或readline）的调用有时会挂起，尽管之前调用的是poll。所以我求助于select 来查看是否有可读数据。但是，如果进程关闭，没有超时的select 也会挂起。所以我在半忙循环中调用 select，每次迭代都有一个很小的超时（见下文）。

我不确定您是否可以将其调整为 readline，因为如果最终的 \n 丢失，或者进程在关闭其标准输入和/或终止它之前没有关闭其标准输出，则 readline 可能会挂起。您可以将其包装在生成器中，并且每次在 stdout_collected 中遇到 \n 时，生成当前行。

另外请注意，在我的实际代码中，我使用伪终端 (pty) 来包装 popen 句柄（以更接近地伪造用户输入），但它应该可以在没有的情况下工作。

# handle to read from
handle = self.popen.stdout

# how many seconds to wait without data
timeout = 1

begin = datetime.now()
stdout_collected = ""

while self.popen.poll() is None:
    try:
        fds = select.select([handle], [], [], 0.01)[0]
    except select.error, exc:
        print exc
        break

    if len(fds) == 0:
        # select timed out, no new data
        delta = (datetime.now() - begin).total_seconds()
        if delta > timeout:
            return stdout_collected

        # try longer
        continue
    else:
        # have data, timeout counter resets again
        begin = datetime.now()

    for fd in fds:
        if fd == handle:
            data = os.read(handle, 1024)
            # can handle the bytes as they come in here
            # self._handle_stdout(data)
            stdout_collected += data

# process exited
# if using a pseudoterminal, close the handles here
self.popen.wait()

【讨论】：

【解决方案6】：

我用 bash 子进程编写了一个 demo，可以轻松探索。在readline() 的输出中，'' 可以识别封闭管道，而空行的输出是'\n'。

from subprocess import Popen, PIPE, STDOUT
p = Popen(['bash'], stdout=PIPE, stderr=STDOUT)
out = []
while True:
    outdata = p.stdout.readline()
    if not outdata:
        break
    #output_consumer.process_output(outdata)
    print "* " + repr(outdata)
    out.append(outdata)
print "* closed", repr(out)
print "* returncode", p.wait()

输入/输出示例：关闭管道明显在终止进程之前。这就是为什么应该使用 wait() 而不是 poll()

[prompt] $ python myscript.py
echo abc
* 'abc\n'
exec 1>&- # close stdout
exec 2>&- # close stderr
* closed ['abc\n']
exit
* returncode 0
[prompt] $

对于这种情况，您的代码确实输出了大量的空字符串。

示例：在最后一行没有'\n' 的快速终止进程：

echo -n abc
exit
* 'abc'
* closed ['abc']
* returncode 0

【讨论】：