#!/usr/bin/env python3 """ AIVA Queen Performance Testing Suite ===================================== Comprehensive performance testing infrastructure for the AIVA Queen system. Provides tools for load testing, latency profiling, throughput measurement, memory profiling, concurrency testing, and benchmark execution. Components: - LoadTester: Generate and execute load patterns - LatencyProfiler: Profile response latency with percentiles - ThroughputMeasurer: Measure system throughput - MemoryProfiler: Profile memory usage and detect leaks - ConcurrencyTester: Test concurrent access patterns - BenchmarkRunner: Execute standardized benchmarks Usage: from test_02_performance_suite import ( LoadTester, LatencyProfiler, ThroughputMeasurer, MemoryProfiler, ConcurrencyTester, BenchmarkRunner ) # Run comprehensive benchmark runner = BenchmarkRunner() results = await runner.run_all_benchmarks(target_function) Author: AIVA Genesis System Version: 1.0.0 """ from __future__ import annotations import asyncio import gc import json import logging import math import multiprocessing import os import random import statistics import sys import threading import time import traceback import tracemalloc import uuid from abc import ABC, abstractmethod from collections import defaultdict, deque from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor from contextlib import contextmanager from dataclasses import dataclass, field, asdict from datetime import datetime, timedelta from enum import Enum, auto from functools import wraps from pathlib import Path from typing import ( Any, Callable, Coroutine, Dict, Generic, List, Optional, Set, Tuple, TypeVar, Union, ) # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("aiva.performance_suite") # Type variables T = TypeVar("T") R = TypeVar("R") # ============================================================================= # ENUMS AND CONSTANTS # ============================================================================= class LoadPattern(Enum): """Load patterns for testing.""" CONSTANT = "constant" # Steady load RAMP_UP = "ramp_up" # Gradually increasing RAMP_DOWN = "ramp_down" # Gradually decreasing SPIKE = "spike" # Sudden spike WAVE = "wave" # Sinusoidal pattern STEP = "step" # Step function RANDOM = "random" # Random load BURST = "burst" # Periodic bursts class TestStatus(Enum): """Status of a test run.""" PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" class MetricType(Enum): """Types of performance metrics.""" LATENCY = "latency" THROUGHPUT = "throughput" MEMORY = "memory" CPU = "cpu" ERRORS = "errors" CONCURRENT = "concurrent" # Default thresholds DEFAULT_LATENCY_THRESHOLD_MS = 100.0 DEFAULT_THROUGHPUT_THRESHOLD_RPS = 100.0 DEFAULT_MEMORY_THRESHOLD_MB = 512.0 DEFAULT_ERROR_RATE_THRESHOLD = 0.01 # 1% # ============================================================================= # DATA CLASSES # ============================================================================= @dataclass class LatencyMetrics: """Latency measurement results.""" min_ms: float = 0.0 max_ms: float = 0.0 mean_ms: float = 0.0 median_ms: float = 0.0 std_dev_ms: float = 0.0 p50_ms: float = 0.0 p75_ms: float = 0.0 p90_ms: float = 0.0 p95_ms: float = 0.0 p99_ms: float = 0.0 p999_ms: float = 0.0 sample_count: int = 0 def to_dict(self) -> Dict[str, Any]: return asdict(self) @classmethod def from_samples(cls, samples: List[float]) -> "LatencyMetrics": """Create metrics from latency samples.""" if not samples: return cls() sorted_samples = sorted(samples) n = len(sorted_samples) def percentile(p: float) -> float: k = (n - 1) * p / 100 f = int(k) c = min(f + 1, n - 1) return sorted_samples[f] + (k - f) * (sorted_samples[c] - sorted_samples[f]) return cls( min_ms=min(sorted_samples), max_ms=max(sorted_samples), mean_ms=statistics.mean(sorted_samples), median_ms=statistics.median(sorted_samples), std_dev_ms=statistics.stdev(sorted_samples) if n > 1 else 0.0, p50_ms=percentile(50), p75_ms=percentile(75), p90_ms=percentile(90), p95_ms=percentile(95), p99_ms=percentile(99), p999_ms=percentile(99.9), sample_count=n, ) @dataclass class ThroughputMetrics: """Throughput measurement results.""" total_requests: int = 0 successful_requests: int = 0 failed_requests: int = 0 duration_seconds: float = 0.0 requests_per_second: float = 0.0 peak_rps: float = 0.0 min_rps: float = 0.0 avg_rps: float = 0.0 bytes_transferred: int = 0 throughput_mbps: float = 0.0 @property def success_rate(self) -> float: if self.total_requests == 0: return 0.0 return self.successful_requests / self.total_requests @property def error_rate(self) -> float: return 1.0 - self.success_rate def to_dict(self) -> Dict[str, Any]: d = asdict(self) d["success_rate"] = self.success_rate d["error_rate"] = self.error_rate return d @dataclass class MemoryMetrics: """Memory usage metrics.""" initial_mb: float = 0.0 peak_mb: float = 0.0 final_mb: float = 0.0 growth_mb: float = 0.0 gc_collections: int = 0 allocation_count: int = 0 top_allocations: List[Dict[str, Any]] = field(default_factory=list) potential_leaks: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: return asdict(self) @dataclass class ConcurrencyMetrics: """Concurrency test metrics.""" max_concurrent: int = 0 actual_concurrent: int = 0 contention_events: int = 0 deadlock_detected: bool = False race_conditions: int = 0 avg_wait_time_ms: float = 0.0 max_wait_time_ms: float = 0.0 def to_dict(self) -> Dict[str, Any]: return asdict(self) @dataclass class TestResult: """Result of a performance test.""" test_name: str test_type: str status: TestStatus start_time: datetime end_time: datetime duration_seconds: float latency: Optional[LatencyMetrics] = None throughput: Optional[ThroughputMetrics] = None memory: Optional[MemoryMetrics] = None concurrency: Optional[ConcurrencyMetrics] = None errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) passed_thresholds: Dict[str, bool] = field(default_factory=dict) metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { "test_name": self.test_name, "test_type": self.test_type, "status": self.status.value, "start_time": self.start_time.isoformat(), "end_time": self.end_time.isoformat(), "duration_seconds": self.duration_seconds, "latency": self.latency.to_dict() if self.latency else None, "throughput": self.throughput.to_dict() if self.throughput else None, "memory": self.memory.to_dict() if self.memory else None, "concurrency": self.concurrency.to_dict() if self.concurrency else None, "errors": self.errors, "warnings": self.warnings, "passed_thresholds": self.passed_thresholds, "metadata": self.metadata, } @property def is_passed(self) -> bool: return ( self.status == TestStatus.COMPLETED and len(self.errors) == 0 and all(self.passed_thresholds.values()) ) @dataclass class BenchmarkConfig: """Configuration for benchmarks.""" warmup_iterations: int = 10 min_iterations: int = 100 max_iterations: int = 10000 min_duration_seconds: float = 5.0 max_duration_seconds: float = 60.0 confidence_level: float = 0.95 target_relative_error: float = 0.05 gc_between_runs: bool = True parallel_workers: int = 0 # 0 = auto timeout_seconds: float = 300.0 # ============================================================================= # LOAD TESTER # ============================================================================= class LoadTester: """ Generate and execute load patterns against a target system. Supports various load patterns including constant, ramp-up, spike, and custom patterns for realistic load testing. """ def __init__( self, target: Callable[..., Any], pattern: LoadPattern = LoadPattern.CONSTANT, base_rate: float = 10.0, # requests per second duration_seconds: float = 60.0, max_workers: int = 0, ): self.target = target self.pattern = pattern self.base_rate = base_rate self.duration_seconds = duration_seconds self.max_workers = max_workers or multiprocessing.cpu_count() * 2 self._results: List[Tuple[float, bool, float]] = [] # (timestamp, success, latency_ms) self._running = False self._lock = threading.Lock() self._start_time: Optional[float] = None self._executor: Optional[ThreadPoolExecutor] = None def _get_rate_at_time(self, elapsed_seconds: float) -> float: """Calculate target rate based on pattern and elapsed time.""" progress = elapsed_seconds / self.duration_seconds if self.pattern == LoadPattern.CONSTANT: return self.base_rate elif self.pattern == LoadPattern.RAMP_UP: return self.base_rate * progress elif self.pattern == LoadPattern.RAMP_DOWN: return self.base_rate * (1 - progress) elif self.pattern == LoadPattern.SPIKE: # Spike at 50% duration spike_center = 0.5 spike_width = 0.1 if abs(progress - spike_center) < spike_width: return self.base_rate * 5 # 5x spike return self.base_rate elif self.pattern == LoadPattern.WAVE: # Sinusoidal pattern return self.base_rate * (1 + 0.5 * math.sin(progress * math.pi * 4)) elif self.pattern == LoadPattern.STEP: # Step function with 4 steps step = int(progress * 4) return self.base_rate * (step + 1) / 2 elif self.pattern == LoadPattern.RANDOM: return self.base_rate * random.uniform(0.5, 1.5) elif self.pattern == LoadPattern.BURST: # Burst every 10 seconds cycle = elapsed_seconds % 10 if cycle < 2: # 2-second burst return self.base_rate * 3 return self.base_rate * 0.3 # Low load between bursts return self.base_rate async def _execute_request( self, request_id: str, *args, **kwargs, ) -> Tuple[bool, float, Optional[Exception]]: """Execute a single request and measure latency.""" start_time = time.time() try: if asyncio.iscoroutinefunction(self.target): await self.target(*args, **kwargs) else: loop = asyncio.get_event_loop() await loop.run_in_executor( self._executor, lambda: self.target(*args, **kwargs) ) latency_ms = (time.time() - start_time) * 1000 return True, latency_ms, None except Exception as e: latency_ms = (time.time() - start_time) * 1000 return False, latency_ms, e async def run(self, *args, **kwargs) -> TestResult: """Run the load test.""" test_name = f"load_test_{self.pattern.value}_{self.base_rate}rps" start_time = datetime.now() self._start_time = time.time() self._running = True self._results.clear() self._executor = ThreadPoolExecutor(max_workers=self.max_workers) errors: List[str] = [] warnings: List[str] = [] try: logger.info( f"Starting load test: pattern={self.pattern.value}, " f"base_rate={self.base_rate} rps, duration={self.duration_seconds}s" ) # Scheduler loop tasks: List[asyncio.Task] = [] request_count = 0 last_check = time.time() requests_this_second = 0 while self._running: elapsed = time.time() - self._start_time if elapsed >= self.duration_seconds: break target_rate = self._get_rate_at_time(elapsed) current_time = time.time() # Calculate how many requests to send this iteration time_since_check = current_time - last_check expected_requests = target_rate * time_since_check if requests_this_second < expected_requests: request_id = f"req_{request_count}" task = asyncio.create_task( self._execute_request(request_id, *args, **kwargs) ) tasks.append(task) request_count += 1 requests_this_second += 1 # Reset counter each second if current_time - last_check >= 1.0: last_check = current_time requests_this_second = 0 # Yield control await asyncio.sleep(0.001) # Wait for all pending requests if tasks: results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, Exception): errors.append(str(result)) with self._lock: self._results.append((time.time(), False, 0.0)) elif isinstance(result, tuple): success, latency_ms, exc = result with self._lock: self._results.append((time.time(), success, latency_ms)) if exc: errors.append(str(exc)) except Exception as e: errors.append(f"Load test failed: {e}") logger.exception("Load test error") finally: self._running = False if self._executor: self._executor.shutdown(wait=False) end_time = datetime.now() duration = (end_time - start_time).total_seconds() # Calculate metrics latency_samples = [r[2] for r in self._results if r[1]] latency_metrics = LatencyMetrics.from_samples(latency_samples) successful = sum(1 for r in self._results if r[1]) failed = sum(1 for r in self._results if not r[1]) throughput_metrics = ThroughputMetrics( total_requests=len(self._results), successful_requests=successful, failed_requests=failed, duration_seconds=duration, requests_per_second=len(self._results) / duration if duration > 0 else 0, avg_rps=successful / duration if duration > 0 else 0, ) # Check thresholds passed_thresholds = { "latency_p99": latency_metrics.p99_ms <= DEFAULT_LATENCY_THRESHOLD_MS * 10, "error_rate": throughput_metrics.error_rate <= DEFAULT_ERROR_RATE_THRESHOLD, "min_throughput": throughput_metrics.requests_per_second >= self.base_rate * 0.8, } return TestResult( test_name=test_name, test_type="load_test", status=TestStatus.COMPLETED if not errors else TestStatus.FAILED, start_time=start_time, end_time=end_time, duration_seconds=duration, latency=latency_metrics, throughput=throughput_metrics, errors=errors[:100], # Limit errors warnings=warnings, passed_thresholds=passed_thresholds, metadata={ "pattern": self.pattern.value, "base_rate": self.base_rate, "max_workers": self.max_workers, }, ) def stop(self) -> None: """Stop the load test.""" self._running = False def get_current_rate(self) -> float: """Get current request rate.""" if not self._start_time: return 0.0 elapsed = time.time() - self._start_time return self._get_rate_at_time(elapsed) # ============================================================================= # LATENCY PROFILER # ============================================================================= class LatencyProfiler: """ Profile response latency with detailed statistical analysis. Provides percentile calculations, distribution analysis, and trend detection for latency measurements. """ def __init__( self, window_size: int = 1000, histogram_buckets: int = 50, ): self.window_size = window_size self.histogram_buckets = histogram_buckets self._samples: deque = deque(maxlen=window_size) self._timestamps: deque = deque(maxlen=window_size) self._lock = threading.Lock() def record(self, latency_ms: float) -> None: """Record a latency sample.""" with self._lock: self._samples.append(latency_ms) self._timestamps.append(time.time()) @contextmanager def measure(self): """Context manager for measuring latency.""" start = time.time() yield latency_ms = (time.time() - start) * 1000 self.record(latency_ms) async def profile( self, target: Callable[..., Any], iterations: int = 100, warmup: int = 10, *args, **kwargs, ) -> LatencyMetrics: """ Profile a target function's latency. Args: target: Function to profile iterations: Number of iterations warmup: Warmup iterations (not counted) *args, **kwargs: Arguments for target function """ # Warmup for _ in range(warmup): try: if asyncio.iscoroutinefunction(target): await target(*args, **kwargs) else: target(*args, **kwargs) except Exception: pass # Actual measurement samples = [] for _ in range(iterations): start = time.time() try: if asyncio.iscoroutinefunction(target): await target(*args, **kwargs) else: target(*args, **kwargs) latency_ms = (time.time() - start) * 1000 samples.append(latency_ms) self.record(latency_ms) except Exception: pass return LatencyMetrics.from_samples(samples) def get_metrics(self) -> LatencyMetrics: """Get current latency metrics.""" with self._lock: return LatencyMetrics.from_samples(list(self._samples)) def get_histogram(self) -> Dict[str, Any]: """Get latency histogram.""" with self._lock: if not self._samples: return {"buckets": [], "counts": [], "min": 0, "max": 0} samples = list(self._samples) min_val = min(samples) max_val = max(samples) range_val = max_val - min_val if range_val == 0: return { "buckets": [min_val], "counts": [len(samples)], "min": min_val, "max": max_val, } bucket_size = range_val / self.histogram_buckets buckets = [] counts = [] for i in range(self.histogram_buckets): bucket_start = min_val + i * bucket_size bucket_end = bucket_start + bucket_size count = sum(1 for s in samples if bucket_start <= s < bucket_end) buckets.append(bucket_start) counts.append(count) return { "buckets": buckets, "counts": counts, "bucket_size": bucket_size, "min": min_val, "max": max_val, } def detect_trend(self, window: int = 50) -> Dict[str, Any]: """Detect latency trend (increasing/decreasing/stable).""" with self._lock: if len(self._samples) < window * 2: return {"trend": "insufficient_data", "slope": 0.0} samples = list(self._samples) first_half = samples[-window * 2:-window] second_half = samples[-window:] first_avg = statistics.mean(first_half) second_avg = statistics.mean(second_half) change_pct = (second_avg - first_avg) / first_avg if first_avg > 0 else 0 if change_pct > 0.1: trend = "increasing" elif change_pct < -0.1: trend = "decreasing" else: trend = "stable" return { "trend": trend, "slope": change_pct, "first_half_avg": first_avg, "second_half_avg": second_avg, } def detect_outliers(self, threshold_std: float = 3.0) -> List[float]: """Detect latency outliers using standard deviation.""" with self._lock: if len(self._samples) < 10: return [] samples = list(self._samples) mean = statistics.mean(samples) std = statistics.stdev(samples) threshold = mean + threshold_std * std return [s for s in samples if s > threshold] def reset(self) -> None: """Reset all collected samples.""" with self._lock: self._samples.clear() self._timestamps.clear() # ============================================================================= # THROUGHPUT MEASURER # ============================================================================= class ThroughputMeasurer: """ Measure system throughput with sliding window tracking. Provides real-time RPS calculation, peak detection, and throughput trend analysis. """ def __init__( self, window_seconds: float = 60.0, granularity_seconds: float = 1.0, ): self.window_seconds = window_seconds self.granularity_seconds = granularity_seconds self._buckets: Dict[int, Dict[str, int]] = defaultdict( lambda: {"success": 0, "failure": 0, "bytes": 0} ) self._lock = threading.Lock() self._start_time = time.time() def _get_bucket(self, timestamp: float) -> int: """Get bucket index for a timestamp.""" return int(timestamp / self.granularity_seconds) def record_request( self, success: bool = True, bytes_transferred: int = 0, ) -> None: """Record a completed request.""" bucket = self._get_bucket(time.time()) with self._lock: if success: self._buckets[bucket]["success"] += 1 else: self._buckets[bucket]["failure"] += 1 self._buckets[bucket]["bytes"] += bytes_transferred # Cleanup old buckets self._cleanup_old_buckets() def _cleanup_old_buckets(self) -> None: """Remove buckets older than window.""" current_bucket = self._get_bucket(time.time()) window_buckets = int(self.window_seconds / self.granularity_seconds) min_bucket = current_bucket - window_buckets old_buckets = [b for b in self._buckets if b < min_bucket] for b in old_buckets: del self._buckets[b] async def measure( self, target: Callable[..., Any], duration_seconds: float = 10.0, *args, **kwargs, ) -> ThroughputMetrics: """ Measure throughput of a target function. Args: target: Function to measure duration_seconds: Measurement duration *args, **kwargs: Arguments for target """ start_time = time.time() total_requests = 0 successful = 0 failed = 0 bytes_total = 0 rps_samples = [] last_second = int(start_time) requests_this_second = 0 while time.time() - start_time < duration_seconds: try: start = time.time() if asyncio.iscoroutinefunction(target): result = await target(*args, **kwargs) else: result = target(*args, **kwargs) successful += 1 total_requests += 1 if isinstance(result, (bytes, str)): bytes_total += len(result) self.record_request(success=True, bytes_transferred=len(str(result)) if result else 0) except Exception: failed += 1 total_requests += 1 self.record_request(success=False) # Track RPS per second current_second = int(time.time()) if current_second != last_second: rps_samples.append(requests_this_second) requests_this_second = 1 last_second = current_second else: requests_this_second += 1 duration = time.time() - start_time return ThroughputMetrics( total_requests=total_requests, successful_requests=successful, failed_requests=failed, duration_seconds=duration, requests_per_second=total_requests / duration if duration > 0 else 0, peak_rps=max(rps_samples) if rps_samples else 0, min_rps=min(rps_samples) if rps_samples else 0, avg_rps=statistics.mean(rps_samples) if rps_samples else 0, bytes_transferred=bytes_total, throughput_mbps=(bytes_total / duration / 1_000_000) * 8 if duration > 0 else 0, ) def get_current_rps(self) -> float: """Get current requests per second.""" current_bucket = self._get_bucket(time.time()) with self._lock: bucket_data = self._buckets.get(current_bucket - 1, {"success": 0, "failure": 0}) return (bucket_data["success"] + bucket_data["failure"]) / self.granularity_seconds def get_metrics(self) -> ThroughputMetrics: """Get current throughput metrics.""" with self._lock: self._cleanup_old_buckets() total_success = sum(b["success"] for b in self._buckets.values()) total_failure = sum(b["failure"] for b in self._buckets.values()) total_bytes = sum(b["bytes"] for b in self._buckets.values()) total = total_success + total_failure # Calculate RPS per bucket rps_values = [ (b["success"] + b["failure"]) / self.granularity_seconds for b in self._buckets.values() ] return ThroughputMetrics( total_requests=total, successful_requests=total_success, failed_requests=total_failure, duration_seconds=self.window_seconds, requests_per_second=total / self.window_seconds if self.window_seconds > 0 else 0, peak_rps=max(rps_values) if rps_values else 0, min_rps=min(rps_values) if rps_values else 0, avg_rps=statistics.mean(rps_values) if rps_values else 0, bytes_transferred=total_bytes, ) def reset(self) -> None: """Reset all measurements.""" with self._lock: self._buckets.clear() # ============================================================================= # MEMORY PROFILER # ============================================================================= class MemoryProfiler: """ Profile memory usage and detect potential memory leaks. Uses tracemalloc for detailed allocation tracking and provides leak detection through growth analysis. """ def __init__( self, track_allocations: bool = True, top_allocations_count: int = 10, growth_threshold_mb: float = 10.0, ): self.track_allocations = track_allocations self.top_allocations_count = top_allocations_count self.growth_threshold_mb = growth_threshold_mb self._snapshots: List[Tuple[float, tracemalloc.Snapshot]] = [] self._initial_memory: Optional[float] = None self._peak_memory: float = 0.0 @staticmethod def _get_current_memory_mb() -> float: """Get current memory usage in MB.""" import resource try: return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 except Exception: # Fallback for Windows try: import psutil return psutil.Process().memory_info().rss / (1024 * 1024) except ImportError: return 0.0 def start(self) -> None: """Start memory profiling.""" if self.track_allocations: tracemalloc.start() self._initial_memory = self._get_current_memory_mb() self._peak_memory = self._initial_memory def stop(self) -> MemoryMetrics: """Stop profiling and return metrics.""" final_memory = self._get_current_memory_mb() self._peak_memory = max(self._peak_memory, final_memory) top_allocations = [] potential_leaks = [] if self.track_allocations: try: snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics("lineno")[:self.top_allocations_count] for stat in top_stats: top_allocations.append({ "file": str(stat.traceback), "size_kb": stat.size / 1024, "count": stat.count, }) # Detect potential leaks by comparing snapshots if len(self._snapshots) >= 2: first_snap = self._snapshots[0][1] last_snap = self._snapshots[-1][1] diff = last_snap.compare_to(first_snap, "lineno") for stat in diff[:5]: if stat.size_diff > 0: potential_leaks.append( f"{stat.traceback}: +{stat.size_diff / 1024:.1f}KB" ) except Exception as e: logger.warning(f"Failed to get allocation stats: {e}") finally: tracemalloc.stop() return MemoryMetrics( initial_mb=self._initial_memory or 0.0, peak_mb=self._peak_memory, final_mb=final_memory, growth_mb=final_memory - (self._initial_memory or 0.0), gc_collections=sum(gc.get_count()), allocation_count=len(top_allocations), top_allocations=top_allocations, potential_leaks=potential_leaks, ) def take_snapshot(self) -> None: """Take a memory snapshot for comparison.""" if self.track_allocations and tracemalloc.is_tracing(): snapshot = tracemalloc.take_snapshot() self._snapshots.append((time.time(), snapshot)) # Keep only last 10 snapshots if len(self._snapshots) > 10: self._snapshots = self._snapshots[-10:] def update_peak(self) -> None: """Update peak memory usage.""" current = self._get_current_memory_mb() self._peak_memory = max(self._peak_memory, current) async def profile( self, target: Callable[..., Any], iterations: int = 100, *args, **kwargs, ) -> MemoryMetrics: """ Profile memory usage of a target function. Args: target: Function to profile iterations: Number of iterations *args, **kwargs: Arguments for target """ self.start() try: for i in range(iterations): if asyncio.iscoroutinefunction(target): await target(*args, **kwargs) else: target(*args, **kwargs) # Take snapshots periodically if i % (iterations // 5) == 0: self.take_snapshot() self.update_peak() # Force GC occasionally if i % (iterations // 3) == 0: gc.collect() finally: return self.stop() # ============================================================================= # CONCURRENCY TESTER # ============================================================================= class ConcurrencyTester: """ Test concurrent access patterns and detect issues. Detects potential race conditions, deadlocks, and measures contention levels under concurrent load. """ def __init__( self, max_concurrent: int = 100, timeout_seconds: float = 30.0, ): self.max_concurrent = max_concurrent self.timeout_seconds = timeout_seconds self._active_count = 0 self._peak_concurrent = 0 self._contention_events = 0 self._wait_times: List[float] = [] self._lock = threading.Lock() self._semaphore: Optional[asyncio.Semaphore] = None async def _execute_with_tracking( self, target: Callable[..., Any], task_id: int, shared_state: Dict[str, Any], *args, **kwargs, ) -> Tuple[bool, float, Optional[Exception]]: """Execute a task with concurrency tracking.""" wait_start = time.time() await self._semaphore.acquire() wait_time = (time.time() - wait_start) * 1000 if wait_time > 0.1: # More than 0.1ms wait with self._lock: self._contention_events += 1 self._wait_times.append(wait_time) with self._lock: self._active_count += 1 self._peak_concurrent = max(self._peak_concurrent, self._active_count) start_time = time.time() try: # Track shared state modifications for race detection state_before = shared_state.copy() if asyncio.iscoroutinefunction(target): await target(*args, **kwargs) else: loop = asyncio.get_event_loop() await loop.run_in_executor(None, target, *args) latency_ms = (time.time() - start_time) * 1000 # Check for unexpected state changes (potential race) state_after = shared_state.copy() for key in state_before: if key in state_after and state_before[key] != state_after.get(key): if shared_state.get("_race_detected"): shared_state["_race_count"] = shared_state.get("_race_count", 0) + 1 return True, latency_ms, None except Exception as e: latency_ms = (time.time() - start_time) * 1000 return False, latency_ms, e finally: with self._lock: self._active_count -= 1 self._semaphore.release() async def test_concurrent_access( self, target: Callable[..., Any], num_tasks: int = 100, *args, **kwargs, ) -> TestResult: """ Test concurrent access to a target function. Args: target: Function to test num_tasks: Number of concurrent tasks *args, **kwargs: Arguments for target """ test_name = f"concurrency_test_{num_tasks}_tasks" start_time = datetime.now() errors: List[str] = [] self._semaphore = asyncio.Semaphore(self.max_concurrent) self._active_count = 0 self._peak_concurrent = 0 self._contention_events = 0 self._wait_times.clear() shared_state = {"counter": 0, "_race_detected": False, "_race_count": 0} latencies: List[float] = [] try: # Create all tasks tasks = [ self._execute_with_tracking(target, i, shared_state, *args, **kwargs) for i in range(num_tasks) ] # Execute with timeout results = await asyncio.wait_for( asyncio.gather(*tasks, return_exceptions=True), timeout=self.timeout_seconds, ) for result in results: if isinstance(result, Exception): errors.append(str(result)) elif isinstance(result, tuple): success, latency_ms, exc = result latencies.append(latency_ms) if exc: errors.append(str(exc)) except asyncio.TimeoutError: errors.append(f"Concurrency test timed out after {self.timeout_seconds}s") end_time = datetime.now() duration = (end_time - start_time).total_seconds() # Calculate metrics concurrency_metrics = ConcurrencyMetrics( max_concurrent=self.max_concurrent, actual_concurrent=self._peak_concurrent, contention_events=self._contention_events, deadlock_detected=False, # Would be True if timeout occurred race_conditions=shared_state.get("_race_count", 0), avg_wait_time_ms=statistics.mean(self._wait_times) if self._wait_times else 0, max_wait_time_ms=max(self._wait_times) if self._wait_times else 0, ) latency_metrics = LatencyMetrics.from_samples(latencies) passed_thresholds = { "no_deadlock": not concurrency_metrics.deadlock_detected, "no_race_conditions": concurrency_metrics.race_conditions == 0, "acceptable_contention": concurrency_metrics.contention_events < num_tasks * 0.5, } return TestResult( test_name=test_name, test_type="concurrency_test", status=TestStatus.COMPLETED if not errors else TestStatus.FAILED, start_time=start_time, end_time=end_time, duration_seconds=duration, latency=latency_metrics, concurrency=concurrency_metrics, errors=errors[:50], passed_thresholds=passed_thresholds, metadata={ "num_tasks": num_tasks, "max_concurrent": self.max_concurrent, "timeout": self.timeout_seconds, }, ) async def test_incremental_concurrency( self, target: Callable[..., Any], start_concurrent: int = 1, end_concurrent: int = 100, step: int = 10, requests_per_level: int = 50, *args, **kwargs, ) -> List[TestResult]: """Test with incrementally increasing concurrency levels.""" results = [] for concurrent in range(start_concurrent, end_concurrent + 1, step): self.max_concurrent = concurrent result = await self.test_concurrent_access( target, num_tasks=requests_per_level, *args, **kwargs ) result.metadata["concurrency_level"] = concurrent results.append(result) # Stop if we start seeing failures if result.concurrency and result.concurrency.deadlock_detected: break return results # ============================================================================= # BENCHMARK RUNNER # ============================================================================= class BenchmarkRunner: """ Execute standardized benchmarks against target functions. Provides a comprehensive benchmark suite including latency, throughput, memory, and concurrency benchmarks with configurable parameters and threshold validation. """ def __init__( self, config: Optional[BenchmarkConfig] = None, output_dir: Optional[Path] = None, ): self.config = config or BenchmarkConfig() self.output_dir = output_dir or Path("./benchmark_results") self._latency_profiler = LatencyProfiler() self._throughput_measurer = ThroughputMeasurer() self._memory_profiler = MemoryProfiler() self._concurrency_tester = ConcurrencyTester() self._results: List[TestResult] = [] self._start_time: Optional[datetime] = None async def benchmark_latency( self, target: Callable[..., Any], name: str = "latency_benchmark", *args, **kwargs, ) -> TestResult: """Run latency benchmark.""" start_time = datetime.now() # Warmup for _ in range(self.config.warmup_iterations): try: if asyncio.iscoroutinefunction(target): await target(*args, **kwargs) else: target(*args, **kwargs) except Exception: pass if self.config.gc_between_runs: gc.collect() # Measure latency_metrics = await self._latency_profiler.profile( target, iterations=self.config.min_iterations, warmup=0, # Already warmed up *args, **kwargs, ) end_time = datetime.now() duration = (end_time - start_time).total_seconds() passed_thresholds = { "p99_under_threshold": latency_metrics.p99_ms <= DEFAULT_LATENCY_THRESHOLD_MS, "median_under_half_threshold": latency_metrics.median_ms <= DEFAULT_LATENCY_THRESHOLD_MS / 2, "consistent": latency_metrics.std_dev_ms <= latency_metrics.mean_ms * 0.5, } return TestResult( test_name=name, test_type="latency_benchmark", status=TestStatus.COMPLETED, start_time=start_time, end_time=end_time, duration_seconds=duration, latency=latency_metrics, passed_thresholds=passed_thresholds, metadata={ "iterations": self.config.min_iterations, "warmup": self.config.warmup_iterations, }, ) async def benchmark_throughput( self, target: Callable[..., Any], name: str = "throughput_benchmark", duration: float = 10.0, *args, **kwargs, ) -> TestResult: """Run throughput benchmark.""" start_time = datetime.now() if self.config.gc_between_runs: gc.collect() throughput_metrics = await self._throughput_measurer.measure( target, duration_seconds=duration, *args, **kwargs, ) end_time = datetime.now() actual_duration = (end_time - start_time).total_seconds() passed_thresholds = { "min_rps": throughput_metrics.requests_per_second >= DEFAULT_THROUGHPUT_THRESHOLD_RPS, "error_rate": throughput_metrics.error_rate <= DEFAULT_ERROR_RATE_THRESHOLD, "stable": throughput_metrics.min_rps >= throughput_metrics.avg_rps * 0.5, } return TestResult( test_name=name, test_type="throughput_benchmark", status=TestStatus.COMPLETED, start_time=start_time, end_time=end_time, duration_seconds=actual_duration, throughput=throughput_metrics, passed_thresholds=passed_thresholds, metadata={"target_duration": duration}, ) async def benchmark_memory( self, target: Callable[..., Any], name: str = "memory_benchmark", iterations: int = 100, *args, **kwargs, ) -> TestResult: """Run memory benchmark.""" start_time = datetime.now() if self.config.gc_between_runs: gc.collect() memory_metrics = await self._memory_profiler.profile( target, iterations=iterations, *args, **kwargs, ) end_time = datetime.now() duration = (end_time - start_time).total_seconds() passed_thresholds = { "peak_under_threshold": memory_metrics.peak_mb <= DEFAULT_MEMORY_THRESHOLD_MB, "no_major_growth": memory_metrics.growth_mb <= self._memory_profiler.growth_threshold_mb, "no_detected_leaks": len(memory_metrics.potential_leaks) == 0, } return TestResult( test_name=name, test_type="memory_benchmark", status=TestStatus.COMPLETED, start_time=start_time, end_time=end_time, duration_seconds=duration, memory=memory_metrics, passed_thresholds=passed_thresholds, metadata={"iterations": iterations}, ) async def benchmark_concurrency( self, target: Callable[..., Any], name: str = "concurrency_benchmark", num_tasks: int = 100, *args, **kwargs, ) -> TestResult: """Run concurrency benchmark.""" if self.config.gc_between_runs: gc.collect() return await self._concurrency_tester.test_concurrent_access( target, num_tasks=num_tasks, *args, **kwargs, ) async def run_all_benchmarks( self, target: Callable[..., Any], name_prefix: str = "aiva_queen", *args, **kwargs, ) -> Dict[str, TestResult]: """Run all benchmark types and return results.""" self._start_time = datetime.now() results = {} logger.info("Starting comprehensive benchmark suite") # Latency benchmark logger.info("Running latency benchmark...") results["latency"] = await self.benchmark_latency( target, f"{name_prefix}_latency", *args, **kwargs ) self._results.append(results["latency"]) # Throughput benchmark logger.info("Running throughput benchmark...") results["throughput"] = await self.benchmark_throughput( target, f"{name_prefix}_throughput", *args, **kwargs ) self._results.append(results["throughput"]) # Memory benchmark logger.info("Running memory benchmark...") results["memory"] = await self.benchmark_memory( target, f"{name_prefix}_memory", *args, **kwargs ) self._results.append(results["memory"]) # Concurrency benchmark logger.info("Running concurrency benchmark...") results["concurrency"] = await self.benchmark_concurrency( target, f"{name_prefix}_concurrency", *args, **kwargs ) self._results.append(results["concurrency"]) logger.info("Benchmark suite completed") return results def get_summary(self) -> Dict[str, Any]: """Get summary of all benchmark results.""" if not self._results: return {"status": "no_results"} passed = sum(1 for r in self._results if r.is_passed) failed = len(self._results) - passed return { "start_time": self._start_time.isoformat() if self._start_time else None, "total_tests": len(self._results), "passed": passed, "failed": failed, "pass_rate": passed / len(self._results) if self._results else 0, "total_duration_seconds": sum(r.duration_seconds for r in self._results), "results": [r.to_dict() for r in self._results], } def save_results(self, filename: Optional[str] = None) -> Path: """Save benchmark results to JSON file.""" self.output_dir.mkdir(parents=True, exist_ok=True) if not filename: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"benchmark_results_{timestamp}.json" filepath = self.output_dir / filename with open(filepath, "w") as f: json.dump(self.get_summary(), f, indent=2, default=str) logger.info(f"Results saved to {filepath}") return filepath def generate_report(self) -> str: """Generate a human-readable benchmark report.""" summary = self.get_summary() lines = [ "=" * 60, "AIVA QUEEN PERFORMANCE BENCHMARK REPORT", "=" * 60, "", f"Start Time: {summary.get('start_time', 'N/A')}", f"Total Tests: {summary['total_tests']}", f"Passed: {summary['passed']}", f"Failed: {summary['failed']}", f"Pass Rate: {summary['pass_rate'] * 100:.1f}%", f"Total Duration: {summary['total_duration_seconds']:.2f}s", "", "-" * 60, ] for result in self._results: lines.extend([ f"\nTest: {result.test_name}", f"Type: {result.test_type}", f"Status: {'PASSED' if result.is_passed else 'FAILED'}", f"Duration: {result.duration_seconds:.2f}s", ]) if result.latency: lines.extend([ "Latency Metrics:", f" Mean: {result.latency.mean_ms:.2f}ms", f" P50: {result.latency.p50_ms:.2f}ms", f" P95: {result.latency.p95_ms:.2f}ms", f" P99: {result.latency.p99_ms:.2f}ms", ]) if result.throughput: lines.extend([ "Throughput Metrics:", f" RPS: {result.throughput.requests_per_second:.2f}", f" Success Rate: {result.throughput.success_rate * 100:.2f}%", ]) if result.memory: lines.extend([ "Memory Metrics:", f" Peak: {result.memory.peak_mb:.2f}MB", f" Growth: {result.memory.growth_mb:.2f}MB", ]) if result.concurrency: lines.extend([ "Concurrency Metrics:", f" Peak Concurrent: {result.concurrency.actual_concurrent}", f" Contention Events: {result.concurrency.contention_events}", ]) if result.errors: lines.extend([ "Errors:", *[f" - {e}" for e in result.errors[:5]], ]) lines.append("-" * 60) lines.append("") lines.append("=" * 60) return "\n".join(lines) # ============================================================================= # DECORATORS AND UTILITIES # ============================================================================= def benchmark( name: Optional[str] = None, iterations: int = 100, warmup: int = 10, ): """Decorator for benchmarking functions.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: benchmark_name = name or func.__name__ @wraps(func) async def async_wrapper(*args, **kwargs) -> T: profiler = LatencyProfiler() # Warmup for _ in range(warmup): try: await func(*args, **kwargs) except Exception: pass # Measure samples = [] for _ in range(iterations): start = time.time() result = await func(*args, **kwargs) samples.append((time.time() - start) * 1000) metrics = LatencyMetrics.from_samples(samples) logger.info( f"Benchmark '{benchmark_name}': " f"mean={metrics.mean_ms:.2f}ms, " f"p99={metrics.p99_ms:.2f}ms" ) return result @wraps(func) def sync_wrapper(*args, **kwargs) -> T: samples = [] # Warmup for _ in range(warmup): try: func(*args, **kwargs) except Exception: pass # Measure for _ in range(iterations): start = time.time() result = func(*args, **kwargs) samples.append((time.time() - start) * 1000) metrics = LatencyMetrics.from_samples(samples) logger.info( f"Benchmark '{benchmark_name}': " f"mean={metrics.mean_ms:.2f}ms, " f"p99={metrics.p99_ms:.2f}ms" ) return result if asyncio.iscoroutinefunction(func): return async_wrapper return sync_wrapper return decorator def assert_performance( max_latency_ms: Optional[float] = None, min_throughput_rps: Optional[float] = None, max_memory_mb: Optional[float] = None, ): """Decorator that asserts performance meets thresholds.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) async def async_wrapper(*args, **kwargs) -> T: start_mem = MemoryProfiler._get_current_memory_mb() start_time = time.time() result = await func(*args, **kwargs) latency_ms = (time.time() - start_time) * 1000 end_mem = MemoryProfiler._get_current_memory_mb() if max_latency_ms and latency_ms > max_latency_ms: raise AssertionError( f"Latency {latency_ms:.2f}ms exceeds threshold {max_latency_ms}ms" ) if max_memory_mb and (end_mem - start_mem) > max_memory_mb: raise AssertionError( f"Memory growth {end_mem - start_mem:.2f}MB exceeds threshold {max_memory_mb}MB" ) return result @wraps(func) def sync_wrapper(*args, **kwargs) -> T: start_mem = MemoryProfiler._get_current_memory_mb() start_time = time.time() result = func(*args, **kwargs) latency_ms = (time.time() - start_time) * 1000 end_mem = MemoryProfiler._get_current_memory_mb() if max_latency_ms and latency_ms > max_latency_ms: raise AssertionError( f"Latency {latency_ms:.2f}ms exceeds threshold {max_latency_ms}ms" ) if max_memory_mb and (end_mem - start_mem) > max_memory_mb: raise AssertionError( f"Memory growth {end_mem - start_mem:.2f}MB exceeds threshold {max_memory_mb}MB" ) return result if asyncio.iscoroutinefunction(func): return async_wrapper return sync_wrapper return decorator # ============================================================================= # MAIN - TESTING AND DEMONSTRATION # ============================================================================= async def demo_target_function(delay: float = 0.01) -> str: """Demo target function for testing.""" await asyncio.sleep(delay + random.uniform(0, delay)) return "success" async def main(): """Run performance suite demonstration.""" print("=" * 60) print("AIVA Queen Performance Testing Suite - Demo") print("=" * 60) # Test LoadTester print("\n--- Load Tester Demo ---") load_tester = LoadTester( target=demo_target_function, pattern=LoadPattern.RAMP_UP, base_rate=10.0, duration_seconds=5.0, ) result = await load_tester.run(delay=0.005) print(f"Load Test Status: {result.status.value}") print(f"Total Requests: {result.throughput.total_requests}") print(f"P99 Latency: {result.latency.p99_ms:.2f}ms") # Test LatencyProfiler print("\n--- Latency Profiler Demo ---") profiler = LatencyProfiler() metrics = await profiler.profile(demo_target_function, iterations=50, warmup=5) print(f"Mean Latency: {metrics.mean_ms:.2f}ms") print(f"P95 Latency: {metrics.p95_ms:.2f}ms") print(f"P99 Latency: {metrics.p99_ms:.2f}ms") # Test ThroughputMeasurer print("\n--- Throughput Measurer Demo ---") measurer = ThroughputMeasurer() throughput = await measurer.measure(demo_target_function, duration_seconds=3.0) print(f"RPS: {throughput.requests_per_second:.2f}") print(f"Success Rate: {throughput.success_rate * 100:.2f}%") # Test MemoryProfiler print("\n--- Memory Profiler Demo ---") mem_profiler = MemoryProfiler() memory = await mem_profiler.profile(demo_target_function, iterations=50) print(f"Peak Memory: {memory.peak_mb:.2f}MB") print(f"Memory Growth: {memory.growth_mb:.2f}MB") # Test ConcurrencyTester print("\n--- Concurrency Tester Demo ---") conc_tester = ConcurrencyTester(max_concurrent=20) conc_result = await conc_tester.test_concurrent_access( demo_target_function, num_tasks=50 ) print(f"Peak Concurrent: {conc_result.concurrency.actual_concurrent}") print(f"Contention Events: {conc_result.concurrency.contention_events}") # Test BenchmarkRunner print("\n--- Benchmark Runner Demo ---") runner = BenchmarkRunner() results = await runner.run_all_benchmarks(demo_target_function) print("\n" + runner.generate_report()) # Save results filepath = runner.save_results() print(f"\nResults saved to: {filepath}") if __name__ == "__main__": asyncio.run(main())