mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-15 17:30:13 +08:00
75fa652ccb
- Comprehensive performance benchmarks for CommandFilter operations - Memory usage benchmarks with tracemalloc tracking - High-throughput benchmarks with ops/sec metrics - Scoring system (0-100) for performance tracking - Overall performance score summary Benchmarks include: - CommandFilter.get_complete_command_names - Boolean and integer parameter validation - Memory footprint of filter creation - High-throughput validation throughput
339 lines
12 KiB
Python
339 lines
12 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 Callable, Any
|
|
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."""
|
|
gc.collect()
|
|
self.tracemalloc.start()
|
|
snapshot_before = self.tracemalloc.take_snapshot()
|
|
|
|
start = asyncio.get_event_loop().time()
|
|
for _ in range(self.operations):
|
|
func(*args, **kwargs)
|
|
end = asyncio.get_event_loop().time()
|
|
|
|
snapshot_after = self.tracemalloc.take_snapshot()
|
|
self.tracemalloc.stop()
|
|
|
|
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)
|
|
|
|
# Calculate memory delta
|
|
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"
|