pumpwood_communication.cache

 19class PumpwoodCache:
 20    """Class to implement local cache for Pumpwood Comunication requests."""
 21
 22    _INIT_LOCK = threading.Lock()
 23    """Lock to serialize FanoutCache initialization across threads."""
 24
 25    def __init__(self):
 26        """Initialize cache settings from configuration."""
 27        self._cache = None
 28        self._size_limit = CACHE_LIMIT_MB
 29        self._expire_time = CACHE_DEFAULT_EXPIRE
 30        self._transaction_timeout = CACHE_TRANSACTION_TIMEOUT
 31        self._n_shards = CACHE_N_SHARDS
 32        self._enable = CACHE_ENABLE
 33        self._retry_attempts = CACHE_RETRY_ATTEMPTS
 34        self._retry_delay = CACHE_RETRY_DELAY
 35        self._cache_path = (
 36            Path('/tmp/pumpwood_cache/') /
 37            CACHE_BASE_PATH)
 38
 39    def _build_fanout_cache(self) -> Optional[FanoutCache]:
 40        """Build FanoutCache with retries on SQLite lock contention.
 41
 42        Returns:
 43            Optional[FanoutCache]:
 44                Configured cache instance, or None when initialization
 45                fails after all retry attempts.
 46        """
 47        last_error = None
 48        for attempt in range(1, self._retry_attempts + 1):
 49            try:
 50                return FanoutCache(
 51                    directory=self._cache_path,
 52                    cache_size=self._size_limit,
 53                    tag_index=True,
 54                    timeout=self._transaction_timeout,
 55                    shards=self._n_shards)
 56            except (Timeout, sqlite3.OperationalError) as error:
 57                last_error = error
 58                if attempt >= self._retry_attempts:
 59                    break
 60                delay = self._retry_delay * attempt
 61                warning_msg = (
 62                    "Cache init locked, retry {attempt}/{total} "
 63                    "after {delay}s")
 64                logger.warning(
 65                    warning_msg.format(
 66                        attempt=attempt,
 67                        total=self._retry_attempts,
 68                        delay=delay))
 69                time.sleep(delay)
 70
 71        warning_msg = (
 72            "Cache init failed after {total} attempts: {error}")
 73        logger.warning(
 74            warning_msg.format(
 75                total=self._retry_attempts,
 76                error=last_error))
 77        return None
 78
 79    def refresh_cache(self) -> bool:
 80        """Refresh cache.
 81
 82        Returns:
 83            True if cache is refreshed.
 84        """
 85        self._cache = self._build_fanout_cache()
 86        return True
 87    
 88    def _create_cache_object(self) -> None:
 89        """Create FanoutCache once in a thread-safe way."""
 90        if self._cache is not None:
 91            return
 92        with self._INIT_LOCK:
 93            if self._cache is None:
 94                self._cache = self._build_fanout_cache()
 95
 96    def _execute_with_retry(self, operation: Callable[[], Any],
 97                            default: Any,
 98                            operation_name: str) -> Any:
 99        """Execute a cache operation with retries on lock contention.
100
101        Args:
102            operation (Callable[[], Any]):
103                Callable that performs the cache operation.
104            default (Any):
105                Value returned when all retries are exhausted or when
106                cache initialization fails.
107            operation_name (str):
108                Operation label used in warning logs.
109
110        Returns:
111            Any:
112                Result from ``operation``, or ``default`` on failure.
113        """
114        # Try to execute the operation with retries on lock contention.
115        last_error = None
116        for attempt in range(1, self._retry_attempts + 1):
117            try:
118                self._create_cache_object()
119                if self._cache is None:
120                    return default
121                return operation()
122            except (Timeout, sqlite3.OperationalError) as error:
123                last_error = error
124                if attempt >= self._retry_attempts:
125                    break
126                delay = self._retry_delay * attempt
127                warning_msg = (
128                    "Cache {operation} locked, retry {attempt}/{total} "
129                    "after {delay}s")
130                logger.warning(
131                    warning_msg.format(
132                        operation=operation_name,
133                        attempt=attempt,
134                        total=self._retry_attempts,
135                        delay=delay))
136                time.sleep(delay)
137
138        warning_msg = (
139            "Cache {operation} failed after {total} attempts: {error}")
140        logger.warning(
141            warning_msg.format(
142                operation=operation_name, total=self._retry_attempts,
143                error=last_error))
144        return default
145
146    @classmethod
147    def generate_hash(cls, hash_dict: dict) -> str:
148        """Generate a hash to be used to storage and retrieve cache.
149
150        It will use pumpJsonDump function from serializers to dump correctly
151        any complex data such as date, geometry and numpy.
152
153        Expose _generate_hash
154
155        Args:
156            hash_dict (dict):
157                A dictonary with information that will be used on hash.
158
159        Returns:
160            Return a hash that will be used as cache.
161        """
162        return cls._generate_hash(hash_dict=hash_dict)
163
164    @classmethod
165    def _generate_hash(cls, hash_dict: dict) -> str:
166        """Generate a hash to be used to storage and retrieve cache.
167
168        It will use pumpJsonDump function from serializers to dump correctly
169        any complex data such as date, geometry and numpy.
170
171        Args:
172            hash_dict (dict):
173                A dictonary with information that will be used on hash.
174
175        Returns:
176            Return a hash that will be used as cache.
177        """
178        str_hash_dict = pumpJsonDump(hash_dict, sort_keys=True)
179        return hashlib.sha512(str_hash_dict).hexdigest()
180
181    def clear(self) -> bool:
182        """Invalidate cache.
183
184        Returns:
185            True is ok.
186        """
187        return self._execute_with_retry(
188            operation=lambda: self._cache.clear(),
189            default=False, operation_name="clear")
190
191    def evict(self, tag_dict: dict) -> bool:
192        """Invalidate cache from a tag.
193
194        Returns:
195            True is ok.
196        """
197        if tag_dict is None:
198            msg = (
199                "At pumpwood_communication cache.evict tag_dict should not be "
200                "'None'. To envict all databse use clear function.")
201            raise PumpWoodCacheError(msg)
202
203        hash_str = self.generate_hash(hash_dict=tag_dict)
204        return self._execute_with_retry(
205            operation=lambda: self._cache.evict(hash_str),
206            default=False, operation_name="evict")
207
208    def get(self, hash_dict: dict) -> Any:
209        """Get a value from cache.
210
211        Args:
212            hash_dict (dict):
213                A dictonary with information that will be used on hash.
214
215        Returns:
216            Return the cached value or None if not found.
217        """
218        if not self._enable:
219            logger.info("Get cache not enable")
220            return None
221
222        # It cache time is set to 0, than disable cache,
223        # this is usefull for testing
224        if self._expire_time == 0:
225            return None
226
227        hash_str = self.generate_hash(hash_dict=hash_dict)
228        return self._execute_with_retry(
229            operation=lambda: self._cache.get(hash_str),
230            default=None, operation_name="get")
231
232    def set(self, hash_dict: dict, value: Any, expire: int = None,
233            tag_dict: dict = None) -> bool:
234        """Set cache value.
235
236        Args:
237            hash_dict (dict):
238                A dictonary with information that will be used on hash.
239            value (Any):
240                Value that will be set on diskcache.
241            expire (int):
242                Number of seconds that will be considered as expirity time.
243            tag_dict (dict):
244                Optional parameter to set a tag to cache. Tagged cache can be
245                envicted together using envict function.
246
247        Returns:
248            Return a boolean value
249        """
250        if not self._enable:
251            logger.info("Set cache not enable")
252            return True
253
254        if hash_dict is None:
255            msg = (
256                "At pumpwood_communication cache.set hash_dict should not be "
257                "'None'")
258            raise PumpWoodCacheError(msg)
259        expire_time = expire or self._expire_time
260        # Do not store cache if expire_time == 0
261        if expire_time == 0:
262            return True
263
264        hash_str = self.generate_hash(hash_dict=hash_dict)
265        tag_str = None
266        if tag_dict is not None:
267            tag_str = self.generate_hash(hash_dict=tag_dict)
268
269        def _do_set() -> bool:
270            return self._cache.set(
271                hash_str, value=value, expire=expire_time,
272                tag=tag_str)
273
274        try:
275            return self._execute_with_retry(
276                operation=_do_set, default=False,
277                operation_name="set")
278        except Exception as error:
279            msg = (
280                'Error when setting cache not associated with lock '
281                'contention. {error}')
282            raise PumpWoodCacheError(
283                message=msg, payload={'error': str(error)})