Промывка трубы (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

Я бы рекомендовал использовать 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() 

Однако, если вы действительно хотите использовать os.pipe тогда у вас будет легче рассматривать их как объекты файлов. Встроенный менеджер контекстов файла Python будет уверен, что файлы будут правильно очищены и закрыты.

например.

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)