プロセス出力のバッファリングによる切り捨てが発生しましたか？

Question

私はcatでサブプロセスを使って読んでいる100行のサンプルファイルを持っています。ただし、キュー内の出力は常に切り捨てられます。私はそれがパイプを検出するので、その出力をバッファリングするcatのためかもしれないと思う。

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)) 

これらのスレッドが開始され、それらがキューに読んだダンプ：

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

私は catのstdoutとstderrパイプから読み取るために別のスレッドを使用しています。 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を区別したいしています。ヘルプは非常に高く評価されます。

Answer 1

catからのすべての出力が読み込まれ、キューに置かれる前に、メインスレッドがtryブロックを終了していることを確認します。

catは、すべての出力を読み取っていない場合でも終了できます。この一連のイベントを考えてみましょう ：リーダースレッドがメインスレッドがcatを持っていることを検出しcatからの出力の最後のビットを読み込むための変更を持って前に

1. catはその最後の行から
2. cat終了
3. を書き込み（p.poll()経由）
4. メインスレッドがtryブロックを終了し、runningをfalseに設定します。
5. リーダースレッドe runningはfalseですが、その前に 最後の入力が読み取られているので、xit。以下

リーダースレッドが終了したことをメインスレッドに通知するキュー にセンチネル値を使用して簡単な方法です。

catが終了すると、最終的にパイプ上のEOFに到達します。これは、 のモニタリングです。そしてそれが起こるとき、それは完了しているメインスレッドに通知するために、キューに を置きます。両方のリーダースレッドで が完了すると、メインスレッドはキューの監視を安全に停止し、 スレッドに参加できます。

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()