Python subprocess模块实现外部进程的非阻塞I/O与控制

import subprocess

# 启动一个Python脚本 x.py
process = subprocess.Popen(
    ["python", "x.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True # 设置为True可以直接处理字符串，否则需要encode/decode
)

# 尝试读取输出
# output_line = process.stdout.readline() # 这会阻塞直到有换行符或EOF
# print(output_line)

# 尝试写入输入
# process.stdin.write("some input\n")
# process.stdin.flush() # 确保数据被发送

print("Hello from x.py")
name = input("Enter your name: ")
print(f"Your name is {name}")

import subprocess
from threading import Thread

class InitialRunner:
    def __init__(self, command):
        self.process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            # text=True # 方便处理字符串
        )

    def run_process_wait(self):
        """在单独线程中等待进程结束"""
        self.process.wait()

    def poll_stdout(self):
        """尝试轮询标准输出"""
        # 注意：readline() 是阻塞的，直到遇到换行符或EOF
        line = self.process.stdout.readline().decode().strip()
        if line:
            print(f"Got stdout: {line}")
        return line

    def give_input(self, text):
        """提供标准输入"""
        self.process.stdin.write(text.encode() + b"\n") # 确保发送换行符
        self.process.stdin.flush()

    def kill_process(self):
        """终止进程"""
        self.process.kill()

# 示例 x.py 内容：
# print("hi")
# name = input("Your name: ")
# print(f"Hello, {name}")

# 运行示例
# runner = InitialRunner(["python", "x.py"])
# process_thread = Thread(target=runner.run_process_wait)
# process_thread.start()

# runner.poll_stdout() # 期望输出 "hi"
# runner.poll_stdout() # 期望输出 "Your name:"
# runner.give_input("Alice")
# # ... 之后可能还有更多交互
# runner.kill_process()
# process_thread.join()

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

class AdvancedRunner:
    def __init__(self, command: list, stdin_input: str = ""):
        """
        初始化Runner，启动子进程并提供初始stdin输入。
        :param command: 启动子进程的命令列表。
        :param stdin_input: 预先提供给子进程的stdin输入字符串。
        """
        self.process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=1, # 缓冲区大小，通常设置为1或无缓冲
            close_fds=False, # 在Unix上，防止子进程继承父进程的打开文件描述符
        )

        # 立即将所有stdin输入写入管道
        if stdin_input:
            self.process.stdin.write(stdin_input.encode() + b"\n")
            self.process.stdin.flush()
        self.process.stdin.close() # 关闭stdin，表示不会再有输入

        self.stdout_queue: Queue[bytes] = Queue()
        self.stderr_queue: Queue[bytes] = Queue()

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

    def _enqueue_output(self, out: IO[bytes], queue: Queue[bytes]):
        """
        辅助函数：从指定输出流读取数据并放入队列。
        使用 io.open(fileno, "rb", closefd=False).read1() 实现非阻塞读取。
        """
        # 注意：这里out是subprocess.PIPE，其fileno是可用的
        stream = io.open(out.fileno(), "rb", closefd=False)
        while True:
            # read1() 会尽可能多地读取数据，但不会阻塞等待更多数据
            # 如果没有数据，它会立即返回空字节串
            n = stream.read1()
            if len(n) > 0:
                queue.put(n)
            elif self.process.poll() is not None: # 进程已结束且管道已空
                break
            else:
                # 管道暂时为空，但进程可能还在运行，稍作等待避免CPU空转
                time.sleep(0.01) # 避免忙等待
        # stream.close() # 注意：不要关闭subprocess.PIPE，它由Popen管理

    def _start_reader_thread(self, out_pipe: IO[bytes], queue: Queue[bytes]):
        """
        为给定的输出管道启动一个守护线程来读取数据并放入队列。
        """
        t = Thread(target=self._enqueue_output, args=(out_pipe, queue))
        t.daemon = True  # 设置为守护线程，主程序退出时自动终止
        t.start()

    def get_all_output(self, timeout: float = None) -> tuple[str, str]:
        """
        等待进程结束或达到超时，然后收集所有标准输出和标准错误。
        :param timeout: 等待进程结束的秒数。
        :return: 包含标准输出和标准错误的元组。
        """
        try:
            # communicate() 会等待进程结束，或达到超时。
            # 由于我们已经通过队列异步读取了输出，这里的communicate
            # 主要用于等待进程结束和获取其返回码。
            # 如果不传input，它不会阻塞stdin。
            self.process.communicate(timeout=timeout)
        except subprocess.TimeoutExpired:
            print(f"ERROR: Process timed out after {timeout} seconds. Attempting to kill.")
            self.process.kill()
            self.process.wait() # 确保进程完全终止
        except Exception as e:
            print(f"An error occurred during communicate: {e}")
        finally:
            stdout_content = self._drain_queue(self.stdout_queue)
            stderr_content = self._drain_queue(self.stderr_queue)
            return stdout_content, stderr_content

    def _drain_queue(self, queue: Queue[bytes]) -> str:
        """从队列中清空所有数据并解码为字符串。"""
        collected_output = []
        try:
            while True:
                collected_output.append(queue.get_nowait())
        except Empty:
            pass # 队列已空
        return b"".join(collected_output).decode(errors='ignore') # 忽略解码错误

# -------------------------- 示例使用 --------------------------

# 准备一个测试脚本 x.py
# print("hi")
# time.sleep(0.5)
# name = input("Your name: ")
# print(f"Hello, {name}!")
# time.sleep(1)
# print("Exiting x.py")
# import sys
# print("Error message", file=sys.stderr)

# 示例1：正常运行并提供输入
print("--- 示例1: 正常运行并提供输入 ---")
runner1 = AdvancedRunner(["python", "x.py"], stdin_input="Alice")
stdout1, stderr1 = runner1.get_all_output(timeout=5)
print("\n=== STDOUT ===")
print(stdout1)
print("=== STDERR ===")
print(stderr1)
print(f"Process exited with code: {runner1.process.returncode}\n")

# 示例2：模拟进程超时
print("--- 示例2: 模拟进程超时 ---")
# 假设x.py中有一个很长的sleep或者等待输入
# 为演示，我们可以用一个简单的无限循环脚本
# infinite_loop.py:
# import time
# while True:
#     print("Looping...", flush=True)
#     time.sleep(1)
runner2 = AdvancedRunner(["python", "-c", "import time; while True: print('Looping...', flush=True); time.sleep(1)"], timeout=2)
stdout2, stderr2 = runner2.get_all_output(timeout=2)
print("\n=== STDOUT ===")
print(stdout2)
print("=== STDERR ===")
print(stderr2)
print(f"Process exited with code: {runner2.process.returncode}\n")

# 示例3：无输入，只捕获输出
print("--- 示例3: 无输入，只捕获输出 ---")
# simple_output.py:
# print("Just some output.")
# import sys
# print("And some error.", file=sys.stderr)
runner3 = AdvancedRunner(["python", "-c", "print('Just some output.'); import sys; print('And some error.', file=sys.stderr)"])
stdout3, stderr3 = runner3.get_all_output(timeout=5)
print("\n=== STDOUT ===")
print(stdout3)
print("=== STDERR ===")
print(stderr3)
print(f"Process exited with code: {runner3.process.returncode}\n")