通过Python subprocess向二进制程序发空输入或触发read失败的问题

你的核心需求是通过Python的subprocess与二进制程序交互时，实现两个目标：

1. 让二进制的read调用不修改目标变量（保持初始值），同时不终止程序后续交互；
2. 让read调用失败返回-1，且维持管道打开状态，程序可继续运行。

当前代码的问题在于：ps.stdin.write(b'')并未向管道写入任何数据，二进制的read(0, &num, 4)会一直阻塞等待输入，因此无法继续执行后续逻辑。

方案1：让read返回0（EOF），保持变量原值且可继续交互 #

普通管道中，read返回0意味着写端关闭，无法后续交互。改用**伪终端（PTY）**可以解决这个问题：伪终端中发送Ctrl+D（ASCII 0x04）会触发read返回0（模拟EOF），但伪终端的写端不会关闭，后续仍可继续发送数据。

修改后的Python代码 #

import pty
import os
import time

def read_all(fd, timeout=0.5):
    start = time.time()
    content = b''
    os.set_blocking(fd, False)
    while time.time() - start < timeout:
        try:
            data = os.read(fd, 1024)
            if data:
                content += data
        except BlockingIOError:
            time.sleep(0.01)
    os.set_blocking(fd, True)
    return content.decode('utf-8')

# 创建伪终端并启动二进制程序
pid, fd = pty.fork()
if pid == 0:
    os.execv('./test', ['./test'])
else:
    # 读取初始输出
    print(read_all(fd))
    # 发送Ctrl+D触发EOF，让read返回0
    os.write(fd, b'\x04')
    # 读取read后的输出
    print(read_all(fd))
    # 后续继续交互（示例）
    os.write(fd, b'\x00\x00\x00\x00')
    print(read_all(fd))

效果 #

二进制的read会返回0，num保持初始值some_random_value，输出SUCCESS，且后续仍可接收输入。

方案2：让read返回-1（读取失败），维持程序运行 #

要让read返回-1，可通过发送信号中断read调用。默认信号会终止程序，因此需要修改二进制程序忽略指定信号，让read被中断后返回-1并继续执行。

步骤1：修改二进制程序（test.c） #

添加信号处理函数，忽略SIGUSR1信号，使read被中断后返回-1：

#include<stdio.h>
#include<stdlib.h>
#include<signal.h>
#include<errno.h>

#define some_random_value 0

void sigusr1_handler(int sig) {
    // 空处理函数，仅用于中断read
}

int main()
{
        // 注册SIGUSR1信号处理
        signal(SIGUSR1, sigusr1_handler);
        int num = some_random_value;
        setvbuf(stdout, NULL, _IONBF, 0);

        printf("Communication with the python code\n");
        printf("Enter Something: \n");
        int ret = read(0, &num, 0x4);
        printf("%d bytes read, errno: %d\n", ret, errno);

        if (ret == -1)
                printf("READ FAILED\n");
        if (num == some_random_value)
                printf("SUCCESS\n");
        if (num != some_random_value)
                printf("FAILED\n");

        // 后续继续交互示例
        printf("Enter again: \n");
        ret = read(0, &num, 0x4);
        printf("%d bytes read\n", ret);
        printf("New num: %d\n", num);
        return 0;
}

步骤2：修改Python代码 #

发送SIGUSR1信号中断read调用：

import subprocess
import signal
import time
import os

def read_all(pipe, timeout=0.5):
    start = time.time()
    content = b''
    os.set_blocking(pipe.fileno(), False)
    while time.time() - start < timeout:
        try:
            line = pipe.readline()
            if line:
                content += line
        except BlockingIOError:
            time.sleep(0.01)
    os.set_blocking(pipe.fileno(), True)
    return content.decode('utf-8')

ps = subprocess.Popen('./test', stdin=subprocess.PIPE, stdout=subprocess.PIPE)
# 读取初始输出
print(read_all(ps.stdout))
# 等待二进制进入read调用
time.sleep(0.1)
# 发送SIGUSR1中断read
os.kill(ps.pid, signal.SIGUSR1)
# 读取read失败后的输出
print(read_all(ps.stdout))
# 后续继续交互
ps.stdin.write(b'\x01\x00\x00\x00')
ps.stdin.flush()
print(read_all(ps.stdout))

效果 #

read被信号中断后返回-1，errno为EINTR（4），程序输出READ FAILED和SUCCESS（因为num未被修改），且后续可继续接收输入。

内容的提问来源于stack exchange，提问作者nadav levin