cpp_compilation_benchmark_new.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PaimonCore Boost - C++编译性能基准测试器
验证P核激活对C++编译任务的性能提升效果

作者: Geoffrey Wang
项目: PaimonCore Boost - Adaptive Hybrid Core Scheduler
灵感: Genshin Impact's Idle Boost Phenomenon
许可证: Apache License 2.0
"""

import os
import sys
import time
import json
import subprocess
import threading
from datetime import datetime
from colorama import init, Fore, Style
import psutil
from system_env import SystemEnvironment

init(autoreset=True)

class CppCompilationBenchmark:
    def __init__(self):
        self.benchmark_dir = "cpp_benchmark"
        self.results_dir = "cpp_compile_results"
        self.system_env = SystemEnvironment()
        self.compilers = self.system_env.detect_compilers()
        
        self.test_files = [
            ("fast_template_test.cpp", "模板编译测试"),
            ("fast_stl_test.cpp", "STL库密集型测试"), 
            ("fast_math_test.cpp", "数学计算密集型测试"),
            ("simple_test.cpp", "简单编译测试")
        ]

    def setup_compiler_config(self):
        """
        配置编译器设置，使用动态环境检测替代硬编码路径
        """
        print(f"\n{Fore.CYAN}🔧 配置编译器环境...")
        
        # 验证系统环境
        is_valid, issues = self.system_env.validate_environment()
        if not is_valid:
            print(f"{Fore.RED}❌ 系统环境验证失败:")
            for issue in issues:
                print(f"   - {issue}")
            return False
            
        print(f"{Fore.GREEN}✅ 检测到 {len(self.compilers)} 个可用编译器")
        
        for compiler_id, config in self.compilers.items():
            print(f"   {Fore.YELLOW}{config['name']}: {config['command']}")
            
        return True

    def check_compiler_availability(self, compiler_id):
        """检查编译器是否可用，使用动态环境检测"""
        if compiler_id not in self.compilers:
            return False
            
        config = self.compilers[compiler_id]
        
        try:
            # 使用系统环境检测的路径
            if compiler_id == "msvc":
                # MSVC需要特殊的环境设置
                return self.system_env._detect_msvc() is not None
            else:
                # 其他编译器直接检查命令可用性
                result = subprocess.run([config["command"], "--version"], 
                                      capture_output=True, text=True, timeout=5, check=False)
                return result.returncode == 0
        except (subprocess.TimeoutExpired, subprocess.SubprocessError, FileNotFoundError):
            return False

    def compile_with_timing(self, source_path, output_path, compiler_id):
        """
        使用指定编译器编译文件并测量时间
        使用动态环境配置替代硬编码路径
        """
        if compiler_id not in self.compilers:
            return None, f"未知编译器: {compiler_id}"
            
        compiler_config = self.compilers[compiler_id]
        
        # 构建编译命令
        if compiler_id == "msvc":
            # 使用动态检测的MSVC环境
            msvc_env = self.system_env._detect_msvc()
            if not msvc_env:
                return None, "MSVC环境未找到"
                
            vcvars_path = msvc_env["vcvars_path"]
            
            cmd_parts = [
                f'"{vcvars_path}" x64',
                '&&',
                compiler_config["command"],
                *compiler_config["flags"],
                f'"{source_path}"',
                f'{compiler_config["output_flag"]}"{output_path}"'
            ]
            command_str = ' '.join(cmd_parts)
            
            # 使用shell=True来执行cmd命令
            command_for_subprocess = command_str
            use_shell = True
        else:
            # Clang等其他编译器正常处理
            command = [
                compiler_config["command"],
                *compiler_config["flags"],
                source_path
            ]
            
            if compiler_config["output_flag"] == "-o":
                command.extend([compiler_config["output_flag"], output_path])
            else:
                command.extend([compiler_config["output_flag"] + output_path])
            
            command_for_subprocess = command
            command_str = ' '.join(command)
            use_shell = False
        
        print(f"   执行命令: {command_str}")
        
        # 记录编译开始时的系统状态
        start_cpu = psutil.cpu_percent(interval=0.1)
        start_memory = psutil.virtual_memory().percent
        
        compile_start_time = time.perf_counter()
        
        try:
            # 执行编译命令
            if use_shell:
                process = subprocess.Popen(
                    command_for_subprocess,
                    shell=True,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True
                )
            else:
                process = subprocess.Popen(
                    command_for_subprocess,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True
                )
            
            stdout, stderr = process.communicate(timeout=60)
            compile_end_time = time.perf_counter()
            
            # 记录编译结束时的系统状态
            end_cpu = psutil.cpu_percent(interval=0.1)
            end_memory = psutil.virtual_memory().percent
            
            compile_time = compile_end_time - compile_start_time
            
            if process.returncode == 0:
                # 编译成功
                compilation_info = {
                    'time': compile_time,
                    'success': True,
                    'output_size': os.path.getsize(output_path) if os.path.exists(output_path) else 0,
                    'cpu_usage': {'start': start_cpu, 'end': end_cpu},
                    'memory_usage': {'start': start_memory, 'end': end_memory},
                    'stdout': stdout,
                    'stderr': stderr
                }
                
                print(f"   {Fore.GREEN}✅ 编译成功: {compile_time:.3f}s")
                return compilation_info, None
            else:
                # 编译失败
                print(f"   {Fore.RED}❌ 编译失败 (返回码: {process.returncode})")
                print(f"   错误输出: {stderr}")
                return None, f"编译失败: {stderr}"
                
        except subprocess.TimeoutExpired:
            process.kill()
            return None, "编译超时"
        except Exception as e:
            return None, f"编译异常: {str(e)}"

    def create_test_files(self):
        """创建测试用的C++源文件"""
        
        if not os.path.exists(self.benchmark_dir):
            os.makedirs(self.benchmark_dir)
            
        # 复杂模板编译测试
        template_test = '''
#include <iostream>
#include <vector>
#include <map>
#include <string>
#include <algorithm>
#include <memory>
#include <chrono>
#include <thread>

template<typename T, int N>
class ComplexTemplate {
public:
    std::vector<T> data;
    std::map<std::string, T> lookup;
    
    ComplexTemplate() {
        data.reserve(N);
        for(int i = 0; i < N; ++i) {
            data.push_back(static_cast<T>(i));
            lookup["item_" + std::to_string(i)] = static_cast<T>(i);
        }
    }
    
    template<typename U>
    auto process(U func) -> decltype(func(T{})) {
        decltype(func(T{})) result{};
        for(const auto& item : data) {
            result += func(item);
        }
        return result;
    }
};

int main() {
    ComplexTemplate<double, 1000> ct;
    auto result = ct.process([](double x) { return x * x + 1.0; });
    std::cout << "模板处理结果: " << result << std::endl;
    return 0;
}
'''

        # STL密集型测试
        stl_test = '''
#include <iostream>
#include <vector>
#include <list>
#include <set>
#include <map>
#include <unordered_map>
#include <algorithm>
#include <numeric>
#include <string>

int main() {
    // 向量操作
    std::vector<int> vec(10000);
    std::iota(vec.begin(), vec.end(), 1);
    
    // 映射操作
    std::map<std::string, int> string_map;
    std::unordered_map<int, std::string> int_map;
    
    for(size_t i = 0; i < vec.size(); ++i) {
        string_map["key_" + std::to_string(i)] = vec[i];
        int_map[vec[i]] = "value_" + std::to_string(i);
    }
    
    // 排序和查找
    std::sort(vec.begin(), vec.end(), std::greater<int>());
    auto it = std::find(vec.begin(), vec.end(), 5000);
    
    // 集合操作
    std::set<int> unique_values(vec.begin(), vec.end());
    
    std::cout << "STL操作完成，处理了 " << vec.size() << " 个元素" << std::endl;
    std::cout << "唯一值数量: " << unique_values.size() << std::endl;
    return 0;
}
'''

        # 数学计算密集型测试  
        math_test = '''
#include <iostream>
#include <cmath>
#include <vector>
#include <complex>
#include <random>

class MathProcessor {
private:
    std::vector<double> data;
    std::mt19937 gen;
    
public:
    MathProcessor(size_t size) : gen(std::random_device{}()) {
        data.resize(size);
        std::uniform_real_distribution<double> dist(-100.0, 100.0);
        for(auto& val : data) {
            val = dist(gen);
        }
    }
    
    double calculateComplexOperations() {
        double result = 0.0;
        for(const auto& val : data) {
            result += std::sin(val) * std::cos(val * 2);
            result += std::exp(val / 100.0);
            result += std::log(std::abs(val) + 1.0);
            result += std::sqrt(std::abs(val));
        }
        return result;
    }
    
    std::complex<double> processComplex() {
        std::complex<double> result(0.0, 0.0);
        for(size_t i = 0; i < data.size(); ++i) {
            std::complex<double> c(data[i], data[(i + 1) % data.size()]);
            result += std::pow(c, 2.0) + std::sin(c);
        }
        return result;
    }
};

int main() {
    MathProcessor processor(5000);
    
    double real_result = processor.calculateComplexOperations();
    auto complex_result = processor.processComplex();
    
    std::cout << "数学计算结果: " << real_result << std::endl;
    std::cout << "复数计算结果: " << complex_result << std::endl;
    return 0;
}
'''

        # 简单测试
        simple_test = '''
#include <iostream>
#include <vector>

int main() {
    std::vector<int> numbers = {1, 2, 3, 4, 5};
    
    int sum = 0;
    for(int num : numbers) {
        sum += num;
    }
    
    std::cout << "简单测试完成，总和: " << sum << std::endl;
    return 0;
}
'''

        # 写入测试文件
        test_content = {
            "fast_template_test.cpp": template_test,
            "fast_stl_test.cpp": stl_test,
            "fast_math_test.cpp": math_test,
            "simple_test.cpp": simple_test
        }
        
        for filename, content in test_content.items():
            filepath = os.path.join(self.benchmark_dir, filename)
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(content)
                
        print(f"{Fore.GREEN}✅ 创建了 {len(test_content)} 个测试文件")

    def run_single_benchmark(self, compiler_id, boost_enabled=False):
        """运行单个编译器的基准测试"""
        
        print(f"\n{Fore.CYAN}🔧 运行 {self.compilers[compiler_id]['name']} 基准测试 "
              f"({'P核激活' if boost_enabled else '正常模式'})")
        
        results = []
        
        for filename, description in self.test_files:
            print(f"\n{Fore.YELLOW}📝 {description} ({filename})")
            
            source_path = os.path.join(self.benchmark_dir, filename)
            output_name = filename.replace('.cpp', '_' + compiler_id + ('_boost' if boost_enabled else '') + '.exe')
            output_path = os.path.join(self.results_dir, output_name)
            
            # 编译并计时
            compile_result, error = self.compile_with_timing(source_path, output_path, compiler_id)
            
            if compile_result:
                result_data = {
                    'test_file': filename,
                    'description': description,
                    'compiler': compiler_id,
                    'compiler_name': self.compilers[compiler_id]['name'],
                    'boost_enabled': boost_enabled,
                    'compile_time': compile_result['time'],
                    'success': True,
                    'output_size': compile_result['output_size'],
                    'cpu_usage': compile_result['cpu_usage'],
                    'memory_usage': compile_result['memory_usage'],
                    'timestamp': datetime.now().isoformat()
                }
                results.append(result_data)
                
                print(f"   编译时间: {compile_result['time']:.3f}s")
                print(f"   输出大小: {compile_result['output_size']} 字节")
            else:
                result_data = {
                    'test_file': filename,
                    'description': description,
                    'compiler': compiler_id,
                    'compiler_name': self.compilers[compiler_id]['name'],
                    'boost_enabled': boost_enabled,
                    'success': False,
                    'error': error,
                    'timestamp': datetime.now().isoformat()
                }
                results.append(result_data)
                
                print(f"   {Fore.RED}编译失败: {error}")
        
        return results

    def run_comprehensive_benchmark(self):
        """运行完整的基准测试"""
        
        print(f"{Fore.MAGENTA}🚀 PaimonCore Boost - C++编译性能基准测试")
        print(f"{Fore.MAGENTA}{'='*60}")
        
        # 设置编译器配置
        if not self.setup_compiler_config():
            print(f"{Fore.RED}❌ 编译器配置失败，无法继续测试")
            return
        
        # 创建输出目录
        if not os.path.exists(self.results_dir):
            os.makedirs(self.results_dir)
            
        # 创建测试文件
        print(f"\n{Fore.CYAN}📁 准备测试文件...")
        self.create_test_files()
        
        all_results = []
        available_compilers = []
        
        # 检查可用编译器
        for compiler_id in self.compilers.keys():
            if self.check_compiler_availability(compiler_id):
                available_compilers.append(compiler_id)
                print(f"{Fore.GREEN}✅ {self.compilers[compiler_id]['name']} 可用")
            else:
                print(f"{Fore.YELLOW}⚠️  {self.compilers[compiler_id]['name']} 不可用")
        
        if not available_compilers:
            print(f"{Fore.RED}❌ 没有可用的编译器")
            return
        
        # 运行基准测试
        for compiler_id in available_compilers:
            # 正常模式测试
            normal_results = self.run_single_benchmark(compiler_id, boost_enabled=False)
            all_results.extend(normal_results)
            
            # P核激活模式测试  
            boost_results = self.run_single_benchmark(compiler_id, boost_enabled=True)
            all_results.extend(boost_results)
        
        # 保存结果
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        results_file = f"cpp_benchmark_results_{timestamp}.json"
        results_path = os.path.join(self.results_dir, results_file)
        
        with open(results_path, 'w', encoding='utf-8') as f:
            json.dump(all_results, f, ensure_ascii=False, indent=2)
        
        print(f"\n{Fore.GREEN}✅ 结果已保存到: {results_path}")
        
        # 分析和显示结果
        self.analyze_results(all_results)
        
        return all_results

    def analyze_results(self, results):
        """分析和显示基准测试结果"""
        
        print(f"\n{Fore.MAGENTA}📊 性能分析结果")
        print(f"{Fore.MAGENTA}{'='*60}")
        
        # 按编译器和测试文件分组
        grouped_results = {}
        
        for result in results:
            if not result['success']:
                continue
                
            compiler = result['compiler']
            test_file = result['test_file']
            
            key = f"{compiler}_{test_file}"
            
            if key not in grouped_results:
                grouped_results[key] = {'normal': None, 'boost': None}
                
            if result['boost_enabled']:
                grouped_results[key]['boost'] = result
            else:
                grouped_results[key]['normal'] = result
        
        # 计算性能提升
        improvements = []
        
        print(f"\n🔍 总体性能对比:")
        print(f"{'测试项目':<25} {'编译器':<12} {'正常模式':<12} {'P核激活':<12} {'提升幅度':<12}")
        print("-" * 80)
        
        for key, data in grouped_results.items():
            if data['normal'] and data['boost']:
                normal_time = data['normal']['compile_time']
                boost_time = data['boost']['compile_time']
                improvement = ((normal_time - boost_time) / normal_time) * 100
                
                improvements.append(improvement)
                
                compiler_name = data['normal']['compiler_name'][:10]
                test_desc = data['normal']['description'][:23]
                
                print(f"{test_desc:<25} {compiler_name:<12} {normal_time:<12.3f} {boost_time:<12.3f} {improvement:>+11.2f}%")
        
        if improvements:
            avg_improvement = sum(improvements) / len(improvements)
            max_improvement = max(improvements)
            min_improvement = min(improvements)
            
            print(f"\n📋 详细对比:")
            print(f"平均性能提升: {avg_improvement:+.2f}%")
            print(f"最大性能提升: {max_improvement:+.2f}%")  
            print(f"最小性能提升: {min_improvement:+.2f}%")
            
            if avg_improvement > 0:
                print(f"\n{Fore.GREEN}🎉 P核激活平均提升编译性能 {avg_improvement:.2f}%")
            else:
                print(f"\n{Fore.YELLOW}⚠️  P核激活在当前环境下性能提升有限")

def main():
    try:
        benchmark = CppCompilationBenchmark()
        benchmark.run_comprehensive_benchmark()
        
    except KeyboardInterrupt:
        print(f"\n{Fore.YELLOW}⚠️  测试被用户中断")
    except Exception as e:
        print(f"\n{Fore.RED}❌ 测试过程中发生错误: {e}")
        return 1
    
    return 0

if __name__ == "__main__":
    sys.exit(main())