Python: select() не сигнализирует все входные данные из канала

Я пытаюсь загрузить внешнюю программу командной строки с помощью Python и связываться с ней через каналы. Программа принимает текстовый ввод через stdin и производит вывод текста в строках в stdout. Коммуникация должна быть асинхронной, используя select().

Проблема заключается в том, что не весь вывод программы сигнализируется в select(). Обычно одна или две строки не сигнализируются. Если select() возвращается с таймаутом, и я пытаюсь читать из канала, в любом случае readline() немедленно возвращает строку, отправленную из программы. См. Код ниже.

Программа не буферизует вывод и отправляет весь вывод в текстовые строки. Подключение к программе через каналы во многих других языках и средах пока прекратилось.

Я пробовал Python 3.1 и 3.2 на Mac OSX 10.6.

import subprocess
import select

engine = subprocess.Popen("Engine", bufsize=0, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
engine.stdin.write(b"go\n")
engine.stdin.flush()

while True:
    inputready,outputready,exceptready = select.select( [engine.stdout.fileno()] , [], [], 10.0)

    if (inputready, outputready, exceptready) == ([], [], []):
        print("trying to read from engine anyway...")
        line = engine.stdout.readline()
        print(line)

     for s in inputready:
        line = engine.stdout.readline()
        print(line)

Ответы

Ответ 1

Обратите внимание, что внутри file.readlines([size]) выполняется цикл и вызывает одноименный вызов read() более одного раза, пытаясь заполнить внутренний буфер size. Первый вызов read() будет немедленно возвращен, так как select() указал, что fd читается. Однако 2-й вызов блокируется до тех пор, пока не будут доступны данные, что приведет к победе над целью использования выбора. В любом случае сложно использовать file.readlines([size]) в асинхронном приложении.

Вы должны называть os.read(fd, size) один раз на каждом fd для каждого прохода через select. Это выполняет неблокирующее чтение и позволяет вам частично или частично буферизировать данные до тех пор, пока данные не будут доступны и не обнаружит EOF однозначно.

Я изменил ваш код, чтобы проиллюстрировать его с помощью os.read. Он также считывает из процесса 'stderr:

import os
import select
import subprocess
from cStringIO import StringIO

target = 'Engine'
PIPE = subprocess.PIPE
engine = subprocess.Popen(target, bufsize=0, stdin=PIPE, stdout=PIPE, stderr=PIPE)
engine.stdin.write(b"go\n")
engine.stdin.flush()

class LineReader(object):

    def __init__(self, fd):
        self._fd = fd
        self._buf = ''

    def fileno(self):
        return self._fd

    def readlines(self):
        data = os.read(self._fd, 4096)
        if not data:
            # EOF
            return None
        self._buf += data
        if '\n' not in data:
            return []
        tmp = self._buf.split('\n')
        lines, self._buf = tmp[:-1], tmp[-1]
        return lines

proc_stdout = LineReader(engine.stdout.fileno())
proc_stderr = LineReader(engine.stderr.fileno())
readable = [proc_stdout, proc_stderr]

while readable:
    ready = select.select(readable, [], [], 10.0)[0]
    if not ready:
        continue
    for stream in ready:
        lines = stream.readlines()
        if lines is None:
            # got EOF on this stream
            readable.remove(stream)
            continue
        for line in lines:
            print line