Python子进程高级交互：非阻塞I/O、超时控制与输出捕获

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

class Runner:
    def __init__(self, command: list, stdin_input: str = ""):
        """
        初始化Runner，启动子进程并提供初始stdin。
        :param command: 包含程序及其参数的列表，例如 ["python", "x.py"]。
        :param stdin_input: 要一次性发送给子进程的初始stdin字符串。
        """
        self.process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            # bufsize=-1 表示使用系统默认缓冲，通常是最佳选择
            # close_fds=True 在Unix上通常是推荐做法，避免文件描述符泄露，
            # 但如果需要在子进程中继承特定的FD，则设为False。
            # 这里我们直接操作FD，所以设为False以确保FD可用。
            close_fds=False,
        )
        if stdin_input:
            # 写入初始stdin，并确保以换行符结束
            self.process.stdin.write(stdin_input.encode() + b"\n")
            self.process.stdin.flush() # 确保数据立即发送

    # ... 其他方法 ...

    # ... __init__ 方法 ...

    def _enqueue_output(self, out_stream: IO[bytes], queue: Queue[bytes]):
        """
        在单独的线程中运行，从给定的流中读取数据并放入队列。
        """
        # 使用io.open从文件描述符创建二进制流，closefd=False表示不关闭原始FD
        stream = io.open(out_stream.fileno(), "rb", closefd=False)
        while True:
            # read1() 是非阻塞的，有数据就返回，无数据返回空字节串
            n = stream.read1(4096) # 尝试读取最多4KB数据
            if len(n) > 0:
                queue.put(n)
            else:
                # 当子进程关闭其输出流时，read1()会返回空字节串，此时退出循环
                break

    def start_reader_thread(self, out_stream: IO[bytes], queue: Queue[bytes]) -> Thread:
        """
        为给定的输出流启动一个守护线程，用于异步读取数据。
        :param out_stream: 子进程的stdout或stderr文件对象。
        :param queue: 用于存储读取数据的队列。
        :return: 启动的线程对象。
        """
        t = Thread(target=self._enqueue_output, args=(out_stream, queue))
        t.daemon = True  # 将线程设置为守护线程，主程序退出时自动终止
        t.start()
        return t

    # ... run 方法 ...

    # ... start_reader_thread 方法 ...

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

        # 启动stdout和stderr的读取线程
        stdout_thread = self.start_reader_thread(self.process.stdout, stdout_queue)
        stderr_thread = self.start_reader_thread(self.process.stderr, stderr_queue)

        try:
            # communicate() 会等待子进程终止，并关闭stdin。
            # 如果设置了timeout，它会在指定时间后抛出TimeoutExpired异常。
            # 注意：如果子进程在timeout之前完成，communicate会正常返回。
            # communicate() 返回 (stdout_data, stderr_data)，但我们通过队列收集，
            # 所以这里可以忽略其返回值，主要利用其等待和超时功能。
            self.process.communicate(timeout=timeout)
        except subprocess.TimeoutExpired:
            print(f"WARN: 子进程在 {timeout} 秒后超时，尝试终止。")
            self.process.kill() # 超时后强制终止子进程
            # 再次communicate以确保所有管道被关闭并清理资源
            self.process.communicate()
        except Exception as e:
            print(f"ERROR: 运行子进程时发生异常: {e}")
        finally:
            # 确保在子进程终止后，读取线程有机会处理完剩余数据
            # 给予短暂时间让守护线程完成其队列填充，虽然守护线程不会被join
            time.sleep(0.1)

            print("\n=== 标准输出 (STDOUT) ===")
            collected_stdout = []
            try:
                while True:
                    # get_nowait() 非阻塞地从队列中获取数据
                    collected_stdout.append(stdout_queue.get_nowait().decode(errors='ignore'))
            except Empty:
                pass # 队列为空，退出循环
            print("".join(collected_stdout), end="")
            print("=== STDOUT 结束 ===\n")

            print("=== 标准错误 (STDERR) ===")
            collected_stderr = []
            try:
                while True:
                    collected_stderr.append(stderr_queue.get_nowait().decode(errors='ignore'))
            except Empty:
                pass
            print("".join(collected_stderr), end="")
            print("=== STDERR 结束 ===\n")

            # 确保进程已完全终止
            if self.process.poll() is None:
                print("WARN: 子进程在收集输出后仍未终止，尝试kill。")
                self.process.kill()
                self.process.wait() # 等待进程彻底结束

# x.py
import sys
import time

print("Hello from x.py!")
sys.stdout.flush() # 确保立即输出

name = input("Please enter your name: ")
print(f"Nice to meet you, {name}!")

age = input("Please enter your age: ")
try:
    age_int = int(age)
    print(f"You are {age_int} years old.")
except ValueError:
    print(f"'{age}' is not a valid age.", file=sys.stderr)

# 模拟长时间运行
time.sleep(2)
print("x.py is done.")

# main.py
# 导入Runner类（假设Runner类和x.py在同一目录下）
# from your_module import Runner # 如果Runner在一个模块中
# 或者直接将Runner类定义在此文件中

# 确保Runner类已经定义在当前作用域

# 示例 1: 正常运行并提供输入
print("--- 示例 1: 正常运行并提供输入 ---")
runner1 = Runner(["python", "x.py"], stdin_input="Alice\n30")
runner1.run(timeout=10) # 给予足够的时间完成

print("\n--- 示例 2: 子进程超时 ---")
# 示例 2: 子进程超时 (x.py有sleep(2)，我们设置timeout=1)
runner2 = Runner(["python", "x.py"], stdin_input="Bob\n25")
runner2.run(timeout=1) # 会在sleep(2)完成前超时

print("\n--- 示例 3: 错误输入导致stderr ---")
runner3 = Runner(["python", "x.py"], stdin_input="Charlie\nnot_an_age")
runner3.run(timeout=5)

--- 示例 1: 正常运行并提供输入 ---

=== 标准输出 (STDOUT) ===
Hello from x.py!
Please enter your name: Nice to meet you, Alice!
Please enter your age: You are 30 years old.
x.py is done.
=== STDOUT 结束 ===

=== 标准错误 (STDERR) ===

=== STDERR 结束 ===


--- 示例 2: 子进程超时 ---
WARN: 子进程在 1.0 秒后超时，尝试终止。

=== 标准输出 (STDOUT) ===
Hello from x.py!
Please enter your name: Nice to meet you, Bob!
Please enter your age: === STDOUT 结束 ===

=== 标准错误 (STDERR) ===

=== STDERR 结束 ===


--- 示例 3: 错误输入导致stderr ---

=== 标准输出 (STDOUT) ===
Hello from x.py!
Please enter your name: Nice to meet you, Charlie!
Please enter your age: === STDOUT 结束 ===

=== 标准错误 (STDERR) ===
'not_an_age' is not a valid age.
=== STDERR 结束 ===