"""ASV benchmarks for RegistryCache, CacheFactory, and cached resolution. Measures performance of: - get_package cache hit and cache miss+fetch - resolve_package cache hit and cache miss+fetch - singleflight coalescing (concurrent-miss deduplication) - manual put (content warming), invalidate, and clear - CacheFactory.create() construction - End-to-end PackageContentResolver resolution with cache """ from __future__ import annotations import asyncio from typing import Any from unittest.mock import AsyncMock, MagicMock _STANDARD_CONTENT: dict[str, Any] = { "name": "bench-pkg", "type": "actor", "version": "v1.0.0", "config": { "agent_type": "llm", "model": "gpt-4", "temperature": 0.7, "system_prompt": "You are a benchmarking assistant.", }, } _RESOLVE_CONTENT: dict[str, Any] = { "package_id": "pkg_act_0123456789abcdef0123456789abcdef01234567", "type": "actor", "name": "bench-pkg", "version": "v1.0.0", "config": { "agent_type": "llm", "model": "gpt-4", "temperature": 0.7, }, } _STANDARD_PKG_ID: str = "pkg_act_0123456789abcdef0123456789abcdef01234567" def _create_mock_client() -> MagicMock: from cleveractors.registry.client import RegistryClient mock = MagicMock(spec=RegistryClient) mock.get_package = AsyncMock(return_value=dict(_STANDARD_CONTENT)) mock.resolve_package = AsyncMock(return_value=dict(_RESOLVE_CONTENT)) mock.close = AsyncMock() return mock class CacheGetPackageHitBenchmark: """Benchmark get_package() when content is already cached.""" def setup(self) -> None: from cleveractors.registry.cache import RegistryCache self.cache = RegistryCache(_create_mock_client(), validate_content=False) self.pkg_id = _STANDARD_PKG_ID asyncio.run(self.cache.get_package(self.pkg_id)) def time_get_package_hit(self) -> None: asyncio.run(self.cache.get_package(self.pkg_id)) def peakmem_get_package_hit(self) -> None: asyncio.run(self.cache.get_package(self.pkg_id)) class CacheGetPackageMissBenchmark: """Benchmark get_package() on cold cache (miss + upstream fetch).""" def setup(self) -> None: from cleveractors.registry.cache import RegistryCache self.mock_client = _create_mock_client() self.pkg_id = _STANDARD_PKG_ID def time_get_package_miss(self) -> None: from cleveractors.registry.cache import RegistryCache cache = RegistryCache(self.mock_client, validate_content=False) asyncio.run(cache.get_package(self.pkg_id)) class CacheResolvePackageHitBenchmark: """Benchmark resolve_package() when the resolution is cached.""" params: list[str] = ["actor", "template", "skill"] def setup(self, pkg_type: str) -> None: from cleveractors.registry.cache import RegistryCache self.cache = RegistryCache(_create_mock_client(), validate_content=False) asyncio.run( self.cache.resolve_package( package_type=pkg_type, namespace="bench-ns", name="bench-pkg", version="v1.0.0", ) ) def time_resolve_package_hit(self, pkg_type: str) -> None: asyncio.run( self.cache.resolve_package( package_type=pkg_type, namespace="bench-ns", name="bench-pkg", version="v1.0.0", ) ) class CacheResolvePackageMissBenchmark: """Benchmark resolve_package() on cold cache.""" params: list[str] = ["actor", "template", "skill"] def setup(self, pkg_type: str) -> None: from cleveractors.registry.cache import RegistryCache self.mock_client = _create_mock_client() def time_resolve_package_miss(self, pkg_type: str) -> None: from cleveractors.registry.cache import RegistryCache cache = RegistryCache(self.mock_client, validate_content=False) asyncio.run( cache.resolve_package( package_type=pkg_type, namespace="bench-ns", name="bench-pkg", version="v1.0.0", ) ) class CacheSingleflightBenchmark: """Benchmark the singleflight coalescing under concurrent access. When multiple coroutines request the same cold key concurrently, only one upstream fetch is performed — the rest join the in-flight future. This benchmark measures the throughput of N concurrent get_package calls for the same cold key. """ params: list[int] = [10, 50] def setup(self, concurrency: int) -> None: from cleveractors.registry.cache import RegistryCache self.cache = RegistryCache(_create_mock_client(), validate_content=False) self.pkg_id = _STANDARD_PKG_ID self.concurrency = concurrency def time_get_package_concurrent(self, concurrency: int) -> None: async def _run(): tasks = [ asyncio.ensure_future(self.cache.get_package(self.pkg_id)) for _ in range(self.concurrency) ] await asyncio.gather(*tasks) asyncio.run(_run()) class CacheInvalidateBenchmark: """Benchmark per-entry invalidation.""" def setup(self) -> None: from cleveractors.registry.cache import RegistryCache self.cache = RegistryCache(_create_mock_client(), validate_content=False) self.pkg_id = _STANDARD_PKG_ID asyncio.run(self.cache.get_package(self.pkg_id)) def time_invalidate(self) -> None: asyncio.run(self.cache.invalidate(self.pkg_id)) class CacheClearBenchmark: """Benchmark full cache clear with N pre-loaded entries.""" params: list[int] = [10, 100] def setup(self, num_entries: int) -> None: from cleveractors.registry.cache import RegistryCache self.cache = RegistryCache( _create_mock_client(), max_size=max(num_entries, 10), validate_content=False ) self.entries = [f"pkg_act_{i:040x}" for i in range(num_entries)] async def _warm(): for pid in self.entries: await self.cache.put(pid, dict(_STANDARD_CONTENT)) asyncio.run(_warm()) def time_clear(self, num_entries: int) -> None: asyncio.run(self.cache.clear()) class CachePutBenchmark: """Benchmark manual cache warming via put().""" def setup(self) -> None: from cleveractors.registry.cache import RegistryCache self.cache = RegistryCache(_create_mock_client(), validate_content=False) self.pkg_id = _STANDARD_PKG_ID self.content = dict(_STANDARD_CONTENT) def time_put(self) -> None: asyncio.run(self.cache.put(self.pkg_id, self.content)) class CacheFactoryBenchmark: """Benchmark CacheFactory.create() construction.""" def setup(self) -> None: from cleveractors.registry.cache import CacheFactory self.factory = CacheFactory(max_size=256, ttl=300.0, validate_content=False) self.mock_client = _create_mock_client() def time_create(self) -> None: self.factory.create(self.mock_client) class CacheContentValidationBenchmark: """Benchmark SHA-1 content validation overhead. Compares get_package throughput with and without validation enabled to measure the cost of canonicalize + SHA-1 recompute. When validation fails (content hash ≠ stored ID) the entry is evicted and re-fetched — the computational cost of the failed validation is the same but the re-fetch allocates additional memory. """ params: list[bool] = [True, False] def setup(self, validate: bool) -> None: from cleveractors.registry.cache import RegistryCache self.cache = RegistryCache(_create_mock_client(), validate_content=validate) self.pkg_id = _STANDARD_PKG_ID asyncio.run(self.cache.get_package(self.pkg_id)) def time_get_package(self, validate: bool) -> None: asyncio.run(self.cache.get_package(self.pkg_id))