CPU压力测试

工具环境：python3

运行环境：SOC端内部

测试用途：给SOC的CPU单个核以及MEM加压

文件说明以及主要用法：

""" CPU Loader """ import os import sys import time import math import signal import argparse from multiprocessing import Process TOTAL_CORES = 12 def cpu_worker(core_id, load_pct, interval): """ 绑定核 core_id: 指定核的id load_pct：占用百分比 interval：控制时长(秒) """ try: os.sched_setaffinity(0, {core_id}) except Exception as e: print("[Worker-%d] Cannot bind to core %d: %s" % (core_id, core_id, e)) return print("[Worker-%d] Started, target load: %d%%" % (core_id, load_pct)) while True: busy = interval * (load_pct / 100.0) idle = interval - busy start = time.perf_counter() while (time.perf_counter() - start) < busy: math.sqrt(12345.6789) math.sin(0.5) math.cos(0.5) math.pow(2, 10) if idle > 0: time.sleep(idle) def mem_worker(size_mb): """ MEM占用 size_mb：占用大小（MB） """ data = [] chunk = 1024 * 1024 print("[Memory] Allocating %d MB" % size_mb) for i in range(size_mb): data.append(bytearray(chunk)) if i % 10 == 0: for j in range(0, len(data), max(1, len(data)//10)): data[j][0] = 1 while True: time.sleep(1) for chunk in data: chunk[0] = 1 def signal_handler(signum, frame): print("Stopping...") sys.exit(0) def main(): """ 命令基本描述： -c, --cores 指定核心 0,1,2 或 all 0,1,2,3,4,5 -l, --load 负载百分比 0-100 50 -a, --all 使用全部12核 - -m, --memory 额外内存占用(MB) 0 -d, --daemon 后台守护运行 - -i, --interval 控制周期(秒) 0.1 """ parser = argparse.ArgumentParser(description="CPU Loader") parser.add_argument("-c", "--cores", default="0,1,2,3,4,5", help="Cores to use: 0,1,2 or all") parser.add_argument("-l", "--load", type=int, default=50, help="Load percent 0-100") parser.add_argument("-a", "--all", action="store_true", help="Use all 12 cores") parser.add_argument("-m", "--memory", type=int, default=0, help="Memory to allocate in MB") parser.add_argument("-d", "--daemon", action="store_true", help="Run as daemon") parser.add_argument("-i", "--interval", type=float, default=0.1, help="Control interval in seconds") args = parser.parse_args() signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) # Parse core list if args.all or args.cores == "all": cores = list(range(TOTAL_CORES)) elif "," in args.cores: cores = [int(x.strip()) for x in args.cores.split(",")] else: cores = [int(args.cores)] # Validate for c in cores: if c < 0 or c >= TOTAL_CORES: print("Error: Core %d out of range (0-%d)" % (c, TOTAL_CORES-1)) sys.exit(1) if args.load < 0 or args.load > 100: print("Error: Load must be 0-100") sys.exit(1) print("=" * 50) print("CPU Loader") print("=" * 50) print("Cores: %s" % cores) print("Load: %d%%" % args.load) print("Interval: %.2fs" % args.interval) if args.memory > 0: print("Memory: %d MB" % args.memory) print("=" * 50) # Daemon mode if args.daemon: pid = os.fork() if pid > 0: print("Daemon PID: %d" % pid) sys.exit(0) os.setsid() # Start workers procs = [] for c in cores: p = Process(target=cpu_worker, args=(c, args.load, args.interval)) p.start() procs.append(p) if args.memory > 0: p = Process(target=mem_worker, args=(args.memory,)) p.start() procs.append(p) print("Started %d worker(s), Ctrl+C to stop" % len(cores)) try: while True: time.sleep(1) except KeyboardInterrupt: pass finally: print("Stopping workers...") for p in procs: p.terminate() p.join(timeout=1) if p.is_alive(): p.kill() print("Done") if __name__ == "__main__": main()

给CPU和MEM加压的主要工具

""" CPU Monitor """ import os import time import sys def read_stats(): with open('/proc/stat', 'r') as f: lines = f.readlines() data = {} for line in lines: if line.startswith('cpu'): parts = line.split() data[parts[0]] = list(map(int, parts[1:])) return data def calc_usage(prev, curr): p_total = sum(prev) c_total = sum(curr) diff = c_total - p_total if diff == 0: return 0.0 p_idle = prev[3] + prev[4] c_idle = curr[3] + curr[4] idle_diff = c_idle - p_idle return (diff - idle_diff) / diff * 100 def get_freq(core): try: path = '/sys/devices/system/cpu/cpu%d/cpufreq/scaling_cur_freq' % core with open(path, 'r') as f: return int(f.read().strip()) / 1000 except: return 0 def get_temp(): temps = [] for i in range(10): try: path = '/sys/class/thermal/thermal_zone%d/temp' % i with open(path, 'r') as f: temps.append(int(f.read().strip()) / 1000) except: pass return max(temps) if temps else 0 def main(): print("=" * 60) print("AGX CPU Monitor") print("=" * 60) print("Press Ctrl+C to exit") prev = read_stats() try: while True: time.sleep(1) curr = read_stats() os.system('clear') print("=" * 60) print("CPU Monitor - %s" % time.strftime('%H:%M:%S')) print("=" * 60) if 'cpu' in prev and 'cpu' in curr: total = calc_usage(prev['cpu'], curr['cpu']) temp = get_temp() print("Total: %.1f%% | Temp: %.1fC" % (total, temp)) print("%-6s %-8s %-10s %-8s %-8s %-8s" % ( "Core", "Load", "Freq", "User", "Sys", "Idle")) print("-" * 60) for i in range(12): name = 'cpu%d' % i if name in prev and name in curr: usage = calc_usage(prev[name], curr[name]) freq = get_freq(i) d = [c - p for c, p in zip(curr[name], prev[name])] total_d = sum(d) if sum(d) > 0 else 1 user = d[0] / total_d * 100 sys_ = d[2] / total_d * 100 idle = (d[3] + d[4]) / total_d * 100 color = "" reset = "" if usage >= 80: color = "[31m" elif usage >= 50: color = "[33m" else: color = "[32m" reset = "[0m" print("%sCPU%-3d %6.1f%% %6.0fMHz %6.1f%% %6.1f%% %6.1f%%%s" % ( color, i, usage, freq, user, sys_, idle, reset)) print("" + "=" * 60) print("Green: <50%% | Yellow: 50-80%% | Red: >80%%") prev = curr except KeyboardInterrupt: print("==========================Exited==========================") if __name__ == "__main__": main()

正常运行的结果如下：

运行参数举例说明：

占用6个核心，50%负载

python3 cpu_loader.py -c 0,1,2,3,4,5 -l 50

占用所有12核，30%负载

python3 cpu_loader.py -a -l 30

指定核心满载

python3 cpu_loader.py -c 0,1,2 -l 100

后台运行

python3 cpu_loader.py -c 6 -l 80 -d

同时占用内存

python3 cpu_loader.py -c 6 -l 80 -m 1000

运行python3 cpu_monitor.py

一个在SOC内部实时监控CPU情况的终端

正常运行的情况如下：

退出请按ctrl+c

基于上面的工具的快速测试脚本

# CPU 负载测试脚本 # 一键测试各种负载场景 echo "==========================================" echo "🖥️ CPU 负载测试工具" echo "==========================================" echo "" # 检查是否 root if [ "$EUID" -ne 0 ]; then echo "⚠️ 建议使用 sudo 运行以获得最佳效果" echo "" fi # 测试场景 declare -a tests=( "6核50%负载:python3 cpu_loader.py -c 0,1,2,3,4,5 -l 50" "6核80%负载:python3 cpu_loader.py -c 0,1,2,3,4,5 -l 80" "6核100%负载:python3 cpu_loader.py -c 0,1,2,3,4,5 -l 100" "12核30%负载:python3 cpu_loader.py -a -l 30" "12核50%负载:python3 cpu_loader.py -a -l 50" "12核100%负载:python3 cpu_loader.py -a -l 100" "单核满载:python3 cpu_loader.py -c 0 -l 100" "4核满载+内存:python3 cpu_loader.py -c 0,1,2,3 -l 100 -m 500" ) echo "选择测试场景:" for i in "${!tests[@]}"; do name="${tests[$i]%%:*}" echo " $((i+1)). $name" done echo " 0. 退出" echo "" read -p "请输入编号: " choice if [ "$choice" -eq 0 ]; then exit 0 fi if [ "$choice" -lt 1 ] || [ "$choice" -gt "${#tests[@]}" ]; then echo "❌ 无效选择" exit 1 fi # 获取命令 cmd="${tests[$((choice-1))]#*:}" name="${tests[$((choice-1))]%%:*}" echo "" echo "==========================================" echo "🚀 启动测试: $name" echo "命令: $cmd" echo "==========================================" echo "" echo "按 Ctrl+C 停止" echo "" # 在新终端中运行监控 if command -v gnome-terminal &> /dev/null; then gnome-terminal -- python3 agx_cpu_monitor.py & elif command -v xterm &> /dev/null; then xterm -e python3 agx_cpu_monitor.py & fi # 运行负载 $cmd # 清理 echo "" echo "✅ 测试结束"