Get or create a lock for a specific cache key
(self, key: str)
| 146 | self._cache.clear() |
| 147 | |
| 148 | def _get_compute_lock(self, key: str) -> threading.Lock: |
| 149 | """Get or create a lock for a specific cache key""" |
| 150 | with self._compute_locks_lock: |
| 151 | if key not in self._compute_locks: |
| 152 | self._compute_locks[key] = threading.Lock() |
| 153 | return self._compute_locks[key] |
| 154 | |
| 155 | def _cleanup_compute_lock(self, key: str): |
| 156 | """Clean up compute lock after use to avoid memory leak""" |