python - select() 的这种用法有什么问题？

Question

基本上，我想编写一个函数，将 subprocess.Popen 的 stdout 和 stderr (或任何输入文件描述符的一般情况)合并到一个生成器中，该生成器为我提供 (file_descriptor,行) 元组。

我的第一次尝试是这样的: 从选择导入选择 导入子流程

def _merge_proc_output( process ):
    inputs = (process.stdout, process.stderr)
    while process.poll() is None:
        for f in select(inputs , (), () )[0]:
            line = f.readline()
            if len(line): yield f, line

它似乎在大部分时间都有效。例如，10 次运行中只有一次被损坏(重点是它是随机)。有时它会遗漏一些行，我认为它总是在末尾。不幸的是，我无法一致地重现它，因此很难调试。

任何人都可以看到上面的代码有什么问题会导致它从一个流的末尾删除行吗？

目前我使用更多资源消耗和冗长的代码，但更可移植:

import subprocess
import threading
from Queue import Queue

def _merge_proc_output( process ):
    q = Queue()
    def push(fd):
        for l in fd:
            q.put((fd, l))
        q.put(None)

    pipes   = (process.stdout, process.stderr)
    threads = [ threading.Thread( target = push, args = (fd,) ) for fd in pipes ]

    [ t.start() for t in threads ]

    for t in threads:
        while True:
            w = q.get()
            if w is None:
                break
            yield w

    [ t.join() for t in threads ]

这似乎工作正常(或者至少我还没有注意到问题)。我仍然想知道我的原始代码有什么问题。

附注如果您发现第二个解决方案存在问题，请也对此发表评论。

编辑:嗯，也许我知道为什么会发生这种情况。假设我观察到只有最后一行丢失，可能是 process.poll() 返回了一些东西，但是这些流的输出缓冲区中仍然有东西。我通过添加尝试从输出流读取所有内容的循环来修改原始函数:

def _merge_proc_output( process ):
    inputs = (process.stdout, process.stderr)
    while process.poll() is None:
        for f in select(inputs , (), () )[0]:
            line = f.readline()
            if len(line): yield f, line
    for i in inputs:
        for l in i:
            yield i,l

我必须玩一会儿才能检查这是否可以解决我的问题。

Answer 1

当子进程终止时，您立即停止读取其输出。这意味着如果您没有读完已经生成的内容，您就会丢失最后的行。