Python中交互式控制子进程：非阻塞I/O与生命周期管理

print("hi")
input("Your name: ")

import subprocess
from threading import Thread

class Runner:
    def __init__(self):
        self.process = subprocess.Popen(
            ["python", "x.py"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

    def run(self):
        self.process.wait() # 等待子进程结束

    def poll(self):
        # 尝试读取一行输出
        print("got stdout:", self.process.stdout.readline().decode(), end="")

    def give_input(self, text=""):
        # 写入输入
        return self.process.stdin.write(bytes(text, 'utf-8'))

    def kill(self):
        self.process.kill() # 终止子进程

# 示例运行
r = Runner()
t = Thread(target=r.run)
t.start()
r.poll() # 期望输出 "hi"
r.poll() # 期望输出 "Your name:"
r.give_input("hi\n")
r.kill()
t.join()

import subprocess
from queue import Queue, Empty
from threading import Thread
from typing import IO
import io

class Runner:
    def __init__(self, stdin_input: str):
        """
        初始化Runner，启动子进程并立即写入stdin。
        :param stdin_input: 要一次性写入子进程stdin的字符串。
        """
        self.process = subprocess.Popen(
            "py x.py",  # 注意：在Windows上可能需要"py"，Linux/macOS上是"python"
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=2,  # 缓冲区大小，可能影响I/O行为
            close_fds=False, # 在某些系统上可能需要，确保文件描述符不被关闭
            text=False # 明确处理字节流，避免编码问题
        )
        # 立即写入stdin，并添加换行符以确保子进程接收到完整输入
        self.process.stdin.write(stdin_input.encode('utf-8') + b"\n")
        self.process.stdin.flush() # 确保数据被发送到子进程

    def _enqueue_output(self, out: IO[bytes], queue: Queue[bytes]):
        """
        在单独的线程中执行，从给定的输出流非阻塞地读取数据并放入队列。
        :param out: 子进程的stdout或stderr流。
        :param queue: 用于存储读取数据的队列。
        """
        # 创建一个非阻塞的二进制文件流
        stream = io.open(out.fileno(), "rb", closefd=False)
        while True:
            n = stream.read1() # 非阻塞读取
            if len(n) > 0:
                queue.put(n)
            else:
                # 当流结束时，read1()会返回空字节串
                break

    def _start_reader_thread(self, out_stream: IO[bytes], queue: Queue[bytes]):
        """
        启动一个守护线程来读取子进程的输出流。
        :param out_stream: 子进程的stdout或stderr。
        :param queue: 存储输出的队列。
        """
        t = Thread(target=self._enqueue_output, args=(out_stream, queue))
        t.daemon = True  # 设置为守护线程，主程序退出时自动终止
        t.start()
        return t

    def run(self, timeout=5):
        """
        运行子进程，并在给定超时时间内收集其所有输出。
        :param timeout: 等待子进程完成的最大秒数。
        """
        stdout_queue: Queue[bytes] = Queue()
        stderr_queue: Queue[bytes] = Queue()

        # 启动读取stdout和stderr的守护线程
        self._start_reader_thread(self.process.stdout, stdout_queue)
        self._start_reader_thread(self.process.stderr, stderr_queue)

        try:
            # 使用communicate带超时等待进程结束并关闭管道
            # 注意：communicate会关闭stdin，因此不能用于后续的交互式输入
            self.process.communicate(timeout=timeout)
        except subprocess.TimeoutExpired:
            print(f"ERROR: 子进程运行超时（{timeout}秒），将被终止。")
            self.process.kill() # 超时则强制终止
            self.process.wait() # 等待子进程真正结束
        except Exception as e:
            print(f"运行子进程时发生错误: {e}")
        finally:
            # 收集并打印所有stdout
            print("\n=== STDOUT ===")
            try:
                while True:
                    print(stdout_queue.get_nowait().decode('utf-8'), end="")
            except Empty:
                pass # 队列为空
            print("\n=== STDOUT END ===\n")

            # 收集并打印所有stderr
            print("=== STDERR ===")
            try:
                while True:
                    print(stderr_queue.get_nowait().decode('utf-8'), end="")
            except Empty:
                pass # 队列为空
            print("=== STDERR END ===\n")

# 示例运行
# 假设x.py内容不变
# print("hi")
# input("Your name: ")
# 如果输入 "5\n6"
# x.py会先打印"hi"，然后等待输入"Your name: "
# 收到"5"后，input函数返回"5"，但程序会立即结束，因为没有后续逻辑
# 实际上，这里只写入一次stdin，如果子进程需要多次输入，此方法不适用
r = Runner("MyName") # 假设输入"MyName"给x.py
r.run(timeout=3)

# 另一个示例，如果x.py需要多行输入
# 假设x.py:
# print("Enter num1:")
# num1 = int(input())
# print("Enter num2:")
# num2 = int(input())
# print(f"Sum: {num1 + num2}")
# r_multi = Runner("5\n6") # 注意这里一次性写入了 "5\n6"
# r_multi.run(timeout=3)