(self, max_size: int = 5)
| 831 | return cls._instance |
| 832 | |
| 833 | def __init__(self, max_size: int = 5): |
| 834 | if self._initialized: |
| 835 | return |
| 836 | |
| 837 | with self._lock: |
| 838 | if not self._initialized: |
| 839 | logger.info("Initializing CacheManager singleton") |
| 840 | self.max_size = max_size |
| 841 | self.model_cache = OrderedDict() |
| 842 | self.tokenizer_cache = OrderedDict() |
| 843 | self.adapter_cache = OrderedDict() |
| 844 | self.model_adapter_map = {} # Maps model ID to list of loaded adapter IDs |
| 845 | self.cache_stats = defaultdict(lambda: {"hits": 0, "misses": 0}) |
| 846 | self._initialized = True |
| 847 | logger.info("CacheManager singleton initialized") |
| 848 | |
| 849 | def get_or_load_model(self, model_key: str, loader_fn) -> Tuple[Any, Any]: |
| 850 | """Get or load model and tokenizer with minimal locking.""" |
nothing calls this directly
no outgoing calls
no test coverage detected