test(core): add ASV performance benchmarks for core module

Added three new ASV benchmark suites for the core module:

1. core_circuit_breaker_bench.py - Benchmarks for CircuitBreaker class:
   - Closed state call overhead
   - Open state fast-fail latency
   - State transition overhead (closed→open, open→half-open, half-open→closed)
   - Async operations
   - Initialization overhead

2. core_retry_patterns_bench.py - Benchmarks for retry decorators:
   - Decorator construction overhead for exponential backoff, jitter, timeout, and result-based retry
   - Happy-path invocation overhead
   - Async decorator operations
   - Decorator usage with function arguments
   - Factory function performance
   - Various configuration scenarios

3. core_retry_service_patterns_bench.py - Benchmarks for service-level retry:
   - retry_service_operation decorator construction and invocation
   - retry_auto_debug decorator performance
   - RetryContext operations
   - is_read_only_plan_operation utility
   - Different wait strategies (exponential, linear, fixed, jitter)

These benchmarks ensure latency regressions in core resilience primitives are caught early.

ISSUES CLOSED: #1921
This commit is contained in:
2026-05-03 01:14:59 +00:00
committed by Forgejo
parent 3f8f8eb0bf
commit ffd83e8712
3 changed files with 868 additions and 0 deletions
+252
View File
@@ -0,0 +1,252 @@
"""ASV benchmarks for circuit breaker pattern.
Measures the overhead of circuit breaker state management,
state transitions, and fast-fail latency in open state.
"""
from __future__ import annotations
import asyncio
import time
from typing import Any
from cleveragents.core.circuit_breaker import (
CircuitBreaker,
CircuitBreakerOpen,
CircuitBreakerState,
)
class CircuitBreakerClosedStateBench:
"""Benchmark circuit breaker overhead in closed state."""
timeout = 60
def setup(self) -> None:
"""Set up circuit breaker in closed state."""
self.breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60.0,
expected_exception=Exception,
half_open_max_successes=2,
cooldown_seconds=30.0,
name="test_service",
)
def time_call_success(self) -> None:
"""Benchmark successful call overhead in closed state."""
def dummy_func() -> str:
return "success"
self.breaker.call(dummy_func)
def time_call_with_args(self) -> None:
"""Benchmark call with arguments in closed state."""
def dummy_func(x: int, y: str) -> str:
return f"{x}:{y}"
self.breaker.call(dummy_func, 42, "test")
def time_state_check(self) -> None:
"""Benchmark state property access."""
_ = self.breaker.state
def time_failure_count_check(self) -> None:
"""Benchmark failure count property access."""
_ = self.breaker.failure_count
class CircuitBreakerOpenStateBench:
"""Benchmark circuit breaker fast-fail latency in open state."""
timeout = 60
def setup(self) -> None:
"""Set up circuit breaker in open state."""
self.breaker = CircuitBreaker(
failure_threshold=2,
recovery_timeout=60.0,
expected_exception=Exception,
half_open_max_successes=2,
cooldown_seconds=30.0,
name="test_service",
)
# Force the breaker into open state
for _ in range(3):
try:
self.breaker.call(lambda: 1 / 0) # Raise ZeroDivisionError
except ZeroDivisionError:
pass
def time_fast_fail(self) -> None:
"""Benchmark fast-fail latency when circuit is open."""
try:
self.breaker.call(lambda: "should not execute")
except CircuitBreakerOpen:
pass
def time_open_state_check(self) -> None:
"""Benchmark state check when open."""
assert self.breaker.state == CircuitBreakerState.OPEN
class CircuitBreakerStateTransitionBench:
"""Benchmark state transition overhead."""
timeout = 60
def setup(self) -> None:
"""Set up circuit breaker."""
self.breaker = CircuitBreaker(
failure_threshold=2,
recovery_timeout=0.1, # Short timeout for testing
expected_exception=Exception,
half_open_max_successes=1,
cooldown_seconds=0.05,
name="test_service",
)
def time_closed_to_open_transition(self) -> None:
"""Benchmark transition from closed to open state."""
breaker = CircuitBreaker(
failure_threshold=2,
recovery_timeout=60.0,
expected_exception=Exception,
half_open_max_successes=1,
cooldown_seconds=30.0,
name="test_service",
)
# Trigger failures to transition to open
for _ in range(3):
try:
breaker.call(lambda: 1 / 0)
except ZeroDivisionError:
pass
def time_open_to_half_open_transition(self) -> None:
"""Benchmark transition from open to half-open state."""
breaker = CircuitBreaker(
failure_threshold=2,
recovery_timeout=0.05, # Very short timeout
expected_exception=Exception,
half_open_max_successes=1,
cooldown_seconds=0.01,
name="test_service",
)
# Force to open
for _ in range(3):
try:
breaker.call(lambda: 1 / 0)
except ZeroDivisionError:
pass
# Wait for recovery timeout
time.sleep(0.1)
# Attempt call to trigger half-open transition
try:
breaker.call(lambda: "success")
except CircuitBreakerOpen:
pass
def time_half_open_to_closed_transition(self) -> None:
"""Benchmark transition from half-open to closed state."""
breaker = CircuitBreaker(
failure_threshold=2,
recovery_timeout=0.05,
expected_exception=Exception,
half_open_max_successes=1,
cooldown_seconds=0.01,
name="test_service",
)
# Force to open
for _ in range(3):
try:
breaker.call(lambda: 1 / 0)
except ZeroDivisionError:
pass
# Wait for recovery timeout
time.sleep(0.1)
# Successful call in half-open should close
breaker.call(lambda: "success")
class CircuitBreakerAsyncBench:
"""Benchmark async circuit breaker operations."""
timeout = 60
def setup(self) -> None:
"""Set up async circuit breaker."""
self.breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60.0,
expected_exception=Exception,
half_open_max_successes=2,
cooldown_seconds=30.0,
name="test_service",
)
def time_async_call_success(self) -> None:
"""Benchmark successful async call overhead."""
async def dummy_async() -> str:
return "success"
asyncio.run(self.breaker.async_call(dummy_async))
def time_async_call_with_args(self) -> None:
"""Benchmark async call with arguments."""
async def dummy_async(x: int, y: str) -> str:
return f"{x}:{y}"
asyncio.run(self.breaker.async_call(dummy_async, 42, "test"))
def time_async_fast_fail(self) -> None:
"""Benchmark async fast-fail when open."""
# First set up breaker in open state
breaker = CircuitBreaker(
failure_threshold=2,
recovery_timeout=60.0,
expected_exception=Exception,
half_open_max_successes=2,
cooldown_seconds=30.0,
name="test_service",
)
# Force to open
for _ in range(3):
try:
breaker.call(lambda: 1 / 0)
except ZeroDivisionError:
pass
async def dummy_async() -> str:
return "should not execute"
try:
asyncio.run(breaker.async_call(dummy_async))
except CircuitBreakerOpen:
pass
class CircuitBreakerInitializationBench:
"""Benchmark circuit breaker initialization overhead."""
timeout = 60
def time_default_initialization(self) -> None:
"""Benchmark default initialization."""
CircuitBreaker()
def time_custom_initialization(self) -> None:
"""Benchmark initialization with custom parameters."""
CircuitBreaker(
failure_threshold=10,
recovery_timeout=120.0,
expected_exception=ValueError,
half_open_max_successes=3,
cooldown_seconds=45.0,
name="custom_service",
)
def time_multiple_breakers(self) -> None:
"""Benchmark creating multiple breakers."""
for i in range(10):
CircuitBreaker(name=f"service_{i}")
+281
View File
@@ -0,0 +1,281 @@
"""ASV benchmarks for retry patterns.
Measures the overhead of retry decorator construction and
happy-path invocation for the public retry factory functions.
"""
from __future__ import annotations
import asyncio
from typing import Any
from cleveragents.core.retry_patterns import (
get_retry_decorator,
retry_on_result,
retry_with_exponential_backoff,
retry_with_jitter,
retry_with_timeout,
)
class RetryDecoratorConstructionBench:
"""Benchmark retry decorator construction overhead."""
timeout = 60
def time_retry_with_exponential_backoff_construction(self) -> None:
"""Benchmark constructing exponential backoff decorator."""
@retry_with_exponential_backoff(max_attempts=3)
def dummy_func() -> str:
return "success"
def time_retry_with_jitter_construction(self) -> None:
"""Benchmark constructing jitter decorator."""
@retry_with_jitter(max_attempts=3)
def dummy_func() -> str:
return "success"
def time_retry_with_timeout_construction(self) -> None:
"""Benchmark constructing timeout decorator."""
@retry_with_timeout(timeout_seconds=5.0)
def dummy_func() -> str:
return "success"
def time_retry_on_result_construction(self) -> None:
"""Benchmark constructing result-based retry decorator."""
@retry_on_result(max_attempts=3)
def dummy_func() -> str:
return "success"
def time_get_retry_decorator_construction(self) -> None:
"""Benchmark constructing decorator via factory."""
get_retry_decorator(
max_attempts=3,
base_delay=1.0,
max_delay=60.0,
strategy="exponential",
)
class RetryDecoratorInvocationBench:
"""Benchmark happy-path retry decorator invocation."""
timeout = 60
def setup(self) -> None:
"""Set up decorated functions."""
@retry_with_exponential_backoff(max_attempts=3)
def exponential_func() -> str:
return "success"
@retry_with_jitter(max_attempts=3)
def jitter_func() -> str:
return "success"
@retry_with_timeout(timeout_seconds=5.0)
def timeout_func() -> str:
return "success"
@retry_on_result(max_attempts=3)
def result_func() -> str:
return "success"
self.exponential_func = exponential_func
self.jitter_func = jitter_func
self.timeout_func = timeout_func
self.result_func = result_func
def time_exponential_backoff_invocation(self) -> None:
"""Benchmark exponential backoff invocation (happy path)."""
self.exponential_func()
def time_jitter_invocation(self) -> None:
"""Benchmark jitter invocation (happy path)."""
self.jitter_func()
def time_timeout_invocation(self) -> None:
"""Benchmark timeout invocation (happy path)."""
self.timeout_func()
def time_result_based_invocation(self) -> None:
"""Benchmark result-based retry invocation (happy path)."""
self.result_func()
class RetryDecoratorAsyncBench:
"""Benchmark async retry decorator operations."""
timeout = 60
def setup(self) -> None:
"""Set up async decorated functions."""
@retry_with_exponential_backoff(max_attempts=3)
async def async_exponential() -> str:
return "success"
@retry_with_jitter(max_attempts=3)
async def async_jitter() -> str:
return "success"
@retry_with_timeout(timeout_seconds=5.0)
async def async_timeout() -> str:
return "success"
self.async_exponential = async_exponential
self.async_jitter = async_jitter
self.async_timeout = async_timeout
def time_async_exponential_invocation(self) -> None:
"""Benchmark async exponential backoff invocation."""
asyncio.run(self.async_exponential())
def time_async_jitter_invocation(self) -> None:
"""Benchmark async jitter invocation."""
asyncio.run(self.async_jitter())
def time_async_timeout_invocation(self) -> None:
"""Benchmark async timeout invocation."""
asyncio.run(self.async_timeout())
class RetryDecoratorWithArgumentsBench:
"""Benchmark retry decorators with function arguments."""
timeout = 60
def setup(self) -> None:
"""Set up decorated functions with arguments."""
@retry_with_exponential_backoff(max_attempts=3)
def func_with_args(x: int, y: str, z: float = 1.0) -> str:
return f"{x}:{y}:{z}"
@retry_with_jitter(max_attempts=3)
def func_with_kwargs(a: int, b: str = "default") -> str:
return f"{a}:{b}"
self.func_with_args = func_with_args
self.func_with_kwargs = func_with_kwargs
def time_exponential_with_positional_args(self) -> None:
"""Benchmark exponential backoff with positional arguments."""
self.func_with_args(42, "test", 2.5)
def time_exponential_with_keyword_args(self) -> None:
"""Benchmark exponential backoff with keyword arguments."""
self.func_with_args(x=42, y="test", z=2.5)
def time_jitter_with_mixed_args(self) -> None:
"""Benchmark jitter with mixed positional and keyword arguments."""
self.func_with_kwargs(100, b="custom")
class RetryDecoratorFactoryBench:
"""Benchmark retry decorator factory function."""
timeout = 60
def time_factory_exponential_strategy(self) -> None:
"""Benchmark factory with exponential strategy."""
decorator = get_retry_decorator(
max_attempts=3,
base_delay=1.0,
max_delay=60.0,
strategy="exponential",
)
@decorator
def dummy() -> str:
return "success"
dummy()
def time_factory_linear_strategy(self) -> None:
"""Benchmark factory with linear strategy."""
decorator = get_retry_decorator(
max_attempts=3,
base_delay=2.0,
max_delay=60.0,
strategy="linear",
)
@decorator
def dummy() -> str:
return "success"
dummy()
def time_factory_fixed_strategy(self) -> None:
"""Benchmark factory with fixed strategy."""
decorator = get_retry_decorator(
max_attempts=3,
base_delay=0.5,
max_delay=60.0,
strategy="fixed",
)
@decorator
def dummy() -> str:
return "success"
dummy()
def time_factory_jitter_strategy(self) -> None:
"""Benchmark factory with jitter strategy."""
decorator = get_retry_decorator(
max_attempts=3,
base_delay=1.0,
max_delay=60.0,
strategy="jitter",
)
@decorator
def dummy() -> str:
return "success"
dummy()
class RetryDecoratorConfigurationBench:
"""Benchmark retry decorator with various configurations."""
timeout = 60
def time_minimal_config(self) -> None:
"""Benchmark decorator with minimal configuration."""
@retry_with_exponential_backoff()
def dummy() -> str:
return "success"
dummy()
def time_aggressive_config(self) -> None:
"""Benchmark decorator with aggressive retry settings."""
@retry_with_exponential_backoff(max_attempts=10)
def dummy() -> str:
return "success"
dummy()
def time_conservative_config(self) -> None:
"""Benchmark decorator with conservative retry settings."""
@retry_with_exponential_backoff(max_attempts=1)
def dummy() -> str:
return "success"
dummy()
def time_long_timeout_config(self) -> None:
"""Benchmark decorator with long timeout."""
@retry_with_timeout(timeout_seconds=300.0)
def dummy() -> str:
return "success"
dummy()
def time_short_timeout_config(self) -> None:
"""Benchmark decorator with short timeout."""
@retry_with_timeout(timeout_seconds=0.1)
def dummy() -> str:
return "success"
dummy()
@@ -0,0 +1,335 @@
"""ASV benchmarks for service-level retry patterns.
Measures the overhead of retry_service_operation decorator
and retry_auto_debug happy-path latency.
"""
from __future__ import annotations
import asyncio
from typing import Any
from cleveragents.core.retry_service_patterns import (
RetryContext,
is_read_only_plan_operation,
retry_auto_debug,
retry_service_operation,
)
class RetryServiceOperationDecoratorBench:
"""Benchmark retry_service_operation decorator overhead."""
timeout = 60
def time_decorator_construction_sync(self) -> None:
"""Benchmark constructing sync service retry decorator."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
)
def dummy_func() -> str:
return "success"
def time_decorator_construction_async(self) -> None:
"""Benchmark constructing async service retry decorator."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
)
async def dummy_func() -> str:
return "success"
def time_decorator_with_circuit_breaker(self) -> None:
"""Benchmark decorator with circuit breaker enabled."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
use_circuit_breaker=True,
)
def dummy_func() -> str:
return "success"
def time_decorator_with_custom_strategy(self) -> None:
"""Benchmark decorator with custom wait strategy."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
wait_strategy="exponential",
base_delay=1.0,
max_delay=60.0,
)
def dummy_func() -> str:
return "success"
class RetryServiceOperationInvocationBench:
"""Benchmark happy-path service retry invocation."""
timeout = 60
def setup(self) -> None:
"""Set up decorated service operations."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
)
def sync_op() -> str:
return "success"
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
)
async def async_op() -> str:
return "success"
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
use_circuit_breaker=True,
)
def breaker_op() -> str:
return "success"
self.sync_op = sync_op
self.async_op = async_op
self.breaker_op = breaker_op
def time_sync_operation_invocation(self) -> None:
"""Benchmark sync service operation invocation (happy path)."""
self.sync_op()
def time_async_operation_invocation(self) -> None:
"""Benchmark async service operation invocation (happy path)."""
asyncio.run(self.async_op())
def time_operation_with_circuit_breaker(self) -> None:
"""Benchmark service operation with circuit breaker (happy path)."""
self.breaker_op()
class RetryServiceOperationWithArgumentsBench:
"""Benchmark service retry with function arguments."""
timeout = 60
def setup(self) -> None:
"""Set up decorated operations with arguments."""
@retry_service_operation(
service_name="test_service",
operation_name="fetch_data",
max_attempts=3,
)
def fetch_data(resource_id: int, include_metadata: bool = False) -> dict[str, Any]:
return {"id": resource_id, "data": "test"}
@retry_service_operation(
service_name="test_service",
operation_name="update_data",
max_attempts=3,
)
async def update_data(resource_id: int, payload: dict[str, Any]) -> dict[str, Any]:
return {"id": resource_id, "updated": True}
self.fetch_data = fetch_data
self.update_data = update_data
def time_sync_with_positional_args(self) -> None:
"""Benchmark sync operation with positional arguments."""
self.fetch_data(42, True)
def time_sync_with_keyword_args(self) -> None:
"""Benchmark sync operation with keyword arguments."""
self.fetch_data(resource_id=42, include_metadata=True)
def time_async_with_complex_payload(self) -> None:
"""Benchmark async operation with complex payload."""
payload = {"key": "value", "nested": {"data": [1, 2, 3]}}
asyncio.run(self.update_data(42, payload))
class RetryAutoDebugBench:
"""Benchmark retry_auto_debug decorator."""
timeout = 60
def time_auto_debug_construction(self) -> None:
"""Benchmark constructing auto-debug decorator."""
@retry_auto_debug(
service_name="test_service",
operation_name="test_op",
)
def dummy_func() -> str:
return "success"
def time_auto_debug_invocation(self) -> None:
"""Benchmark auto-debug invocation (happy path)."""
@retry_auto_debug(
service_name="test_service",
operation_name="test_op",
)
def dummy_func() -> str:
return "success"
dummy_func()
def time_auto_debug_async_invocation(self) -> None:
"""Benchmark async auto-debug invocation (happy path)."""
@retry_auto_debug(
service_name="test_service",
operation_name="test_op",
)
async def dummy_func() -> str:
return "success"
asyncio.run(dummy_func())
def time_auto_debug_with_context(self) -> None:
"""Benchmark auto-debug with retry context."""
@retry_auto_debug(
service_name="test_service",
operation_name="test_op",
)
def dummy_func(ctx: RetryContext | None = None) -> str:
return "success"
dummy_func()
class RetryContextBench:
"""Benchmark RetryContext operations."""
timeout = 60
def time_context_creation(self) -> None:
"""Benchmark creating a retry context."""
RetryContext(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
)
def time_context_with_circuit_breaker(self) -> None:
"""Benchmark context with circuit breaker."""
RetryContext(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
use_circuit_breaker=True,
)
def time_context_property_access(self) -> None:
"""Benchmark accessing context properties."""
ctx = RetryContext(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
)
_ = ctx.service_name
_ = ctx.operation_name
_ = ctx.max_attempts
_ = ctx.attempt_count
class IsReadOnlyPlanOperationBench:
"""Benchmark is_read_only_plan_operation utility."""
timeout = 60
def time_check_strategize(self) -> None:
"""Benchmark checking strategize operation."""
is_read_only_plan_operation("strategize")
def time_check_plan(self) -> None:
"""Benchmark checking plan operation."""
is_read_only_plan_operation("plan")
def time_check_validate(self) -> None:
"""Benchmark checking validate operation."""
is_read_only_plan_operation("validate")
def time_check_review(self) -> None:
"""Benchmark checking review operation."""
is_read_only_plan_operation("review")
def time_check_preview(self) -> None:
"""Benchmark checking preview operation."""
is_read_only_plan_operation("preview")
def time_check_check(self) -> None:
"""Benchmark checking check operation."""
is_read_only_plan_operation("check")
def time_check_execute(self) -> None:
"""Benchmark checking execute operation (not read-only)."""
is_read_only_plan_operation("execute")
def time_check_unknown_operation(self) -> None:
"""Benchmark checking unknown operation."""
is_read_only_plan_operation("unknown_op")
class RetryServiceOperationStrategyBench:
"""Benchmark different retry strategies in service operations."""
timeout = 60
def time_exponential_strategy(self) -> None:
"""Benchmark exponential strategy."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
wait_strategy="exponential",
)
def dummy() -> str:
return "success"
dummy()
def time_linear_strategy(self) -> None:
"""Benchmark linear strategy."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
wait_strategy="linear",
)
def dummy() -> str:
return "success"
dummy()
def time_fixed_strategy(self) -> None:
"""Benchmark fixed strategy."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
wait_strategy="fixed",
)
def dummy() -> str:
return "success"
dummy()
def time_jitter_strategy(self) -> None:
"""Benchmark jitter strategy."""
@retry_service_operation(
service_name="test_service",
operation_name="test_op",
max_attempts=3,
wait_strategy="jitter",
)
def dummy() -> str:
return "success"
dummy()