Чтение вывода из дочернего процесса с использованием Python

Контекст

Я используюsubprocess Модуль для запуска процесса из Python. Я хочу иметь доступ к выводу (stdout, stderr), как только он записан / буферизован.

Решение должно поддерживать Windows 7. Мне также нужно решение для систем Unix, но я подозреваю, что дело с Windows сложнее решить.Решение должно поддерживать Python 2.6. В настоящее время я ограничен Python 2.6, но решения, использующие более поздние версии Python, все еще ценятся.Решение не должно использовать сторонние библиотеки. В идеале мне бы понравилось решение с использованием стандартной библиотеки, но я открыт для предложений.Решение должно работать практически для любого процесса. Предположим, что нет никакого контроля над процессом, который выполняется.Детский процесс

Например, представьте, что я хочу запустить файл Python с именемcounter.py черезsubprocess, Содержаниеcounter.py как следует:

import sys

for index in range(10):

    # Write data to standard out.
    sys.stdout.write(str(index))

    # Push buffered data to disk.
    sys.stdout.flush()
Родительский процесс

Родительский процесс, ответственный за выполнениеcounter.py Пример таков:

import subprocess

command = ['python', 'counter.py']

process = subprocess.Popen(
    cmd,
    bufsize=1,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    ) 
Проблема

С использованиемcounter.py Например, я могу получить доступ к данным до завершения процесса. Это замечательно! Это именно то, что я хочу. Тем не менее, удалениеsys.stdout.flush() вызов предотвращает доступ к данным в то время, когда я этого хочу. Это плохо! Это именно то, что я нене хочу Я понимаю, чтоflush() вызов заставляет данные быть записанными на диск, и прежде чем данные будут записаны на диск, они существуют только в буфере. Помните, я хочу иметь возможность запускать практически любой процесс. Я не ожидаю, что процесс выполнит этот тип очистки, но я все еще ожидаю, что данные будут доступны в режиме реального времени (или близко к нему). Есть ли способ добиться этого?

Краткое примечание о родительском процессе. Вы можете заметить, что я используюbufsize=0 для буферизации строк. Я надеялся, что это приведет к сбросу диска для каждой строки, но это неКажется, так не работает. Как работает этот аргумент?

Вы также заметите, что я используюsubprocess.PIPE, Это потому, что это единственное значение, которое создает объекты ввода-вывода между родительским и дочерним процессами. Я пришел к такому выводу, посмотрев наPopen._get_handles метод вsubprocess модуль (ям, ссылаясь на определение Windows здесь). Есть две важные переменные,c2pread а такжеc2pwrite которые установлены на основеstdout значение передаетсяPopen конструктор. Например, еслиstdout не установлен, тоc2pread переменная не установлена. Это также имеет место при использовании файловых дескрипторов и файловых объектов. Я неНа самом деле я не знаю, важно это или нет, но мой инстинкт инстинкта подсказывает мне, что я бы хотел и читать, и писать объекты ввода-вывода для того, чего я пытаюсь достичь - вот почему я выбралsubprocess.PIPE, Буду очень признателен, если кто-нибудь сможет объяснить это более подробно. Точно так же, если есть веская причина использовать что-то кромеsubprocess.PIPE Я весь внимание.

Метод получения данных из дочернего процесса
import time
import subprocess
import threading
import Queue


class StreamReader(threading.Thread):
    """
    Threaded object used for reading process output stream (stdout, stderr).   
    """

    def __init__(self, stream, queue, *args, **kwargs):
        super(StreamReader, self).__init__(*args, **kwargs)
        self._stream = stream
        self._queue = queue

        # Event used to terminate thread. This way we will have a chance to 
        # tie up loose ends. 
        self._stop = threading.Event()

    def stop(self):
        """
        Stop thread. Call this function to terminate the thread. 
        """
        self._stop.set()

    def stopped(self):
        """
        Check whether the thread has been terminated.
        """
        return self._stop.isSet()

    def run(self):
        while True:
            # Flush buffered data (not sure this actually works?)
            self._stream.flush()

            # Read available data.
            for line in iter(self._stream.readline, b''):
                self._queue.put(line)

            # Breather.
            time.sleep(0.25)

            # Check whether thread has been terminated.
            if self.stopped():
                break


cmd = ['python', 'counter.py']

process = subprocess.Popen(
    cmd,
    bufsize=1,
    stdout=subprocess.PIPE,
    )

stdout_queue = Queue.Queue()
stdout_reader = StreamReader(process.stdout, stdout_queue)
stdout_reader.daemon = True
stdout_reader.start()

# Read standard out of the child process whilst it is active.  
while True:

    # Attempt to read available data.  
    try:
        line = stdout_queue.get(timeout=0.1)
        print '%s' % line

    # If data was not read within time out period. Continue. 
    except Queue.Empty:
        # No data currently available.
        pass

    # Check whether child process is still active.
    if process.poll() != None:

        # Process is no longer active.
        break

# Process is no longer active. Nothing more to read. Stop reader thread.
stdout_reader.stop()

Здесь я выполняю логику, которая читает поток из дочернего процесса в потоке. Это учитывает сценарий, в котором чтение блокируется, пока данные не будут доступны. Вместо того, чтобы ждать некоторый потенциально длительный период времени, мы проверяем, есть ли доступные данные, которые нужно прочитать в течение периода ожидания, и продолжаем цикл, если их нет.

Я также попробовал другой подход, использующий вид неблокирующего чтения. Этот подход используетctypes модуль для доступа к системным вызовам Windows. Обратите внимание, что я неЯ не совсем понимаю, что я здесь делаю - я просто попытался разобраться в примере кода, который я видел в других постах. В любом случае следующий фрагмент нерешить проблему буферизации. Я понимаю, что этоЭто просто еще один способ борьбы с потенциально долгим временем чтения.

import os
import subprocess

import ctypes
import ctypes.wintypes
import msvcrt

cmd = ['python', 'counter.py']

process = subprocess.Popen(
    cmd,
    bufsize=1,
    stdout=subprocess.PIPE,
    )


def read_output_non_blocking(stream):
    data = ''
    available_bytes = 0

    c_read = ctypes.c_ulong()
    c_available = ctypes.c_ulong()
    c_message = ctypes.c_ulong()

    fileno = stream.fileno()
    handle = msvcrt.get_osfhandle(fileno)

    # Read available data.
    buffer_ = None
    bytes_ = 0
    status = ctypes.windll.kernel32.PeekNamedPipe(
        handle,
        buffer_,
        bytes_,
        ctypes.byref(c_read),
        ctypes.byref(c_available),
        ctypes.byref(c_message),
        )

    if status:
        available_bytes = int(c_available.value)

    if available_bytes > 0:
        data = os.read(fileno, available_bytes)
        print data

    return data

while True:

    # Read standard out for child process.
    stdout = read_output_non_blocking(process.stdout)
    print stdout

    # Check whether child process is still active.
    if process.poll() != None:

        # Process is no longer active.
        break

Комментарии очень ценятся.

CheersI»

Ответы на вопрос(2)

Ваш ответ на вопрос