feat: 增强多线程性能基准测试工具功能

- 添加丰富的命令行参数支持，包括静默模式、自定义参数、跳过测试、输出报告等
- 实现超时机制，防止单个测试无限期运行
- 新增JSON格式测试报告生成和保存功能
- 优化用户界面，增加详细的使用示例和更好的日志输出
- 改进错误处理和异常管理，提升程序健壮性
- 在静默模式下自动使用系统默认Python路径
- 更新.gitignore排除输出文件和报告文件
- 重构参数传递机制，统一使用args对象
- 增强测试结果展示，包含相对速度对比

.gitignore

@@ -32,6 +32,10 @@ uv.lock
 # Logs
 *.log
 
+# output files
+output*
+report*
+
 # Runtime data
 pids
 *.pid

main.py

@@ -1,22 +1,57 @@
 # main.py
 import subprocess
 import json
+import csv
 import os
 import argparse
+import multiprocessing
+from datetime import datetime
 
 def parse_arguments():
-    """解析命令行参数"""
-    parser = argparse.ArgumentParser(description="Python 多线程性能基准测试工具")
-    parser.add_argument("-v", "--verbose", action="store_true",
-                        help="启用详细模式，显示每个线程的日志")
+    """
+    解析命令行参数
+    """
+    parser = argparse.ArgumentParser(
+        description="Python 多线程性能基准测试工具",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""示例:
+        1. 查看帮助文档
+            python main.py -h
+        2. 基础测试（默认参数）
+            python main.py
+        3. 详细模式 + 自定义质数数量和线程数
+            python main.py -v -n 50000 -t 12
+        4. 跳过GIL测试 + 输出报告到文件 + 超时3分钟
+            python main.py --skip gil --output report.json --timeout 180
+        5. 静默模式 + 仅测试NoGIL版本
+            python main.py -q --skip single --skip gil
+        """
+    )
+    parser.add_argument("-v", "--verbose", action="store_true",help="启用详细模式，显示每个线程的日志")
+    parser.add_argument("-q", "--quiet", action="store_true", help="静默模式，仅输出最终报告和错误信息")
+    parser.add_argument("-n", "--num-primes", type=int, default=30000, help="要寻找的质数总数（默认：30000）")
+    parser.add_argument("-t", "--threads", type=int, default=multiprocessing.cpu_count(), help="多线程测试的线程数（默认：CPU核心数）")
+    parser.add_argument("-o", "--output", type=str, help="将测试报告保存到指定JSON文件（例：report.json）")
+    parser.add_argument("-s", "--skip", action="append", choices=["single", "gil", "nogil"], help="跳过指定测试类型（可多次使用，例：--skip gil --skip single）")
+    parser.add_argument("-l", "--timeout", type=int, default=300, help="单个测试的超时时间（秒，默认：300）")
+
     return parser.parse_args()
 
-def get_python_path(prompt):
-    """获取用户输入的Python可执行文件路径，并验证其有效性"""
+def get_python_path(prompt, args):
+    """
+    获取用户输入的Python可执行文件路径，并验证其有效性
+    """
     while True:
+        if args.quiet:
+            # 静默模式下自动使用系统默认Python（可根据需求修改）
+            import sys
+            path = sys.executable
+            if not args.quiet:
+                print(f"  使用默认Python路径: {path}")
+            return path
         path = input(prompt).strip()
         if not path:
-            print("路径不能为空。")
+            print("  路径不能为空。")
             continue
         
         if os.path.exists(path) and os.access(path, os.X_OK):
@@ -31,36 +66,50 @@ def get_python_path(prompt):
         else:
             print("  路径不存在或不可执行。")
 
-def run_test(python_path, test_type, total_primes, thread_count, verbose):
-    """使用指定的Python解释器运行测试"""
-    print(f"\n--- 开始测试: {test_type} ---")
+def run_test(python_path, test_type, total_primes, thread_count, args):
+    """
+    使用指定的Python解释器运行自定义测试
+    """
+    if not args.quiet:
+        print(f"\n--- 开始测试: {test_type} ---")
     
     cmd = [python_path, "src/test_worker.py", test_type, str(total_primes)]
     if thread_count:
         cmd.append(str(thread_count))
-    if verbose:
+    if args.verbose:
         cmd.append("--verbose")
     
-    print(f"  执行命令: {' '.join(cmd)}")
+    if not args.quiet:
+        print(f"  执行命令: {' '.join(cmd)}")
+        print(f"  超时时间: {args.timeout} 秒")
     
     try:
         # 使用 Popen 实时捕获输出
-        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        process = subprocess.Popen(cmd, 
+                                   stdout=subprocess.PIPE, 
+                                   stderr=subprocess.PIPE, 
+                                   text=True,
+        )
         
-        stdout_output = ""
-        stderr_output = ""
+        stdout_output, stderr_output = "", ""
         
-        # 实时打印 stderr (线程日志)
+        # 实时打印线程日志
         while True:
             stderr_line = process.stderr.readline()
             if not stderr_line:
                 break
             stderr_output += stderr_line
-            if verbose:
+            if args.verbose:
                 print(f"  [线程日志] {stderr_line.strip()}")
         
-        # 获取 stdout 结果
         stdout_output, _ = process.communicate()
+
+        try:
+            stdout_output, stderr_output = process.communicate(timeout=args.timeout)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            print(f"  测试超时（超过 {args.timeout} 秒）")
+            return None
         
         if process.returncode != 0:
             print(f"  测试执行失败 (返回码 {process.returncode}):")
@@ -70,87 +119,121 @@ def run_test(python_path, test_type, total_primes, thread_count, verbose):
                 print(f"    标准错误: {stderr_output}")
             return None
 
-        # 解析JSON结果
         data = json.loads(stdout_output)
         
         if data.get("error"):
             print(f"  测试失败: {data['error']}")
             return None
         
-        print(f"  测试成功:")
-        print(f"    耗时: {data['duration']:.2f} 秒")
-        print(f"    找到质数数量: {data['prime_count']}")
-        if data['last_prime']:
-            print(f"    最后一个质数: {data['last_prime']}")
+        if not args.quiet:
+            print(f"  测试成功:")
+            print(f"    耗时: {data['duration']:.2f} 秒")
+            print(f"    找到质数数量: {data['prime_count']}")
+            if data['last_prime']:
+                print(f"    最后一个质数: {data['last_prime']}")
         
         return data
         
     except json.JSONDecodeError:
-        print(f"  无法解析测试结果。")
-        print(f"    原始输出: {stdout_output}")
+        print(f"  无法解析测试结果{stdout_output}")
         return None
     except Exception as e:
         print(f"  发生未知错误: {e}")
         return None
 
+def save_report(report, output_path):
+    """
+    保存测试报告
+    """
+    try:
+        report["generated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        with open(output_path, "w", encoding="utf-8") as f:
+            json.dump(report, f, ensure_ascii=False, indent=2)
+        print(f"\n测试报告已保存到: {os.path.abspath(output_path)}")
+    except Exception as e:
+        print(f"\n保存报告失败: {e}")
 def main():
     args = parse_arguments()
 
-    print("=" * 60)
-    print("Python 多线程性能基准测试工具")
-    print("=" * 60)
-    if args.verbose:
-        print("** 已启用详细模式 (--verbose) **")
-    print("本工具将使用不同的Python解释器运行CPU密集型任务，")
-    print("以比较单线程、GIL多线程和NoGIL多线程的性能差异。")
-    print("=" * 60)
+    if not args.quiet:
+        print("=" * 60)
+        print("Python 多线程性能基准测试工具")
+        print("=" * 60)
+        if args.verbose:
+            print("** 已启用详细模式 (--verbose) **")
+        if args.quiet:
+            print("** 已启用静默模式 (--quiet) **")
+        if args.skip:
+            print(f"** 跳过测试类型: {', '.join(args.skip)} **")
+        print("=" * 60)
 
-    print("\n[步骤1/2] 请提供Python解释器路径:")
-    python_paths = {
-        "single": get_python_path("  1. 用于单线程测试的Python路径: "),
-        "gil": get_python_path("  2. 用于GIL多线程测试的Python路径: "),
-        "nogil": get_python_path("  3. 用于NoGIL多线程测试的Python路径: ")
+    python_paths = {}
+    skip_tests = args.skip or []
+
+    if "single" not in skip_tests:
+        python_paths["single"] = get_python_path("  1. 用于单线程测试的Python路径: ", args)
+    if "gil" not in skip_tests:
+        python_paths["gil"] = get_python_path("  2. 用于GIL多线程测试的Python路径: ", args)
+    if "nogil" not in skip_tests:
+        python_paths["nogil"] = get_python_path("  3. 用于NoGIL多线程测试的Python路径: ", args)
+
+    if not args.quiet:
+        print("\n[测试参数]")
+        print(f"  质数总数: {args.num_primes}")
+        print(f"  多线程数: {args.threads}")
+        print(f"  单个测试超时: {args.timeout} 秒")
+        if args.output:
+            print(f"  报告输出文件: {args.output}")
+        print("=" * 60)
+
+        print("开始执行所有测试...")
+        print("=" * 60)
+    
+    test_results = {}
+    report_data = {
+        "parameters": {
+            "num_primes": args.num_primes,
+            "threads": args.threads,
+            "timeout": args.timeout,
+            "skip_tests": args.skip or []
+        },
+        "results": {}
     }
 
-    print("\n[步骤2/2] 请设置测试参数:")
-    while True:
-        try:
-            total_primes = int(input("  要寻找的质数总数 (例如 30000): ").strip())
-            if total_primes > 0:
-                break
-            print("  请输入一个正整数。")
-        except ValueError:
-            print("  请输入一个有效的整数。")
+    if "single" not in skip_tests:
+        result = run_test(python_paths["single"], "single", args.num_primes, None, args)
+        if result:
+            test_results["single"] = result
+            report_data["results"]["single"] = {
+                "python_path": python_paths["single"],
+                "duration": result["duration"],
+                "prime_count": result["prime_count"],
+                "last_prime": result["last_prime"]
+            }
     
-    while True:
-        try:
-            thread_count = int(input("  多线程测试的线程数 (例如 8): ").strip())
-            if thread_count > 0:
-                break
-            print("  请输入一个正整数。")
-        except ValueError:
-            print("  请输入一个有效的整数。")
-
-    print("\n" + "=" * 60)
-    print("开始执行所有测试...")
-    print("=" * 60)
-
-    test_results = {}
+    if "gil" not in skip_tests:
+        result = run_test(python_paths["gil"], "gil", args.num_primes, args.threads, args)
+        if result:
+            test_results["gil"] = result
+            report_data["results"]["gil"] = {
+                "python_path": python_paths["gil"],
+                "threads": args.threads,
+                "duration": result["duration"],
+                "prime_count": result["prime_count"],
+                "last_prime": result["last_prime"]
+            }
     
-    # 单线程测试
-    result = run_test(python_paths["single"], "single", total_primes, None, args.verbose)
-    if result:
-        test_results["single"] = result
-    
-    # GIL多线程测试
-    result = run_test(python_paths["gil"], "gil", total_primes, thread_count, args.verbose)
-    if result:
-        test_results["gil"] = result
-    
-    # NoGIL多线程测试
-    result = run_test(python_paths["nogil"], "nogil", total_primes, thread_count, args.verbose)
-    if result:
-        test_results["nogil"] = result
+    if "nogil" not in skip_tests:
+        result = run_test(python_paths["nogil"], "nogil", args.num_primes, args.threads, args)
+        if result:
+            test_results["nogil"] = result
+            report_data["results"]["nogil"] = {
+                "python_path": python_paths["nogil"],
+                "threads": args.threads,
+                "duration": result["duration"],
+                "prime_count": result["prime_count"],
+                "last_prime": result["last_prime"]
+            }
 
     print("\n" + "=" * 60)
     print("测试完成! 性能对比报告如下:")
@@ -158,25 +241,31 @@ def main():
 
     if not test_results:
         print("未能获取任何有效测试结果。")
+        if args.output:
+            save_report(report_data, args.output)
         return
 
+    # 相对速度
     fastest_duration = min([r["duration"] for r in test_results.values()])
 
+    # 报告表格
     print(f"\n{'测试类型':<15} {'耗时(秒)':<12} {'相对速度':<12} {'质数数量':<10}")
     print("-" * 60)
+
+    test_labels = {
+        "single": "单线程",
+        "gil": "GIL多线程",
+        "nogil": "NoGIL多线程"
+    }
     
     for test_type, result in test_results.items():
         duration = result["duration"]
         speedup = fastest_duration / duration
         prime_count = result["prime_count"]
-        
-        test_type_label = {
-            "single": "单线程",
-            "gil": "GIL多线程",
-            "nogil": "NoGIL多线程"
-        }.get(test_type, test_type)
-        
-        print(f"{test_type_label:<15} {duration:<12.2f} {speedup:<12.2f}x {prime_count:<10}")
+        print(f"{test_labels[test_type]:<15} {duration:<12.2f} {speedup:<12.2f}x {prime_count:<10}")
+
+    if args.output:
+        save_report(report_data, args.output)
 
     print("\n" + "=" * 60)
 

src/test_worker.py

@@ -5,16 +5,16 @@ import json
 import time
 import argparse
 
-# 获取当前脚本 (test_worker.py) 的绝对路径
+# 修复模块导入路径
 current_script_path = os.path.abspath(__file__)
-# 获取脚本所在目录 (src/) 的路径
 src_directory = os.path.dirname(current_script_path)
-# 获取项目根目录 (src/ 的父目录) 的路径
 project_root = os.path.dirname(src_directory)
-# 将项目根目录添加到 Python 的模块搜索路径中
 sys.path.insert(0, project_root)
 
 def parse_arguments():
+    """
+    解析命令行参数
+    """
     parser = argparse.ArgumentParser(description="测试工作线程")
     parser.add_argument("test_type")
     parser.add_argument("total_primes", type=int)
@@ -60,11 +60,9 @@ def main():
 
     except Exception as e:
         result["error"] = str(e)
-        # 在详细模式下，也将异常打印到 stderr
         if verbose:
             print(f"[错误] {e}", file=sys.stderr)
     
-    # 确保结果输出到stdout
     print(json.dumps(result))
 
 if __name__ == "__main__":