diff --git a/core/utils/__init__.py b/core/utils/__init__.py index e69de29..6d8b745 100644 --- a/core/utils/__init__.py +++ b/core/utils/__init__.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python3 +""" +工具函数包 +""" + +# 导出核心工具 +from .logger import logger +from .exceptions import * +from .json_utils import * +from .singleton import singleton +from .executor import run_in_thread_pool, initialize_executor +from .performance import ( + timeit, + profile, + aprofile, + memory_profile, + memory_profile_decorator, + performance_monitor, + PerformanceStats, + performance_stats, + global_stats +) + +__all__ = [ + 'logger', + 'timeit', + 'profile', + 'aprofile', + 'memory_profile', + 'memory_profile_decorator', + 'performance_monitor', + 'PerformanceStats', + 'performance_stats', + 'global_stats', + 'run_in_thread_pool', + 'initialize_executor', + 'singleton' +] diff --git a/core/utils/performance.py b/core/utils/performance.py new file mode 100644 index 0000000..9856f32 --- /dev/null +++ b/core/utils/performance.py @@ -0,0 +1,365 @@ +#!/usr/bin/env python3 +""" +性能分析工具模块 + +提供同步和异步函数的性能分析装饰器、上下文管理器和统计工具。 + +主要功能： +1. 函数执行时间分析（支持同步和异步） +2. 内存使用分析 +3. 性能统计和报告生成 +4. 低开销的生产环境监控 +""" + +import time +import asyncio +import functools +import logging +from typing import Dict, Any, Callable, Optional +import inspect + +# 尝试导入性能分析库 +try: + from pyinstrument import Profiler + from pyinstrument.renderers import HTMLRenderer + PYINSTRUMENT_AVAILABLE = True +except ImportError: + PYINSTRUMENT_AVAILABLE = False + +# 尝试导入内存分析库 +try: + from memory_profiler import memory_usage + MEMORY_PROFILER_AVAILABLE = True +except ImportError: + MEMORY_PROFILER_AVAILABLE = False + +from .logger import logger + + +class PerformanceStats: + """ + 性能统计工具类 + 用于收集和报告函数执行的性能指标 + """ + def __init__(self): + self.stats: Dict[str, Dict[str, Any]] = {} + + def record(self, func_name: str, duration: float, memory_used: Optional[float] = None): + """ + 记录函数执行的性能数据 + + Args: + func_name: 函数名称 + duration: 执行时间（秒） + memory_used: 使用的内存（MB），可选 + """ + if func_name not in self.stats: + self.stats[func_name] = { + "count": 0, + "total_time": 0.0, + "avg_time": 0.0, + "min_time": float('inf'), + "max_time": 0.0, + "total_memory": 0.0, + "avg_memory": 0.0 + } + + stat = self.stats[func_name] + stat["count"] += 1 + stat["total_time"] += duration + stat["avg_time"] = stat["total_time"] / stat["count"] + stat["min_time"] = min(stat["min_time"], duration) + stat["max_time"] = max(stat["max_time"], duration) + + if memory_used is not None: + stat["total_memory"] += memory_used + stat["avg_memory"] = stat["total_memory"] / stat["count"] + + def report(self) -> str: + """ + 生成性能统计报告 + + Returns: + 格式化的性能统计报告字符串 + """ + if not self.stats: + return "暂无性能统计数据" + + report = ["\n=== 性能统计报告 ===\n"] + report.append(f"{'函数名':<40} {'调用次数':<10} {'平均时间(ms)':<15} {'最长时间(ms)':<15} {'内存(MB)':<10}") + report.append("-" * 100) + + for func_name, stat in sorted(self.stats.items(), key=lambda x: x[1]["total_time"], reverse=True): + memory_str = f"{stat['avg_memory']:.2f}" if stat['avg_memory'] > 0 else "-" + report.append( + f"{func_name:<40} {stat['count']:<10} {stat['avg_time']*1000:<15.2f} " + f"{stat['max_time']*1000:<15.2f} {memory_str:<10}" + ) + + report.append("=" * 100) + return "\n".join(report) + + def reset(self): + """ + 重置性能统计数据 + """ + self.stats.clear() + + +# 创建全局性能统计实例 +performance_stats = PerformanceStats() + + +def timeit(func: Callable = None, *, log_level: int = logging.INFO, collect_stats: bool = True): + """ + 函数执行时间分析装饰器（支持同步和异步） + + Args: + func: 要装饰的函数 + log_level: 日志级别 + collect_stats: 是否收集到全局统计中 + + Returns: + 装饰后的函数 + """ + def decorator(func: Callable) -> Callable: + func_name = func.__qualname__ + is_coroutine = inspect.iscoroutinefunction(func) + + if is_coroutine: + @functools.wraps(func) + async def async_wrapper(*args, **kwargs): + start_time = time.perf_counter() + try: + result = await func(*args, **kwargs) + finally: + end_time = time.perf_counter() + duration = end_time - start_time + + if collect_stats: + performance_stats.record(func_name, duration) + + logger.log(log_level, f"[性能] {func_name} 执行时间: {duration*1000:.2f} ms") + + return result + + return async_wrapper + else: + @functools.wraps(func) + def sync_wrapper(*args, **kwargs): + start_time = time.perf_counter() + try: + result = func(*args, **kwargs) + finally: + end_time = time.perf_counter() + duration = end_time - start_time + + if collect_stats: + performance_stats.record(func_name, duration) + + logger.log(log_level, f"[性能] {func_name} 执行时间: {duration*1000:.2f} ms") + + return result + + return sync_wrapper + + if func is None: + return decorator + return decorator(func) + + +class profile: + """ + 性能分析上下文管理器 + 使用 pyinstrument 进行详细的性能分析 + """ + def __init__(self, enabled: bool = True, output_file: Optional[str] = None): + """ + Args: + enabled: 是否启用分析 + output_file: 分析结果输出文件路径（HTML格式） + """ + self.enabled = enabled + self.output_file = output_file + self.profiler = None + + def __enter__(self): + if self.enabled and PYINSTRUMENT_AVAILABLE: + self.profiler = Profiler() + self.profiler.start() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self.enabled and PYINSTRUMENT_AVAILABLE and self.profiler: + self.profiler.stop() + + # 输出到日志 + logger.info(f"[性能分析] {self.profiler.print()}") + + # 如果指定了输出文件，保存为HTML + if self.output_file: + try: + html = self.profiler.render(HTMLRenderer()) + with open(self.output_file, 'w', encoding='utf-8') as f: + f.write(html) + logger.info(f"[性能分析] 报告已保存到: {self.output_file}") + except Exception as e: + logger.error(f"[性能分析] 保存报告失败: {e}") + + +async def aprofile(func: Callable, *args, **kwargs): + """ + 异步函数性能分析 + + Args: + func: 要分析的异步函数 + *args: 函数参数 + **kwargs: 函数关键字参数 + + Returns: + 函数执行结果 + """ + if not PYINSTRUMENT_AVAILABLE: + logger.warning("[性能分析] pyinstrument 未安装，无法进行详细分析") + return await func(*args, **kwargs) + + profiler = Profiler() + profiler.start() + + try: + result = await func(*args, **kwargs) + finally: + profiler.stop() + logger.info(f"[性能分析] {profiler.print()}") + + return result + + +class memory_profile: + """ + 内存分析上下文管理器 + """ + def __init__(self, interval: float = 0.1, enabled: bool = True): + """ + Args: + interval: 内存采样间隔（秒） + enabled: 是否启用内存分析 + """ + self.interval = interval + self.enabled = enabled + self.memory_start = 0.0 + self.memory_end = 0.0 + + def __enter__(self): + if self.enabled and MEMORY_PROFILER_AVAILABLE: + self.memory_start = memory_usage()[0] + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self.enabled and MEMORY_PROFILER_AVAILABLE: + self.memory_end = memory_usage()[0] + memory_used = self.memory_end - self.memory_start + logger.info(f"[内存分析] 使用内存: {memory_used:.2f} MB") + + +def memory_profile_decorator(func: Callable = None, *, interval: float = 0.1): + """ + 内存分析装饰器（支持同步函数） + + Args: + func: 要装饰的函数 + interval: 内存采样间隔 + + Returns: + 装饰后的函数 + """ + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + if not MEMORY_PROFILER_AVAILABLE: + return func(*args, **kwargs) + + mem_usage = memory_usage( + (func, args, kwargs), + interval=interval, + timeout=None, + include_children=False + ) + + max_memory = max(mem_usage) + logger.info(f"[内存分析] {func.__qualname__} 最大内存使用: {max_memory:.2f} MB") + return func(*args, **kwargs) + + return wrapper + + if func is None: + return decorator + return decorator(func) + + +def performance_monitor(func: Callable = None, *, threshold: float = 1.0): + """ + 性能监控装饰器 + 仅当函数执行时间超过阈值时记录日志 + 适合生产环境使用 + + Args: + func: 要装饰的函数 + threshold: 时间阈值（秒） + + Returns: + 装饰后的函数 + """ + def decorator(func: Callable) -> Callable: + func_name = func.__qualname__ + is_coroutine = inspect.iscoroutinefunction(func) + + if is_coroutine: + @functools.wraps(func) + async def async_wrapper(*args, **kwargs): + start_time = time.perf_counter() + result = await func(*args, **kwargs) + end_time = time.perf_counter() + duration = end_time - start_time + + if duration > threshold: + logger.warning(f"[性能监控] {func_name} 执行时间过长: {duration*1000:.2f} ms (阈值: {threshold*1000:.2f} ms)") + + return result + + return async_wrapper + else: + @functools.wraps(func) + def sync_wrapper(*args, **kwargs): + start_time = time.perf_counter() + result = func(*args, **kwargs) + end_time = time.perf_counter() + duration = end_time - start_time + + if duration > threshold: + logger.warning(f"[性能监控] {func_name} 执行时间过长: {duration*1000:.2f} ms (阈值: {threshold*1000:.2f} ms)") + + return result + + return sync_wrapper + + if func is None: + return decorator + return decorator(func) + + +# 全局实例 +global_stats = PerformanceStats() + + +__all__ = [ + 'timeit', + 'profile', + 'aprofile', + 'memory_profile', + 'memory_profile_decorator', + 'performance_monitor', + 'PerformanceStats', + 'performance_stats', + 'global_stats' +] diff --git a/core/utils/singleton.py b/core/utils/singleton.py index 94a7c93..02b5931 100644 --- a/core/utils/singleton.py +++ b/core/utils/singleton.py @@ -2,6 +2,7 @@ 通用单例模式基类 """ from typing import Any, Optional, Type, TypeVar +import functools T = TypeVar('T') @@ -38,3 +39,29 @@ class Singleton: if self._initialized: return self._initialized = True + + +def singleton(cls: Type[T]) -> Type[T]: + """ + 单例装饰器 + + 将普通类转换为单例类，确保整个应用程序中只有一个实例。 + + Args: + cls: 要转换为单例的类 + + Returns: + Type[T]: 单例类 + """ + _instance: Optional[T] = None + _initialized: bool = False + + @functools.wraps(cls) + def wrapper(*args: Any, **kwargs: Any) -> T: + nonlocal _instance, _initialized + + if _instance is None: + _instance = cls(*args, **kwargs) + return _instance + + return wrapper diff --git a/performance_config_example.py b/performance_config_example.py new file mode 100644 index 0000000..f9a11f8 --- /dev/null +++ b/performance_config_example.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +""" +性能分析配置示例 + +展示如何在项目中配置和使用性能分析功能。 +""" + +# 配置性能分析的使用方式 +PERFORMANCE_CONFIG = { + # 全局性能分析开关 + 'enabled': True, + + # 详细性能分析开关（使用 pyinstrument） + 'detailed': False, + + # 内存分析开关 + 'memory': False, + + # 性能监控阈值（秒） + 'threshold': 0.5, + + # 性能报告输出文件 + 'output_file': 'performance_report.html', + + # 要监控的核心组件列表 + 'monitored_components': [ + 'core.ws.WS', + 'core.managers.plugin_manager', + 'core.managers.browser_manager', + 'core.utils.executor.CodeExecutor', + 'core.handlers.event_handler', + ] +} + + +def get_performance_config(): + """ + 获取性能分析配置 + + Returns: + dict: 性能分析配置 + """ + import os + import json + + # 从环境变量加载配置 + config = PERFORMANCE_CONFIG.copy() + + if os.environ.get('PERFORMANCE_PROFILE'): + config['detailed'] = os.environ['PERFORMANCE_PROFILE'] == '1' + + if os.environ.get('PERFORMANCE_MEMORY'): + config['memory'] = os.environ['PERFORMANCE_MEMORY'] == '1' + + if os.environ.get('PERFORMANCE_THRESHOLD'): + try: + config['threshold'] = float(os.environ['PERFORMANCE_THRESHOLD']) + except ValueError: + pass + + if os.environ.get('PERFORMANCE_OUTPUT'): + config['output_file'] = os.environ['PERFORMANCE_OUTPUT'] + + if os.environ.get('PERFORMANCE_STATS'): + config['enabled'] = os.environ['PERFORMANCE_STATS'] == '1' + + return config + + +if __name__ == "__main__": + # 打印当前配置 + print("当前性能分析配置:") + print("=" * 50) + config = get_performance_config() + for key, value in config.items(): + print(f"{key}: {value}") diff --git a/profile_main.py b/profile_main.py new file mode 100644 index 0000000..075ee82 --- /dev/null +++ b/profile_main.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +""" +性能分析入口文件 + +用于启动带有性能分析功能的应用程序。 + +使用方法： + python profile_main.py [options] + +选项： + -h, --help 显示帮助信息 + --profile, -p 启用详细性能分析（使用 pyinstrument） + --memory, -m 启用内存使用分析 + --output, -o FILE 性能分析报告输出文件（HTML格式） + --threshold, -t SEC 设置性能监控阈值（秒） + --stats, -s 在程序结束时输出性能统计报告 +""" + +import sys +import argparse +import os + +# 将项目根目录添加到 sys.path +ROOT_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, ROOT_DIR) + +# 解析命令行参数 +parser = argparse.ArgumentParser(description='性能分析入口文件') +parser.add_argument('--profile', '-p', action='store_true', help='启用详细性能分析（使用 pyinstrument）') +parser.add_argument('--memory', '-m', action='store_true', help='启用内存使用分析') +parser.add_argument('--output', '-o', type=str, default='performance_report.html', help='性能分析报告输出文件（HTML格式）') +parser.add_argument('--threshold', '-t', type=float, default=0.5, help='设置性能监控阈值（秒）') +parser.add_argument('--stats', '-s', action='store_true', help='在程序结束时输出性能统计报告') + +args = parser.parse_args() + +# 设置全局性能分析配置 +os.environ['PERFORMANCE_PROFILE'] = '1' if args.profile else '0' +os.environ['PERFORMANCE_MEMORY'] = '1' if args.memory else '0' +os.environ['PERFORMANCE_OUTPUT'] = args.output +os.environ['PERFORMANCE_THRESHOLD'] = str(args.threshold) +os.environ['PERFORMANCE_STATS'] = '1' if args.stats else '0' + +# 导入并运行主程序 +from core.utils.performance import profile, aprofile +from main import main +import asyncio + +async def main_with_profile(): + """ + 带有性能分析的主函数入口 + """ + if args.profile: + # 使用 pyinstrument 进行详细性能分析 + from pyinstrument import Profiler + from pyinstrument.renderers import HTMLRenderer + + profiler = Profiler() + profiler.start() + + try: + await main() + finally: + profiler.stop() + + # 输出分析结果到控制台 + print("\n" + "=" * 80) + print("性能分析结果") + print("=" * 80) + print(profiler.print()) + + # 保存HTML报告 + try: + html = profiler.render(HTMLRenderer()) + with open(args.output, 'w', encoding='utf-8') as f: + f.write(html) + print(f"\n性能分析报告已保存到: {args.output}") + except Exception as e: + print(f"\n保存性能分析报告失败: {e}") + else: + # 不使用详细分析，直接运行 + await main() + +if __name__ == "__main__": + try: + asyncio.run(main_with_profile()) + finally: + # 输出性能统计报告 + if args.stats: + from core.utils.performance import performance_stats + print("\n" + "=" * 80) + print("性能统计报告") + print("=" * 80) + print(performance_stats.report()) diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..2352bf4 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,4 @@ +# 开发依赖 +pyinstrument>=4.5.0 # 性能分析工具，支持异步代码 +memory-profiler>=0.61.0 # 内存分析工具 +psutil>=5.9.8 # 系统资源监控 diff --git a/test_performance_simple.py b/test_performance_simple.py new file mode 100644 index 0000000..5503473 --- /dev/null +++ b/test_performance_simple.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +""" +简单的性能分析功能测试脚本 +""" + +import asyncio +import time +from core.utils.performance import ( + timeit, + profile, + aprofile, + PerformanceStats, + performance_stats +) + + +print("=" * 80) +print("性能分析功能测试") +print("=" * 80) + +# 重置全局性能统计 +performance_stats.reset() + +# 测试1: 同步函数的时间测量 +@timeit +def sync_test(): + """同步测试函数""" + time.sleep(0.1) + return "sync done" + +# 测试2: 异步函数的时间测量 +@timeit +async def async_test(): + """异步测试函数""" + await asyncio.sleep(0.1) + return "async done" + +# 异步主函数 +async def main(): + # 同步函数测试 + print("执行同步函数...") + sync_result = sync_test() + print(f"同步函数结果: {sync_result}") + + # 异步函数测试 + print("\n执行异步函数...") + async_result = await async_test() + print(f"异步函数结果: {async_result}") + + # 测试3: 详细性能分析 + print("\n2. 测试性能分析上下文管理器:") + print("=" * 80) + + with profile(enabled=False): # 禁用实际分析以避免输出太多 + time.sleep(0.05) + print("性能分析上下文管理器测试完成") + + # 测试4: 性能统计报告 + print("\n3. 测试性能统计报告:") + print("=" * 80) + + # 执行多次函数调用 + for i in range(3): + sync_test() + await async_test() + + # 生成并打印性能报告 + print("\n性能统计报告:") + print(performance_stats.report()) + + +# 执行测试 +print("\n1. 测试时间测量装饰器:") +print("=" * 80) + +# 使用 asyncio.run() 执行异步主函数 +asyncio.run(main()) + +print("\n" + "=" * 80) +print("所有测试完成！") +print("=" * 80) diff --git a/tests/test_performance.py b/tests/test_performance.py new file mode 100644 index 0000000..5a1537b --- /dev/null +++ b/tests/test_performance.py @@ -0,0 +1,266 @@ +#!/usr/bin/env python3 +""" +性能分析工具测试 + +测试各种性能分析功能的正确性和可用性。 +""" + +import asyncio +import time +import pytest +from typing import Optional + +# 导入性能分析工具 +from core.utils.performance import ( + timeit, + profile, + aprofile, + memory_profile, + memory_profile_decorator, + performance_monitor, + PerformanceStats, + performance_stats +) + + +# 重置全局性能统计 +def setup_module(): + performance_stats.reset() + + +def teardown_module(): + performance_stats.reset() + + +class TestTimeitDecorator: + """测试 timeit 装饰器""" + + @timeit(log_level=20) # 使用 INFO 级别 + def test_sync_function(self): + """测试同步函数的时间测量""" + time.sleep(0.1) + return "done" + + @timeit(log_level=20) + async def test_async_function(self): + """测试异步函数的时间测量""" + await asyncio.sleep(0.1) + return "done" + + def test_sync_function_works(self): + """验证同步函数能正常执行""" + result = self.test_sync_function() + assert result == "done" + + @pytest.mark.asyncio + async def test_async_function_works(self): + """验证异步函数能正常执行""" + result = await self.test_async_function() + assert result == "done" + + +class TestProfileContextManager: + """测试 profile 上下文管理器""" + + def test_profile_sync_code(self): + """测试同步代码的性能分析""" + # 捕获标准输出 + import io + import sys + from contextlib import redirect_stdout + + f = io.StringIO() + with redirect_stdout(f): + with profile(enabled=False): # 禁用实际分析以提高测试速度 + time.sleep(0.01) + + output = f.getvalue() + # 应该没有输出（因为 enabled=False） + assert "性能分析" not in output + + @pytest.mark.asyncio + async def test_aprofile_async_function(self): + """测试异步函数的性能分析""" + async def async_test(): + await asyncio.sleep(0.01) + return "test" + + result = await aprofile(async_test) + assert result == "test" + + +class TestPerformanceMonitor: + """测试 performance_monitor 装饰器""" + + @performance_monitor(threshold=0.05) + def test_slow_sync_function(self): + """测试慢速同步函数的监控""" + time.sleep(0.1) # 超过阈值 + return "slow" + + @performance_monitor(threshold=0.05) + def test_fast_sync_function(self): + """测试快速同步函数的监控""" + time.sleep(0.01) # 低于阈值 + return "fast" + + @performance_monitor(threshold=0.05) + async def test_slow_async_function(self): + """测试慢速异步函数的监控""" + await asyncio.sleep(0.1) + return "slow_async" + + def test_slow_function_triggers_warning(self): + """验证慢速函数会触发警告""" + result = self.test_slow_sync_function() + assert result == "slow" + + def test_fast_function_no_warning(self): + """验证快速函数不会触发警告""" + result = self.test_fast_sync_function() + assert result == "fast" + + @pytest.mark.asyncio + async def test_slow_async_function_triggers_warning(self): + """验证慢速异步函数会触发警告""" + result = await self.test_slow_async_function() + assert result == "slow_async" + + +class TestMemoryAnalysis: + """测试内存分析功能""" + + def test_memory_profile_context_manager(self): + """测试内存分析上下文管理器""" + # 禁用内存分析以提高测试速度 + with memory_profile(enabled=False): + data = [i for i in range(1000)] + sum(data) + + @memory_profile_decorator + def test_memory_intensive_function(self): + """测试内存密集型函数""" + # 小数据集，避免测试耗时过长 + data = [i for i in range(1000)] + return sum(data) + + def test_memory_function_works(self): + """验证内存分析函数能正常执行""" + result = self.test_memory_intensive_function() + assert result == 499500 + + +class TestPerformanceStats: + """测试性能统计功能""" + + def test_stats_initialization(self): + """测试性能统计对象初始化""" + stats = PerformanceStats() + assert isinstance(stats, PerformanceStats) + assert stats.stats == {} + + def test_stats_record(self): + """测试记录性能数据""" + stats = PerformanceStats() + stats.record("test_func", 0.1) + + assert "test_func" in stats.stats + assert stats.stats["test_func"]["count"] == 1 + assert stats.stats["test_func"]["total_time"] == 0.1 + assert stats.stats["test_func"]["avg_time"] == 0.1 + + def test_stats_report(self): + """测试生成性能报告""" + stats = PerformanceStats() + stats.record("func1", 0.1) + stats.record("func2", 0.2) + + report = stats.report() + assert isinstance(report, str) + assert "func1" in report + assert "func2" in report + + def test_stats_reset(self): + """测试重置性能统计""" + stats = PerformanceStats() + stats.record("test_func", 0.1) + stats.reset() + + assert stats.stats == {} + + def test_global_stats_recording(self): + """测试全局性能统计记录""" + # 先重置全局统计 + performance_stats.reset() + + @timeit(collect_stats=True) + def test_func(): + time.sleep(0.01) + + test_func() + + # 验证是否记录了性能数据 + assert "test_func" in performance_stats.stats + assert performance_stats.stats["test_func"]["count"] == 1 + + +class TestIntegration: + """综合测试""" + + @pytest.mark.asyncio + async def test_combined_features(self): + """测试多种性能分析功能的组合使用""" + # 重置全局统计 + performance_stats.reset() + + @timeit(collect_stats=True) + @performance_monitor(threshold=0.05) + async def test_async_func(): + await asyncio.sleep(0.06) # 超过阈值 + return "combined" + + result = await test_async_func() + assert result == "combined" + + # 验证性能统计 + assert "test_async_func" in performance_stats.stats + assert performance_stats.stats["test_async_func"]["count"] == 1 + + +if __name__ == "__main__": + # 运行基本测试 + print("开始性能分析功能测试...") + + # 测试同步函数 + @timeit + def test_sync(): + time.sleep(0.1) + return "sync" + + # 测试异步函数 + @timeit + async def test_async(): + await asyncio.sleep(0.1) + return "async" + + # 测试性能监控 + @performance_monitor(threshold=0.05) + def slow_func(): + time.sleep(0.1) + return "slow" + + # 运行测试 + sync_result = test_sync() + async_result = asyncio.run(test_async()) + slow_result = slow_func() + + print(f"\n测试结果:") + print(f"sync_result: {sync_result}") + print(f"async_result: {async_result}") + print(f"slow_result: {slow_result}") + + # 输出性能统计报告 + print("\n性能统计报告:") + print(performance_stats.report()) + + print("\n性能分析功能测试完成！")