Python子进程的非阻塞I/O与生命周期管理

# x.py
print("hi")
input("Your name: ")

import subprocess
from threading import Thread

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

    def run_process_blocking(self):
        # 此方法会阻塞，直到进程结束
        self.process.wait()

    def poll_stdout(self):
        # readline() 在没有换行符时会阻塞
        print("got stdout:", self.process.stdout.readline().decode(), end="")

    def give_input(self, text=""):
        self.process.stdin.write(bytes(text, 'utf-8'))
        self.process.stdin.flush() # 确保输入被发送

    def kill_process(self):
        self.process.kill()

# 示例运行（会遇到阻塞问题）
# r = InitialRunner()
# t = Thread(target=r.run_process_blocking)
# t.start()
# r.poll_stdout() # 第一次会打印 "hi"
# r.poll_stdout() # 第二次会阻塞，因为 "Your name: " 后没有换行符
# r.give_input("hi\n")
# r.kill_process()
# t.join()

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

class Runner:
    def __init__(self, stdin_input: str = ""):
        # 使用列表形式传递命令，避免shell注入问题，或者直接传递字符串并设置shell=True
        # 这里为了兼容x.py的示例，假设py是可执行的python命令别名
        self.process = subprocess.Popen(
            ["python", "x.py"], # 假设x.py在当前目录，或者使用完整路径
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=1, # 设置较小的缓冲区，有助于更频繁地刷新
            close_fds=False, # Windows系统下可能需要
            text=False # 确保stdin/stdout/stderr以字节流处理
        )
        # 立即写入所有预设的stdin输入
        if stdin_input:
            self.process.stdin.write(stdin_input.encode('utf-8'))
            self.process.stdin.flush() # 确保数据被发送
            # 如果进程可能立即退出，并且不再需要stdin，可以关闭它
            # self.process.stdin.close() 

    def _enqueue_output(self, out: IO[bytes], queue: Queue[bytes]):
        """在单独线程中读取输出流并放入队列"""
        # io.open(out.fileno(), "rb", closefd=False) 是为了在不同线程中安全地访问管道
        # out 是 bytes 类型的流，所以直接使用 out.fileno()
        stream = io.open(out.fileno(), "rb", closefd=False)
        while True:
            # read1() 会读取尽可能多的数据，但不会阻塞等待更多数据
            n = stream.read1()
            if len(n) > 0:
                queue.put(n)
            else:
                # 如果 read1() 返回空字节串，表示管道已关闭
                break

    def run(self, timeout=5):
        """
        运行子进程，并在超时后收集所有输出。
        注意：此方法不提供实时交互式输入或周期性轮询输出。
        """
        stdout_queue: Queue[bytes] = Queue()
        stderr_queue: Queue[bytes] = Queue()

        # 启动守护线程来异步读取stdout和stderr
        stdout_reader_thread = Thread(target=self._enqueue_output, args=(self.process.stdout, stdout_queue))
        stderr_reader_thread = Thread(target=self._enqueue_output, args=(self.process.stderr, stderr_queue))
        stdout_reader_thread.daemon = True
        stderr_reader_thread.daemon = True
        stdout_reader_thread.start()
        stderr_reader_thread.start()

        try:
            # 等待进程结束或达到超时
            # communicate() 会关闭 stdin，并等待进程结束
            # 如果有 timeout，它会在 timeout 后抛出 TimeoutExpired 异常
            self.process.communicate(timeout=timeout)
        except subprocess.TimeoutExpired:
            print(f"进程运行超时 ({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 结束 ===\n")

            # 收集所有缓冲的stderr
            print("=== STDERR ===")
            try:
                while True:
                    print(stderr_queue.get_nowait().decode('utf-8'), end="")
            except Empty:
                pass # 队列为空
            print("\n=== STDERR 结束 ===\n")

            # 等待读取线程结束，尽管它们是守护线程，但为了确保所有数据都被读取，
            # 可以在这里稍微等待一下，或者依赖 communicate 后的队列清空逻辑
            # 但由于 communicate 会等待进程结束，通常此时管道也已关闭，
            # 守护线程会自行终止。

# 示例：运行 x.py，并提供输入 "MyName\n"
# x.py 内容:
# print("hi")
# name = input("Your name: ")
# print(f"Hello, {name}!")

print("--- 运行示例 1: 正常输入 ---")
r1 = Runner(stdin_input="MyName\n")
r1.run(timeout=5)

# 示例：运行 x.py，不提供输入，让其超时
print("\n--- 运行示例 2: 进程超时 ---")
r2 = Runner(stdin_input="") # 不提供输入，input() 会阻塞
r2.run(timeout=2) # 设置一个较短的超时

# 示例：一个没有输入的简单脚本
# temp_script.py
# import time
# print("Starting...")
# time.sleep(3)
# print("Done.")
#
# with open("temp_script.py", "w") as f:
#     f.write("import time\nprint('Starting...')\ntime.sleep(3)\nprint('Done.')\n")
#
# print("\n--- 运行示例 3: 简单脚本执行 ---")
# r3 = Runner(stdin_input="")
# r3.process.args = ["python", "temp_script.py"] # 动态修改要运行的脚本
# r3.run(timeout=5)
# import os
# os.remove("temp_script.py")

# x.py
import sys
print("hi")
sys.stdout.flush() # 确保 "hi" 立即被发送
name = input("Your name: ")
print(f"Hello, {name}!")
sys.stdout.flush() # 确保最终输出被发送