Python在Unix环境下监控子进程的内存与CPU时间使用教程

import subprocess
import resource
import sys
import time

def measure_cpu_time(command_list):
    """
    测量子进程的CPU时间（用户态和系统态）。
    """
    print(f"Executing command: {' '.join(command_list)}")

    # 1. 启动前获取资源使用情况（包括已终止的子进程，通常为0）
    usage_start = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 2. 启动子进程
    # subprocess.DEVNULL 避免子进程输出到控制台
    # subprocess.PIPE 捕获标准错误，以便在失败时打印
    process = subprocess.Popen(command_list, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, encoding='utf-8')

    # 3. 等待子进程完成
    # process.wait() 会阻塞直到子进程终止
    # 或者使用 process.poll() 循环等待，以便在循环中进行其他监控（如内存）
    process.wait()

    # 4. 子进程结束后获取资源使用情况
    usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 5. 检查子进程退出码
    if process.returncode != 0:
        error_message = process.stderr.strip()
        sys.exit(f"FAILED: {' '.join(command_list)}\n{error_message}")

    # 6. 计算CPU时间
    cpu_time_user = usage_end.ru_utime - usage_start.ru_utime
    cpu_time_system = usage_end.ru_stime - usage_start.ru_stime

    print(f"User CPU Time: {cpu_time_user:.6f} seconds")
    print(f"System CPU Time: {cpu_time_system:.6f} seconds")
    return cpu_time_user, cpu_time_system

if __name__ == "__main__":
    # 示例：运行一个模拟耗时任务的命令
    # 例如：`sleep 5` 会等待5秒，但CPU时间很低
    # 更能体现CPU时间的命令可以是计算密集型任务，如 `dd if=/dev/zero of=/dev/null count=1000000 bs=1K`
    # 或者一个简单的Python脚本
    # command_to_run = ["sleep", "5"]
    command_to_run = ["python", "-c", "import time; sum(range(10**7)); time.sleep(1); sum(range(10**7))"]
    measure_cpu_time(command_to_run)

import psutil
import time

def get_process_memory_usage(pid):
    """
    获取指定进程的RSS内存使用量（以GB为单位）。
    """
    try:
        process = psutil.Process(pid)
        return process.memory_info().rss / (1024.0 ** 3) # 转换为GB
    except psutil.NoSuchProcess:
        # 进程可能已经结束
        return 0.0
    except psutil.AccessDenied:
        # 权限不足，无法获取进程信息
        print(f"Warning: Access denied for process {pid}. Cannot get memory info.", file=sys.stderr)
        return 0.0

def monitor_memory_snapshots(process_obj, slice_in_seconds=1):
    """
    实时监控并收集子进程的内存快照。
    """
    memory_snapshots = []
    # 循环直到子进程结束
    while process_obj.poll() is None:
        mem_gb = get_process_memory_usage(process_obj.pid)
        if mem_gb > 0: # 只有当成功获取到内存信息时才记录
            memory_snapshots.append(mem_gb)
        time.sleep(slice_in_seconds)
    return memory_snapshots

# 注意：此函数不能单独运行，需要与subprocess结合
# 完整的集成示例将在下一节提供

import sys
import os
import subprocess
import resource
import psutil
import time

def get_process_memory_usage(pid):
    """
    获取指定进程的RSS内存使用量（以GB为单位）。
    """
    try:
        process = psutil.Process(pid)
        return process.memory_info().rss / (1024.0 ** 3) # 转换为GB
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        return 0.0

def monitor_subprocess_performance(command_list, report_file_path="performance_report.txt", slice_in_seconds=1):
    """
    监控子进程的CPU时间（用户态和系统态）及内存使用情况。
    """
    print(f"Executing command: {' '.join(command_list)}")

    # 1. 启动前获取资源使用情况
    usage_start = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 2. 启动子进程
    process = subprocess.Popen(command_list, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, encoding='utf-8')

    # 3. 实时监控内存使用
    memory_snapshots = []
    # 循环直到子进程结束
    while process.poll() is None:
        mem_gb = get_process_memory_usage(process.pid)
        if mem_gb > 0:
            memory_snapshots.append(mem_gb)
        time.sleep(slice_in_seconds)

    # 4. 子进程结束后获取资源使用情况
    usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 5. 检查子进程退出码
    if process.returncode != 0:
        error_message = process.stderr.strip()
        print(f"ERROR: Subprocess failed with code {process.returncode}:\n{error_message}", file=sys.stderr)
        sys.exit(f"FAILED: {' '.join(command_list)}\n{error_message}")

    # 6. 计算CPU时间
    cpu_time_user = usage_end.ru_utime - usage_start.ru_utime
    cpu_time_system = usage_end.ru_stime - usage_start.ru_stime

    # 7. 写入报告文件
    with open(report_file_path, "w") as f:
        f.write(f"Command: {' '.join(command_list)}\n")
        f.write(f"User CPU Time: {cpu_time_user:.6f} seconds\n")
        f.write(f"System CPU Time: {cpu_time_system:.6f} seconds\n")
        f.write(f"Peak RSS Memory: {max(memory_snapshots) if memory_snapshots else 0:.3f} GB\n")
        f.write(f"Memory Snapshots (RSS in GB): {memory_snapshots}\n")

    print(f"Monitoring complete. Report saved to {report_file_path}")
    print(f"User CPU Time: {cpu_time_user:.6f} seconds")
    print(f"System CPU Time: {cpu_time_system:.6f} seconds")
    print(f"Peak RSS Memory: {max(memory_snapshots) if memory_snapshots else 0:.3f} GB")
    print(f"Memory Snapshots (RSS in GB): {memory_snapshots}")

if __name__ == "__main__":
    # 示例：运行一个模拟耗时且有一定CPU和内存使用的命令
    # 这个Python脚本会进行一些计算并暂停，模拟真实工具的行为
    # 注意：在某些系统上，Python解释器本身也会消耗一定的CPU和内存
    command_to_run = [
        sys.executable, "-c",
        "import time, os; "
        "data = [i*i for i in range(10**6)]; "  # 模拟内存分配和一些计算
        "time.sleep(2); " # 暂停2秒
        "data2 = [x+1 for x in data]; " # 更多计算
        "time.sleep(1); " # 暂停1秒
        "print('Subprocess finished calculation and sleeping.');"
    ]

    # 您也可以替换为您的生物信息学工具命令，例如：
    # command_to_run = ["your_bioinformatics_tool", "--input", "file.fa", "--output", "result.txt"]

    monitor_subprocess_performance(command_to_run, "tool_performance_report.txt", slice_in_seconds=1)

    print("\n--- Testing a command with minimal CPU/memory (e.g., sleep) ---")
    monitor_subprocess_performance(["sleep", "3"], "sleep_performance_report.txt", slice_in_seconds=1)