mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-16 01:40:15 +08:00
349 lines
13 KiB
Python
349 lines
13 KiB
Python
"""Comprehensive performance benchmark tests for AstrBot core modules.
|
||
|
||
This module provides performance benchmarks with scoring to track
|
||
performance regressions and improvements over time.
|
||
"""
|
||
|
||
import asyncio
|
||
import gc
|
||
import tracemalloc
|
||
from typing import Any
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass
|
||
|
||
import pytest
|
||
|
||
from astrbot.core.star.filter.command import CommandFilter, GreedyStr
|
||
|
||
|
||
@dataclass
|
||
class BenchmarkResult:
|
||
"""Result of a benchmark test."""
|
||
name: str
|
||
operation_count: int
|
||
total_time_ms: float
|
||
avg_time_ms: float
|
||
ops_per_second: float
|
||
memory_delta_kb: float
|
||
score: int # 0-100, 100 is best
|
||
|
||
def __str__(self) -> str:
|
||
return (
|
||
f"[{self.name}]\n"
|
||
f" Ops/sec: {self.ops_per_second:,.0f} | "
|
||
f"Avg: {self.avg_time_ms:.4f}ms | "
|
||
f"Memory: +{self.memory_delta_kb:.1f}KB | "
|
||
f"Score: {self.score}/100"
|
||
)
|
||
|
||
|
||
class PerformanceBenchmark:
|
||
"""Helper class to run benchmarks with memory tracking."""
|
||
|
||
def __init__(self, name: str, operations: int = 1000):
|
||
self.name = name
|
||
self.operations = operations
|
||
self.tracemalloc = tracemalloc
|
||
|
||
def run(self, func: Callable, *args, **kwargs) -> BenchmarkResult:
|
||
"""Run a function multiple times and measure performance.
|
||
|
||
Throughput and memory are measured in separate passes so that
|
||
``tracemalloc`` (which can be 7-10× slower) does not distort the
|
||
timing numbers.
|
||
"""
|
||
gc.collect()
|
||
|
||
# Pass 1 — throughput (no tracemalloc overhead)
|
||
start = asyncio.get_event_loop().time()
|
||
for _ in range(self.operations):
|
||
func(*args, **kwargs)
|
||
end = asyncio.get_event_loop().time()
|
||
|
||
total_time = (end - start) * 1000 # ms
|
||
avg_time = total_time / self.operations
|
||
ops_per_sec = self.operations / ((end - start) if (end - start) > 0 else 0.001)
|
||
|
||
# Pass 2 — memory delta (tracemalloc, fewer iterations)
|
||
gc.collect()
|
||
self.tracemalloc.start()
|
||
snapshot_before = self.tracemalloc.take_snapshot()
|
||
mem_ops = min(self.operations, 1000)
|
||
for _ in range(mem_ops):
|
||
func(*args, **kwargs)
|
||
snapshot_after = self.tracemalloc.take_snapshot()
|
||
self.tracemalloc.stop()
|
||
top_stats = snapshot_after.compare_to(snapshot_before, 'lineno')
|
||
memory_delta_kb = sum(stat.size_diff for stat in top_stats) / 1024
|
||
|
||
# Calculate score (0-100)
|
||
# Higher ops/sec = better, lower memory = better
|
||
score = self._calculate_score(ops_per_sec, memory_delta_kb)
|
||
|
||
return BenchmarkResult(
|
||
name=self.name,
|
||
operation_count=self.operations,
|
||
total_time_ms=total_time,
|
||
avg_time_ms=avg_time,
|
||
ops_per_second=ops_per_sec,
|
||
memory_delta_kb=memory_delta_kb,
|
||
score=score,
|
||
)
|
||
|
||
def _calculate_score(self, ops_per_sec: float, memory_kb: float) -> int:
|
||
"""Calculate a score from 0-100 based on performance metrics."""
|
||
# Score based on operations per second (log scale)
|
||
# 10k ops/sec = 80 points, 100k = 95 points, 1M = 100 points
|
||
if ops_per_sec >= 1_000_000:
|
||
ops_score = 100
|
||
elif ops_per_sec >= 100_000:
|
||
ops_score = 95
|
||
elif ops_per_sec >= 10_000:
|
||
ops_score = 80
|
||
elif ops_per_sec >= 1_000:
|
||
ops_score = 60
|
||
elif ops_per_sec >= 100:
|
||
ops_score = 40
|
||
else:
|
||
ops_score = 20
|
||
|
||
# Memory penalty (lower is better)
|
||
# < 1KB per op = no penalty, > 100KB = max penalty
|
||
memory_per_op = memory_kb / self.operations
|
||
if memory_per_op < 0.001:
|
||
mem_score = 0
|
||
elif memory_per_op < 0.1:
|
||
mem_score = 5
|
||
elif memory_per_op < 1:
|
||
mem_score = 10
|
||
else:
|
||
mem_score = min(15, int(memory_per_op / 10))
|
||
|
||
return max(0, min(100, ops_score - mem_score))
|
||
|
||
|
||
class TestCommandFilterBenchmarks:
|
||
"""Performance benchmarks for CommandFilter operations."""
|
||
|
||
def test_complete_command_names_performance(self):
|
||
"""Benchmark get_complete_command_names with caching."""
|
||
bench = PerformanceBenchmark("CommandFilter.get_complete_command_names", operations=10000)
|
||
|
||
# Setup: create 100 filters
|
||
filters: list[CommandFilter] = []
|
||
for i in range(100):
|
||
cf = CommandFilter(command_name=f"test_cmd_{i}")
|
||
cf.alias = {f"t{i}", f"alias{i}"}
|
||
cf.parent_command_names = [f"parent{i}"]
|
||
filters.append(cf)
|
||
|
||
result = bench.run(lambda: [cf.get_complete_command_names() for cf in filters])
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Benchmark: {result.name}")
|
||
print(f" Operations: {result.operation_count:,}")
|
||
print(f" Total time: {result.total_time_ms:.2f}ms")
|
||
print(f" Avg per call: {result.avg_time_ms:.6f}ms")
|
||
print(f" Ops/sec: {result.ops_per_second:,.0f}")
|
||
print(f" Memory delta: +{result.memory_delta_kb:.2f}KB")
|
||
print(f" SCORE: {result.score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
assert result.score >= 60, f"Performance score {result.score} is below threshold 60"
|
||
|
||
def test_validate_bool_params_performance(self):
|
||
"""Benchmark boolean parameter validation."""
|
||
bench = PerformanceBenchmark("CommandFilter.validate_bool", operations=50000)
|
||
|
||
cf = CommandFilter(command_name="test")
|
||
cf.handler_params = {"enabled": bool}
|
||
|
||
result = bench.run(
|
||
lambda: cf.validate_and_convert_params(["true"], cf.handler_params)
|
||
)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Benchmark: {result.name}")
|
||
print(f" Operations: {result.operation_count:,}")
|
||
print(f" Total time: {result.total_time_ms:.2f}ms")
|
||
print(f" Avg per call: {result.avg_time_ms:.6f}ms")
|
||
print(f" Ops/sec: {result.ops_per_second:,.0f}")
|
||
print(f" Memory delta: +{result.memory_delta_kb:.2f}KB")
|
||
print(f" SCORE: {result.score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
assert result.score >= 70, f"Performance score {result.score} is below threshold 70"
|
||
|
||
def test_validate_int_params_performance(self):
|
||
"""Benchmark integer parameter validation."""
|
||
bench = PerformanceBenchmark("CommandFilter.validate_int", operations=50000)
|
||
|
||
cf = CommandFilter(command_name="test")
|
||
cf.handler_params = {"count": int}
|
||
|
||
result = bench.run(
|
||
lambda: cf.validate_and_convert_params(["42"], cf.handler_params)
|
||
)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Benchmark: {result.name}")
|
||
print(f" Operations: {result.operation_count:,}")
|
||
print(f" Total time: {result.total_time_ms:.2f}ms")
|
||
print(f" Avg per call: {result.avg_time_ms:.6f}ms")
|
||
print(f" Ops/sec: {result.ops_per_second:,.0f}")
|
||
print(f" Memory delta: +{result.memory_delta_kb:.2f}KB")
|
||
print(f" SCORE: {result.score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
assert result.score >= 70, f"Performance score {result.score} is below threshold 70"
|
||
|
||
|
||
class TestMemoryBenchmarks:
|
||
"""Memory usage benchmarks."""
|
||
|
||
def test_filter_creation_memory(self):
|
||
"""Benchmark memory usage when creating many filters."""
|
||
bench = PerformanceBenchmark("Filter creation (1000x)", operations=10)
|
||
|
||
def create_filters():
|
||
filters = []
|
||
for i in range(1000):
|
||
cf = CommandFilter(command_name=f"cmd_{i}")
|
||
cf.alias = {f"a{i}", f"b{i}"}
|
||
filters.append(cf)
|
||
return filters
|
||
|
||
result = bench.run(create_filters)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Benchmark: {result.name}")
|
||
print(f" Creating 1000 filters")
|
||
print(f" Total memory: +{result.memory_delta_kb:.2f}KB")
|
||
print(f" Per filter: {result.memory_delta_kb:.4f}KB")
|
||
print(f" SCORE: {result.score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
# Each filter should use < 10KB of memory
|
||
per_filter_kb = result.memory_delta_kb / 1000
|
||
assert per_filter_kb < 10, f"Filter memory usage {per_filter_kb:.2f}KB is too high"
|
||
|
||
def test_greedy_str_memory(self):
|
||
"""Benchmark GreedyStr memory usage."""
|
||
bench = PerformanceBenchmark("GreedyStr creation", operations=10000)
|
||
|
||
def create_greedy():
|
||
return GreedyStr(" ".join([f"arg{i}" for i in range(100)]))
|
||
|
||
result = bench.run(create_greedy)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Benchmark: {result.name}")
|
||
print(f" Operations: {result.operation_count:,}")
|
||
print(f" Memory delta: +{result.memory_delta_kb:.2f}KB")
|
||
print(f" Per operation: {result.memory_delta_kb / result.operation_count:.4f}KB")
|
||
print(f" SCORE: {result.score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
assert result.score >= 50, f"Memory score {result.score} is below threshold 50"
|
||
|
||
|
||
class TestThroughputBenchmarks:
|
||
"""High-throughput benchmarks."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_high_throughput_bool_validation(self):
|
||
"""Test boolean validation at high throughput."""
|
||
bench = PerformanceBenchmark("High-throughput bool validation", operations=100000)
|
||
|
||
cf = CommandFilter(command_name="test")
|
||
cf.handler_params = {"enabled": bool}
|
||
values = ["true", "false", "yes", "no", "1", "0"]
|
||
|
||
def validate_many():
|
||
for v in values:
|
||
cf.validate_and_convert_params([v], cf.handler_params)
|
||
|
||
# Run 100k validations across 6 values
|
||
result = bench.run(validate_many)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Benchmark: {result.name}")
|
||
print(f" Total operations: {result.operation_count * 6:,}")
|
||
print(f" Effective ops/sec: {result.ops_per_second * 6:,.0f}")
|
||
print(f" Total time: {result.total_time_ms:.2f}ms")
|
||
print(f" SCORE: {result.score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
# Should handle > 100k validations per second
|
||
effective_ops = result.ops_per_second * 6
|
||
assert effective_ops > 100_000, f"Throughput {effective_ops:,.0f} is too low"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_command_name_resolution_throughput(self):
|
||
"""Test command name resolution at high throughput."""
|
||
bench = PerformanceBenchmark("Command name resolution", operations=50000)
|
||
|
||
filters = []
|
||
for i in range(50):
|
||
cf = CommandFilter(command_name=f"cmd_{i}")
|
||
cf.alias = {f"c{i}", f"d{i}"}
|
||
filters.append(cf)
|
||
|
||
def resolve_all():
|
||
for cf in filters:
|
||
cf.get_complete_command_names()
|
||
|
||
result = bench.run(resolve_all)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Benchmark: {result.name}")
|
||
print(f" Operations: {result.operation_count:,}")
|
||
print(f" Ops/sec: {result.ops_per_second:,.0f}")
|
||
print(f" SCORE: {result.score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
assert result.ops_per_second > 100_000, f"Throughput {result.ops_per_second:,.0f} is too low"
|
||
|
||
|
||
class TestScoringSummary:
|
||
"""Summary test that reports overall score."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_overall_performance_score(self):
|
||
"""Calculate overall performance score across all benchmarks."""
|
||
scores = []
|
||
|
||
# Test 1: CommandFilter operations
|
||
bench1 = PerformanceBenchmark("CommandFilter ops", operations=10000)
|
||
filters = [CommandFilter(command_name=f"c{i}") for i in range(50)]
|
||
result1 = bench1.run(lambda: [f.get_complete_command_names() for f in filters])
|
||
scores.append(result1.score)
|
||
|
||
# Test 2: Bool validation
|
||
bench2 = PerformanceBenchmark("Bool validation", operations=50000)
|
||
cf = CommandFilter(command_name="test")
|
||
cf.handler_params = {"enabled": bool}
|
||
result2 = bench2.run(lambda: cf.validate_and_convert_params(["true"], cf.handler_params))
|
||
scores.append(result2.score)
|
||
|
||
# Test 3: Int validation
|
||
bench3 = PerformanceBenchmark("Int validation", operations=50000)
|
||
cf2 = CommandFilter(command_name="test2")
|
||
cf2.handler_params = {"count": int}
|
||
result3 = bench3.run(lambda: cf2.validate_and_convert_params(["42"], cf2.handler_params))
|
||
scores.append(result3.score)
|
||
|
||
overall_score = sum(scores) // len(scores)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"PERFORMANCE BENCHMARK SUMMARY")
|
||
print(f"{'='*60}")
|
||
print(f" CommandFilter operations: {scores[0]}/100")
|
||
print(f" Bool validation: {scores[1]}/100")
|
||
print(f" Int validation: {scores[2]}/100")
|
||
print(f" {'-'*40}")
|
||
print(f" OVERALL SCORE: {overall_score}/100")
|
||
print(f"{'='*60}\n")
|
||
|
||
assert overall_score >= 60, f"Overall score {overall_score} is below threshold 60"
|