缓冲过程输出导致截断？答案

【问题标题】：buffering of process output causing truncation?缓冲过程输出导致截断？
【发布时间】：2016-08-06 00:16:01
【问题描述】：

我有一个包含 100 行的示例文件，我正在使用带有cat 的子进程读取它。但是，队列中的输出总是被截断。我怀疑这可能是由于cat 缓冲其输出，因为它检测到管道。

p = subprocess.Popen("cat file.txt",
                     stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE,
                     stdin=subprocess.PIPE,
                     shell=True,
                     bufsize=0)

我正在使用单独的线程从cat 的 stdout 和 stderr 管道中读取：

def StdOutThread():
  while not p.stdout.closed and running:
    line = ""
    while not line or line[-1] != "\n":
      r = p.stdout.read(1)
      if not r:
        break
      line += r
      pending_line["line"] = line

    if line and line[-1] == "\n":
      line = line[:-1]
    if line:
      queue.put(("out", line))

这些线程被启动并将它们读取的内容转储到队列中。当cat 处于活动状态时，主线程从这个队列中读取。

with CancelFunction(p.kill):
    try:
      stdout_thread = threading.Thread(target=StdOutThread)
      stdout_thread.start()
      while p.poll() is None:
        ReadFromQueue()
      while not queue.empty():
        ReadFromQueue()  
    finally:
      running = False
      stdout_thread.join()

我考虑过使用 pexpect 来解决这个问题，但同时也想区分 pexpect 似乎不可能的 stdout 和 stderr。非常感谢您的帮助。

【问题讨论】：

标签： python multithreading multiprocessing cat

【解决方案1】：

我确定您的主线程在 cat 的所有输出被读取并放入队列之前退出了 try 块。

请注意，cat 可以退出，即使您尚未阅读其所有输出。考虑一下这一系列事件：

cat 写出最后一行
cat 退出
在读取器线程更改以从cat 读取输出的最后一位之前，主线程检测到cat 已退出（通过p.poll()）
然后主线程退出try块并将running设置为false
阅读器线程退出，因为running 为假，但在此之前已读取最后一个输入。

下面是一个更简单的方法，它使用队列中的 sentinel 值通知主线程读取器线程已退出。

如果cat 退出，那么最终它将到达管道上的 EOF 监控。当这种情况发生时，它会将 None 放入队列通知主线程它完成了。当两个读者线程都有完成主线程可以安全地停止监视队列并加入线程。

import threading
import subprocess
import os
import time
import Queue
import sys

def pipe_thread(queue, name, handle):
  print "in handlehandle"
  for line in handle:
    if line[-1] == "\n":
      line = line[:-1]
    queue.put( (name, line) )
  queue.put(None)

def main():
    p = subprocess.Popen("cat file.txt",
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         stdin=subprocess.PIPE,
                         shell=True,
                         bufsize=0)

    queue = Queue.Queue()

    t1 = threading.Thread(target = pipe_thread,
                             args = [queue, "stdout", p.stdout])
    t2 = threading.Thread(target = pipe_thread,
                             args = [queue, "stderr", p.stderr])

    t1.start()
    t2.start()

    alive = 2
    count = 0
    while alive > 0:
      item = queue.get()
      if item == None:
        alive = alive - 1
      else:
        (which, line) = item
        count += 1
        print count, "got from", which, ":", line
    print "joining..."
    t1.join()
    t2.join()

main()

【讨论】：