Python：select() 不会发出来自管道的所有输入信号

Question

我正在尝试使用 Python 加载外部命令行程序并通过管道与其进行通信。该程序通过标准输入获取文本输入，并按行生成文本输出到标准输出。通信应该使用 select() 进行异步。

问题是，并非程序的所有输出都在 select() 中发出信号。通常最后一两行没有信号。如果 select() 返回超时，并且我尝试从管道读取，无论如何 readline() 会立即返回从程序发送的行。请参阅下面的代码。

该程序不缓冲输出并以文本行发送所有输出。到目前为止，通过许多其他语言和环境中的管道连接到程序都运行良好。

我在 Mac OSX 10.6 上尝试过 Python 3.1 和 3.2。

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)

Answer 1

请注意，file.readlines([size]) 在内部多次循环并调用 read() 系统调用，尝试填充 size 的内部缓冲区代码>.第一次调用 read() 将立即返回，因为 select() 指示 fd 可读。然而，第二次调用将阻塞，直到数据可用，这违背了使用 select 的目的。无论如何，在异步应用程序中使用 file.readlines([size]) 都是很棘手的。

每次通过 select 时，您应该在每个 fd 上调用一次 os.read(fd, size) 。这将执行非阻塞读取，并允许您缓冲部分行，直到数据可用并明确检测到 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