Асинхронный подпроцесс в Windows

Насколько плохо использовать потоки? Я столкнулся с той же проблемой и в конце концов решил использовать потоки для сбора всех данных в stdout и stderr подпроцесса и поместить их в потокобезопасную очередь, которую основной поток может читать блокирующим образом, без необходимости беспокоиться о потоках, происходящих за кулисами.

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

Моя функция потока выглядела примерно так:

def simple_io_thread(pipe, queue, tag, stop_event):
    """
    Read line-by-line from pipe, writing (tag, line) to the
    queue. Also checks for a stop_event to give up before
    the end of the stream.
    """
    while True:
        line = pipe.readline()

        while True:
            try:
                # Post to the queue with a large timeout in case the
                # queue is full.
                queue.put((tag, line), block=True, timeout=60)
                break
            except Queue.Full:
                if stop_event.isSet():
                    break
                continue
        if stop_event.isSet() or line=="":
            break
    pipe.close()

Когда я запускаю подпроцесс, я делаю это:

outputqueue = Queue.Queue(50)
stop_event = threading.Event()
process = subprocess.Popen(
    command,
    cwd=workingdir,
    env=env,
    shell=useshell,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE)
stderr_thread = threading.Thread(
    target=simple_io_thread,
    args=(process.stderr, outputqueue, "STDERR", stop_event)
)
stdout_thread = threading.Thread(
    target=simple_io_thread,
    args=(process.stdout, outputqueue, "STDOUT", stop_event)
)
stderr_thread.daemon = True
stdout_thread.daemon = True
stderr_thread.start()
stdout_thread.start()

Затем, когда я хочу прочитать, я могу просто заблокировать outputqueue - каждый считанный из него элемент содержит либо строку, определяющую, из какого канала он пришел, либо строку текста из этого канала. Очень мало кода выполняется в отдельном потоке, и он взаимодействует с основным потоком только через потокобезопасную очередь (плюс событие на случай, если мне нужно сдаться раньше). Возможно, такой подход был бы полезен и позволил бы решить проблему с потоками и блокировками, но без переписывания большого количества кода?

(Мое решение усложнено, потому что иногда я хочу завершить подпроцессы досрочно и хочу быть уверенным, что все потоки закончатся. Если это не проблема, вы можете избавиться от всех вещей stop_event, и это станет довольно кратким.)

Я предполагаю, что конвейер процесса не заблокируется, если он использует только стандартный ввод и стандартный вывод; и проблема, которую вы пытаетесь решить, заключается в том, как сделать так, чтобы она не блокировалась, если они пишут в stderr (и должны иметь дело с возможным резервным копированием stderr).

Если вы позволяете нескольким процессам писать в stderr, вы должны следить за тем, чтобы их выходные данные не смешивались. Я предполагаю, что вы как-то разобрались с этим; просто поместите это там, чтобы быть уверенным.

Помните о флаге -u для python; это полезно при тестировании, чтобы увидеть, не портит ли вас буферизация ОС.

Если вы хотите эмулировать select() для файловых дескрипторов в win32, ваш единственный выбор — использовать PeekNamedPipe() и подобные. У меня есть фрагмент кода, который одновременно считывает построчный вывод из нескольких процессов, который вы даже можете использовать напрямую — попробуйте передать ему список дескрипторов proc.stderr и вперед.

class NoLineError(Exception): pass
class NoMoreLineError(Exception): pass
class LineReader(object):
    """Helper class for multi_readlines."""
    def __init__(self, f):
        self.fd = f.fileno()
        self.osf = msvcrt.get_osfhandle(self.fd)
        self.buf = ''

    def getline(self):
        """Returns a line of text, or raises NoLineError, or NoMoreLineError."""
        try:
            _, avail, _ = win32pipe.PeekNamedPipe(self.osf, 0)
            bClosed = False
        except pywintypes.error:
            avail = 0
            bClosed = True

        if avail:
            self.buf += os.read(self.fd, avail)

        idx = self.buf.find('\n')
        if idx >= 0:
            ret, self.buf = self.buf[:idx+1], self.buf[idx+1:]
            return ret
        elif bClosed:
            if self.buf:
                ret, self.buf = self.buf, None
                return ret
            else:
                raise NoMoreLineError
        else:
            raise NoLineError


def multi_readlines(fs, timeout=0):
    """Read lines from |fs|, a list of file objects.
    The lines come out in arbitrary order, depending on which files
    have output available first."""
    if type(fs) not in (list, tuple):
        raise Exception("argument must be a list.")
    objs = [LineReader(f) for f in fs]
    for i,obj in enumerate(objs): obj._index = i
    while objs:
        yielded = 0
        for i,obj in enumerate(objs):
            try:
                yield (obj._index, obj.getline())
                yielded += 1
            except NoLineError:
                #time.sleep(timeout)
                pass
            except NoMoreLineError:
                del objs[i]
                break   # Because we mutated the array

        if not yielded:
            time.sleep(timeout)
            pass

Я никогда не видел проблему «Peek возвращает 0 байтов, даже если данные доступны». Если это произойдет с другими, держу пари, их libc буферизует их stdout/stderr перед отправкой данных в ОС; вы ничего не можете сделать с этим извне. Вы должны каким-то образом заставить приложение использовать небуферизованный вывод (-u для python; вызовы win32/libc для изменения дескриптора файла stderr,...)

Тот факт, что вы ничего не видите, а затем тонну обновлений, заставляет меня думать, что ваша проблема заключается в буферизации на стороне источника. win32 libc может буферизоваться по-разному, если пишет в канал, а не в консоль. Опять же, лучшее, что вы можете сделать вне этих программ, — это агрессивно истощать их вывод.

Как насчет использования Twisted FD? http://twistedmatrix.com/documents/8.1.0/api/twisted.internet.fdesc.html

Он не асинхронный, но неблокирующий. Для асинхронных вещей вы можете использовать Twisted?