닫기 전에 파이프 (os.pipe) 플러시

Question

하위 프로세스를 시작하고 각각 stdout 및 stderr을 읽는 데 두 개의 스레드를 사용해야합니다.

다음 코드는 stdout을 고려하고있다 :

def reader(rfd): 
    while True: 
     try: 
      data = os.read(rfd, bufsize) 
     except OSError: 
      break 
     else: 
      chomp(data) 

rout, wout = os.pipe() 
tout = threading.Thread(target=reader, args=(rout,)) 
tout.start() 

subprocess.check_call(command, bufsize=bufsize, stdout=wout, stderr=werr) 

os.close(wout) 
os.close(rout) 
tout.join() 

코드는 작동, 나는 os.close(wout) 기능은 모든 데이터를 읽어 전에 독자 를 죽이는 것처럼 모든 데이터가 처리되는 것으로 나타났습니다 제외. 반면에, 내가 닫지 않으면 wout 나의 과정은 영원히 tout.join()에 걸릴 것입니다.

내가 매우 나쁜 time.sleep(0.1)을 넣었을 때 subprocess.check_call(...)이 마술처럼 작동하기 때문에 이것은 버퍼링 문제라고 말할 수 있습니다.

기다리는 대신 플러시가 좋지만 파이프를 통해 os.fsync()을 호출하면 OSError: [Errno 22] Invalid argument이 표시됩니다.

os.pipe으로 생성 된 파이프를 플러시하는 방법에 대한 힌트가 있습니까?

Answer 1

내가 간 통신을위한 os.pipe보다는 Popen를 사용하는 것이 좋습니다 :

다음은 자체 포함 된 예입니다.

예 : 당신이 정말로 os.pipe을 사용하려는 경우

writer_process = subprocess.Popen(cmd, stdout=subprocess.PIPE) 
reader_thread = threading.Thread(target=reader, args=(writer_process.stdout,)) 
reader_thread.start() 
reader_thread.join() 

그러나, 당신은 쉽게 시간 파일 객체처럼 치료를해야합니다. 파이썬의 내장 파일 컨텍스트 관리자는 파일을 확인하고 플러시 및 닫아 적절하게 닫습니다.

예 :

def reader(fd): 
    with os.fdopen(fd, bufsize=bufsize) as f: 
     while True: 
      data = f.read(bufsize) 
      if not data: 
       break 
      chomp(data) 

및

with os.fdopen(wout, "w", bufsize=bufsize) as f: 
    subprocess.check_call(cmd, stdout=f) 

Answer 2

데이터가 손실 될 수 있기 때문에 프로세스가 완료되었다고 확신 할 때까지 파이프를 닫을 수 없으며 파이프를 닫지 않고 스레드가 완료 될 때까지 기다릴 수 없습니다 (os.read이 영원히).

프로세스 완료를 기다리고 파이프를 작성한 이후 수동으로 파이프의 쓰기 끝을 닫아야합니다.

import os 
import threading 
import subprocess 

def reader(rfd): 
    while True: 
     try: 
      data = os.read(rfd, bufsize) 
      if not data: 
       break #reached EOF 
     except OSError: 
      break 

bufsize=100 

rout, wout = os.pipe() 
rerr, werr = os.pipe() 

tout = threading.Thread(target=reader, args=(rout,)) 
tout.start() 

p= subprocess.Popen("ls", bufsize=bufsize, stdout=wout, stderr=werr) 
p.wait() #wait for the proces to end writing 
os.close(wout) 
os.close(werr) 
tout.join() 
os.close(rout) 
os.close(rerr)