树莓派服务器软件Stdin控制问题求助：PIPE与tempfile均失效

问题描述 #

在树莓派上开发控制服务器软件时，无法为带控制台的受控进程（如Minecraft服务器）提供实时Stdin输入。尝试使用PIPE和tempfile均失败，用测试脚本test.py（代码为print(input())）测试时，出现EOFError: EOF while reading报错，仅tempfile处理stdout和stderr时工作正常。

原实现代码如下：

import shlex
import subprocess
import tempfile
import time

class Process():
    def __init__(self, run_command) -> None:
        self.command = run_command
        self._cmd_list = shlex.split(self.command)

    def start(self):
        self._stdout = tempfile.NamedTemporaryFile("w+", encoding="utf+8")
        self._stdin = tempfile.NamedTemporaryFile("a+", encoding="utf+8")
        self._p = subprocess.Popen(self._cmd_list, stdout=self._stdout, stderr=self._stdout, stdin=self._stdin, text=True)

    def get_output(self):
        return self._stdout.read()
    
    def send_input(self, char, newline=True):
        self._stdin.write(char)
        if newline:
            self._stdin.write("\n")
        self._stdin.flush()
    
    def end(self):
        self._p.kill()
        self._stdout.seek(0)
        self.out = self._stdout.read()
        self._stdout.close()
        return self.out
    

if __name__ == "__main__":
    # run tests
    p = Process("python3 test.py")
    p.start()
    time.sleep(0.5)
    p.send_input(input())
    time.sleep(0.5)
    print(p.end())

其中test.py内容为：

print(input())

问题原因 #

• 使用tempfile作为子进程stdin时，子进程启动后会立即读取到文件末尾，后续写入文件的内容子进程不会再读取，导致子进程判定输入已结束，触发EOFError。
• 若之前尝试PIPE失败，通常是因为未正确处理管道缓冲或未保持管道打开状态，导致子进程提前接收到EOF信号。

解决方案 #

改用subprocess.PIPE直接关联子进程的stdin，同时通过线程处理stdout和stderr的读取（避免阻塞主进程），确保输入能实时传递给子进程。

修改后的代码如下：

import shlex
import subprocess
import threading
import time
from queue import Queue, Empty

class Process():
    def __init__(self, run_command) -> None:
        self.command = run_command
        self._cmd_list = shlex.split(self.command)
        self._output_queue = Queue()
        self._p = None

    def _read_output(self, stream):
        """从流中读取内容并放入队列"""
        for line in iter(stream.readline, ''):
            self._output_queue.put(line)
        stream.close()

    def start(self):
        # 启动子进程，使用PIPE关联stdin/stdout/stderr
        self._p = subprocess.Popen(
            self._cmd_list,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,  # 将stderr合并到stdout
            stdin=subprocess.PIPE,
            text=True,
            bufsize=1,  # 行缓冲，确保实时输出
            universal_newlines=True
        )
        # 启动线程读取输出，避免阻塞主进程
        threading.Thread(target=self._read_output, args=(self._p.stdout,), daemon=True).start()

    def get_output(self):
        """获取所有未读取的输出"""
        output = []
        while True:
            try:
                line = self._output_queue.get_nowait()
                output.append(line)
            except Empty:
                break
        return ''.join(output)
    
    def send_input(self, char, newline=True):
        """向子进程发送输入"""
        if self._p and self._p.stdin:
            input_str = char + ("\n" if newline else "")
            self._p.stdin.write(input_str)
            self._p.stdin.flush()
    
    def end(self):
        """终止进程并返回所有输出"""
        if self._p:
            self._p.stdin.close()  # 关闭输入，让子进程正常结束（可选，也可直接kill）
            self._p.wait()
            # 读取剩余输出
            final_output = self.get_output()
            return final_output
        return ""
    

if __name__ == "__main__":
    # 测试代码
    p = Process("python3 test.py")
    p.start()
    time.sleep(0.5)
    user_input = input("请输入测试内容: ")
    p.send_input(user_input)
    time.sleep(0.5)
    print("子进程输出:", p.end())

关键说明 #

• 使用subprocess.PIPE直接连接子进程的stdin，确保输入能实时传递，子进程不会提前接收到EOF信号。
• 通过线程读取stdout，避免主进程因等待输出而阻塞，同时保证输出能实时获取。
• 设置bufsize=1启用行缓冲，确保子进程的输出能实时刷新到管道中，避免输出延迟。

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