Python中为何select判定可读后read仍会阻塞？

问题背景与现象 #

众所周知，在网络套接字场景中，如果select(2)判定有可读数据后，出现校验和错误或缓冲区操作导致数据丢弃，read可能会阻塞，但Python处理标准管道时本不该出现这种情况。

但在以下示例代码中，通过subprocess.Popen调用openssl speed，用select监听stdout和stderr的可读状态，select判定可读后fd.read()依然会阻塞。添加os.set_blocking设置非阻塞模式后虽有改善，但程序启动时会出现异常延迟，改用二进制流（text=False）结果一致。

示例代码：

import os
import sys
import select
from io import StringIO
import subprocess

DEBUG = True

with subprocess.Popen(["openssl", "speed"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as process:
    #  Something weird is going on. `fd.read()` below blocks even if `select` declares
    #  that there is data available. The `set_blocking` calls below partially help this,
    #  but now there is a strange delay at program start.
    os.set_blocking(process.stdout.fileno(), False)
    os.set_blocking(process.stderr.fileno(), False)

    buffer = StringIO()
    while process.poll() is None:
        rfds, _, _ = select.select([process.stdout, process.stderr], [], [])
        for fd in rfds:
            if fd == process.stdout:
                chunk = fd.read()
                if DEBUG:
                    sys.stdout.write(chunk)
                buffer.write(chunk)
            elif fd == process.stderr:
                sys.stdout.write(fd.read())

原因分析 #

1. Python高层IO对象的缓冲坑 #

你用的process.stdout/stderr不是直接的操作系统文件描述符，而是Python封装的TextIOWrapper或BufferedReader，它们自带用户态缓冲区：

• select盯着的是操作系统内核缓冲区的状态，只要内核里有数据，就会返回“可读”；
• 但Python的IO对象会先把内核里的数据读到自己的缓冲区，之后的read()是从这个用户态缓冲区读。如果select返回后，Python的缓冲区刚好是空的，它会再次尝试从内核读，但此时内核可能已经没有新数据了（或者只有一点），而无参的read()会一直等，直到读到EOF，自然就阻塞了。

2. 非阻塞模式的适配问题 #

你给内核文件描述符设了非阻塞，但Python的高层IO对象根本不知道这件事，它还是按自己的逻辑来读数据。这就导致启动时，Python的IO对象会反复尝试读取内核缓冲区，直到有数据进来，造成空转延迟。

3. read()无参的致命问题 #

无参的fd.read()默认是读到文件结束才停，而不是读当前可用的那点数据。哪怕select告诉你有数据，只要没到EOF，它就会一直等更多数据，直接阻塞。

解决方法 #

方法1：改用os.read()直接操作文件描述符 #

绕过Python的高层IO对象，直接用内核文件描述符读，每次读固定大小的字节，完全可控：

import os
import sys
import select
from io import StringIO
import subprocess

DEBUG = True

with subprocess.Popen(["openssl", "speed"], text=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as process:
    stdout_fd = process.stdout.fileno()
    stderr_fd = process.stderr.fileno()
    os.set_blocking(stdout_fd, False)
    os.set_blocking(stderr_fd, False)

    buffer = StringIO()
    while process.poll() is None:
        rfds, _, _ = select.select([stdout_fd, stderr_fd], [], [])
        for fd in rfds:
            try:
                if fd == stdout_fd:
                    chunk = os.read(fd, 4096)
                    if not chunk:
                        continue
                    chunk_str = chunk.decode('utf-8')
                    if DEBUG:
                        sys.stdout.write(chunk_str)
                    buffer.write(chunk_str)
                elif fd == stderr_fd:
                    chunk = os.read(fd, 4096)
                    if chunk:
                        sys.stdout.write(chunk.decode('utf-8'))
            except BlockingIOError:
                # 没有更多数据，直接跳过
                pass

方法2：用read1()代替read() #

Python的缓冲IO对象提供了read1()方法，它会读取当前内核缓冲区中的所有可用数据，而不是等待到EOF，完全不需要设置非阻塞：

import os
import sys
import select
from io import StringIO
import subprocess

DEBUG = True

with subprocess.Popen(["openssl", "speed"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as process:
    buffer = StringIO()
    while process.poll() is None:
        rfds, _, _ = select.select([process.stdout, process.stderr], [], [])
        for fd in rfds:
            if fd == process.stdout:
                chunk = fd.read1()
                if not chunk:
                    continue
                if DEBUG:
                    sys.stdout.write(chunk)
                buffer.write(chunk)
            elif fd == process.stderr:
                chunk = fd.read1()
                if chunk:
                    sys.stdout.write(chunk)

方法3：使用subprocess原生方法简化逻辑 #

如果不需要实时处理输出，直接用process.communicate()一次性读取所有输出；如果需要实时处理，可以用迭代器逐行读取：

import sys
import subprocess

DEBUG = True

process = subprocess.Popen(["openssl", "speed"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
buffer = []
# 处理stdout
for line in process.stdout:
    if DEBUG:
        sys.stdout.write(line)
    buffer.append(line)
# 处理stderr
for line in process.stderr:
    sys.stdout.write(line)
process.wait()

内容的提问来源于stack exchange，提问作者Benjamin Crawford Ctrl-Alt-Tut