caches: synced with CE changes

vcsserver/lib/rc_cache/archive_cache.py

@@ -1,72 +1,87 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import logging
 import os
 import diskcache
+from diskcache import RLock
 
 log = logging.getLogger(__name__)
 
 cache_meta = None
 
 
+class ReentrantLock(RLock):
+    def __enter__(self):
+        reentrant_lock_key = self._key
+
+        log.debug('Acquire ReentrantLock(key=%s) for archive cache generation...', reentrant_lock_key)
+        #self.acquire()
+        log.debug('Lock for key=%s acquired', reentrant_lock_key)
+
+    def __exit__(self, *exc_info):
+        #self.release()
+        pass
+
+
 def get_archival_config(config):
+
     final_config = {
         'archive_cache.eviction_policy': 'least-frequently-used'
     }
 
     for k, v in config.items():
         if k.startswith('archive_cache'):
             final_config[k] = v
 
     return final_config
 
 
 def get_archival_cache_store(config):
 
     global cache_meta
     if cache_meta is not None:
         return cache_meta
 
     config = get_archival_config(config)
 
     archive_cache_dir = config['archive_cache.store_dir']
     archive_cache_size_gb = config['archive_cache.cache_size_gb']
     archive_cache_shards = config['archive_cache.cache_shards']
     archive_cache_eviction_policy = config['archive_cache.eviction_policy']
 
     log.debug('Initializing archival cache instance under %s', archive_cache_dir)
 
     # check if it's ok to write, and re-create the archive cache
     if not os.path.isdir(archive_cache_dir):
         os.makedirs(archive_cache_dir, exist_ok=True)
 
     d_cache = diskcache.FanoutCache(
         archive_cache_dir, shards=archive_cache_shards,
         cull_limit=0,  # manual eviction required
         size_limit=archive_cache_size_gb * 1024 * 1024 * 1024,
         eviction_policy=archive_cache_eviction_policy,
         timeout=30
     )
     cache_meta = d_cache
     return cache_meta
 
 
 def includeme(config):
     # init our cache at start, for vcsserver we don't init at runtime
     # because our cache config is sent via wire on make archive call, this call just lazy-enables the client
     return

vcsserver/lib/rc_cache/utils.py

@@ -1,242 +1,247 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import functools
 import logging
 import os
 import threading
 import time
 
 import decorator
 from dogpile.cache import CacheRegion
 
-from vcsserver.lib.rc_cache import region_meta
+
+from vcsserver.utils import sha1
 from vcsserver.str_utils import safe_bytes
 from vcsserver.type_utils import str2bool
-from vcsserver.utils import sha1
+
+from . import region_meta
 
 log = logging.getLogger(__name__)
 
 
 class RhodeCodeCacheRegion(CacheRegion):
 
     def __repr__(self):
         return f'{self.__class__}(name={self.name})'
 
     def conditional_cache_on_arguments(
             self, namespace=None,
             expiration_time=None,
             should_cache_fn=None,
             to_str=str,
             function_key_generator=None,
             condition=True):
         """
         Custom conditional decorator, that will not touch any dogpile internals if
         condition isn't meet. This works a bit different from should_cache_fn
         And it's faster in cases we don't ever want to compute cached values
         """
         expiration_time_is_callable = callable(expiration_time)
         if not namespace:
             namespace = getattr(self, '_default_namespace', None)
 
         if function_key_generator is None:
             function_key_generator = self.function_key_generator
 
         def get_or_create_for_user_func(func_key_generator, user_func, *arg, **kw):
 
             if not condition:
                 log.debug('Calling un-cached method:%s', user_func.__name__)
                 start = time.time()
                 result = user_func(*arg, **kw)
                 total = time.time() - start
                 log.debug('un-cached method:%s took %.4fs', user_func.__name__, total)
                 return result
 
             key = func_key_generator(*arg, **kw)
 
             timeout = expiration_time() if expiration_time_is_callable \
                 else expiration_time
 
             log.debug('Calling cached method:`%s`', user_func.__name__)
             return self.get_or_create(key, user_func, timeout, should_cache_fn, (arg, kw))
 
         def cache_decorator(user_func):
             if to_str is str:
                 # backwards compatible
                 key_generator = function_key_generator(namespace, user_func)
             else:
                 key_generator = function_key_generator(namespace, user_func, to_str=to_str)
 
             def refresh(*arg, **kw):
                 """
                 Like invalidate, but regenerates the value instead
                 """
                 key = key_generator(*arg, **kw)
                 value = user_func(*arg, **kw)
                 self.set(key, value)
                 return value
 
             def invalidate(*arg, **kw):
                 key = key_generator(*arg, **kw)
                 self.delete(key)
 
             def set_(value, *arg, **kw):
                 key = key_generator(*arg, **kw)
                 self.set(key, value)
 
             def get(*arg, **kw):
                 key = key_generator(*arg, **kw)
                 return self.get(key)
 
             user_func.set = set_
             user_func.invalidate = invalidate
             user_func.get = get
             user_func.refresh = refresh
             user_func.key_generator = key_generator
             user_func.original = user_func
 
             # Use `decorate` to preserve the signature of :param:`user_func`.
             return decorator.decorate(user_func, functools.partial(
                 get_or_create_for_user_func, key_generator))
 
         return cache_decorator
 
 
 def make_region(*arg, **kw):
     return RhodeCodeCacheRegion(*arg, **kw)
 
 
 def get_default_cache_settings(settings, prefixes=None):
     prefixes = prefixes or []
     cache_settings = {}
     for key in settings.keys():
         for prefix in prefixes:
             if key.startswith(prefix):
                 name = key.split(prefix)[1].strip()
                 val = settings[key]
                 if isinstance(val, str):
                     val = val.strip()
                 cache_settings[name] = val
     return cache_settings
 
 
 def compute_key_from_params(*args):
     """
     Helper to compute key from given params to be used in cache manager
     """
     return sha1(safe_bytes("_".join(map(str, args))))
 
 
 def custom_key_generator(backend, namespace, fn):
     func_name = fn.__name__
 
     def generate_key(*args):
         backend_pref = getattr(backend, 'key_prefix', None) or 'backend_prefix'
         namespace_pref = namespace or 'default_namespace'
         arg_key = compute_key_from_params(*args)
         final_key = f"{backend_pref}:{namespace_pref}:{func_name}_{arg_key}"
 
         return final_key
 
     return generate_key
 
 
 def backend_key_generator(backend):
     """
     Special wrapper that also sends over the backend to the key generator
     """
     def wrapper(namespace, fn):
         return custom_key_generator(backend, namespace, fn)
     return wrapper
 
 
-def get_or_create_region(region_name, region_namespace: str = None):
-    from vcsserver.lib.rc_cache.backends import FileNamespaceBackend
+def get_or_create_region(region_name, region_namespace: str = None, use_async_runner=False):
+    from .backends import FileNamespaceBackend
+    from . import async_creation_runner
 
     region_obj = region_meta.dogpile_cache_regions.get(region_name)
     if not region_obj:
         reg_keys = list(region_meta.dogpile_cache_regions.keys())
         raise EnvironmentError(f'Region `{region_name}` not in configured: {reg_keys}.')
 
     region_uid_name = f'{region_name}:{region_namespace}'
 
     if isinstance(region_obj.actual_backend, FileNamespaceBackend):
         if not region_namespace:
             raise ValueError(f'{FileNamespaceBackend} used requires to specify region_namespace param')
 
         region_exist = region_meta.dogpile_cache_regions.get(region_namespace)
         if region_exist:
             log.debug('Using already configured region: %s', region_namespace)
             return region_exist
 
         expiration_time = region_obj.expiration_time
 
         cache_dir = region_meta.dogpile_config_defaults['cache_dir']
         namespace_cache_dir = cache_dir
 
         # we default the namespace_cache_dir to our default cache dir.
         # however if this backend is configured with filename= param, we prioritize that
         # so all caches within that particular region, even those namespaced end up in the same path
         if region_obj.actual_backend.filename:
             namespace_cache_dir = os.path.dirname(region_obj.actual_backend.filename)
 
         if not os.path.isdir(namespace_cache_dir):
             os.makedirs(namespace_cache_dir)
         new_region = make_region(
             name=region_uid_name,
             function_key_generator=backend_key_generator(region_obj.actual_backend)
         )
 
         namespace_filename = os.path.join(
             namespace_cache_dir, f"{region_name}_{region_namespace}.cache_db")
         # special type that allows 1db per namespace
         new_region.configure(
             backend='dogpile.cache.rc.file_namespace',
             expiration_time=expiration_time,
             arguments={"filename": namespace_filename}
         )
 
         # create and save in region caches
         log.debug('configuring new region: %s', region_uid_name)
         region_obj = region_meta.dogpile_cache_regions[region_namespace] = new_region
 
     region_obj._default_namespace = region_namespace
+    if use_async_runner:
+        region_obj.async_creation_runner = async_creation_runner
     return region_obj
 
 
 def clear_cache_namespace(cache_region: str | RhodeCodeCacheRegion, cache_namespace_uid: str, method: str):
     from . import CLEAR_DELETE, CLEAR_INVALIDATE
 
     if not isinstance(cache_region, RhodeCodeCacheRegion):
         cache_region = get_or_create_region(cache_region, cache_namespace_uid)
     log.debug('clearing cache region: %s with method=%s', cache_region, method)
 
     num_affected_keys = None
 
     if method == CLEAR_INVALIDATE:
         # NOTE: The CacheRegion.invalidate() method’s default mode of
         # operation is to set a timestamp local to this CacheRegion in this Python process only.
         # It does not impact other Python processes or regions as the timestamp is only stored locally in memory.
         cache_region.invalidate(hard=True)
 
     if method == CLEAR_DELETE:
         cache_keys = cache_region.backend.list_keys(prefix=cache_namespace_uid)
         num_affected_keys = len(cache_keys)
         if num_affected_keys:
             cache_region.delete_multi(cache_keys)
 
     return num_affected_keys