caches: synced with CE code

vcsserver/lib/rc_cache/__init__.py

@@ -1,3 +1,5 b''
+# -*- coding: utf-8 -*-
+
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2020 RhodeCode GmbH
 #
@@ -16,9 +18,20 b''
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import logging
+import threading
 
 from dogpile.cache import register_backend
 
+from . import region_meta
+from .utils import (
+    backend_key_generator,
+    clear_cache_namespace,
+    get_default_cache_settings,
+    get_or_create_region,
+    make_region,
+    str2bool,
+)
+
 module_name = 'vcsserver'
 
 register_backend(
@@ -40,14 +53,18 b' register_backend('
 
 log = logging.getLogger(__name__)
 
-from . import region_meta
-from .utils import (
-    backend_key_generator,
-    clear_cache_namespace,
-    get_default_cache_settings,
-    get_or_create_region,
-    make_region,
-)
+
+def async_creation_runner(cache, somekey, creator, mutex):
+
+    def runner():
+        try:
+            value = creator()
+            cache.set(somekey, value)
+        finally:
+            mutex.release()
+
+    thread = threading.Thread(target=runner)
+    thread.start()
 
 
 def configure_dogpile_cache(settings):
@@ -69,13 +86,20 b' def configure_dogpile_cache(settings):'
 
         new_region = make_region(
             name=namespace_name,
-            function_key_generator=None
+            function_key_generator=None,
+            async_creation_runner=None
         )
 
-        new_region.configure_from_config(settings, 'rc_cache.{}.'.format(namespace_name))
+        new_region.configure_from_config(settings, f'rc_cache.{namespace_name}.')
         new_region.function_key_generator = backend_key_generator(new_region.actual_backend)
+
+        async_creator = str2bool(settings.pop(f'rc_cache.{namespace_name}.async_creator', 'false'))
+        if async_creator:
+            log.debug('configuring region %s with async creator', new_region)
+            new_region.async_creation_runner = async_creation_runner
+
         if log.isEnabledFor(logging.DEBUG):
-            region_args = dict(backend=new_region.actual_backend.__class__,
+            region_args = dict(backend=new_region.actual_backend,
                                region_invalidator=new_region.region_invalidator.__class__)
             log.debug('dogpile: registering a new region `%s` %s', namespace_name, region_args)
 

vcsserver/lib/rc_cache/backends.py

@@ -19,9 +19,11 b' import errno'
 import fcntl
 import functools
 import logging
+import os
 import pickle
-import time
+#import time
 
+#import gevent
 import msgpack
 import redis
 
@@ -34,10 +36,10 b' from dogpile.cache.backends import memor'
 from dogpile.cache.backends import redis as redis_backend
 from dogpile.cache.backends.file import FileLock
 from dogpile.cache.util import memoized_property
-from pyramid.settings import asbool
 
 from vcsserver.lib.memory_lru_dict import LRUDict, LRUDictDebug
 from vcsserver.str_utils import safe_bytes, safe_str
+from vcsserver.type_utils import str2bool
 
 _default_max_size = 1024
 
@@ -49,15 +51,21 b' class LRUMemoryBackend(memory_backend.Me'
     pickle_values = False
 
     def __init__(self, arguments):
-        max_size = arguments.pop('max_size', _default_max_size)
+        self.max_size = arguments.pop('max_size', _default_max_size)
 
         LRUDictClass = LRUDict
         if arguments.pop('log_key_count', None):
             LRUDictClass = LRUDictDebug
 
-        arguments['cache_dict'] = LRUDictClass(max_size)
+        arguments['cache_dict'] = LRUDictClass(self.max_size)
         super(LRUMemoryBackend, self).__init__(arguments)
 
+    def __repr__(self):
+        return f'{self.__class__}(maxsize=`{self.max_size}`)'
+
+    def __str__(self):
+        return self.__repr__()
+
     def delete(self, key):
         try:
             del self._cache[key]
@@ -100,7 +108,11 b' class FileNamespaceBackend(PickleSeriali'
         arguments['lock_factory'] = CustomLockFactory
         db_file = arguments.get('filename')
 
-        log.debug('initialing %s DB in %s', self.__class__.__name__, db_file)
+        log.debug('initialing cache-backend=%s db in %s', self.__class__.__name__, db_file)
+        db_file_dir = os.path.dirname(db_file)
+        if not os.path.isdir(db_file_dir):
+            os.makedirs(db_file_dir)
+
         try:
             super(FileNamespaceBackend, self).__init__(arguments)
         except Exception:
@@ -108,7 +120,10 b' class FileNamespaceBackend(PickleSeriali'
             raise
 
     def __repr__(self):
-        return '{} `{}`'.format(self.__class__, self.filename)
+        return f'{self.__class__}(file=`{self.filename}`)'
+
+    def __str__(self):
+        return self.__repr__()
 
     def list_keys(self, prefix: bytes = b''):
         prefix = b'%b:%b' % (safe_bytes(self.key_prefix), safe_bytes(prefix))
@@ -136,14 +151,22 b' class BaseRedisBackend(redis_backend.Red'
     key_prefix = ''
 
     def __init__(self, arguments):
+        self.db_conn = arguments.get('host', '') or arguments.get('url', '') or 'redis-host'
         super(BaseRedisBackend, self).__init__(arguments)
+
         self._lock_timeout = self.lock_timeout
-        self._lock_auto_renewal = asbool(arguments.pop("lock_auto_renewal", True))
+        self._lock_auto_renewal = str2bool(arguments.pop("lock_auto_renewal", True))
 
         if self._lock_auto_renewal and not self._lock_timeout:
             # set default timeout for auto_renewal
             self._lock_timeout = 30
 
+    def __repr__(self):
+        return f'{self.__class__}(conn=`{self.db_conn}`)'
+
+    def __str__(self):
+        return self.__repr__()
+
     def _create_client(self):
         args = {}
 
@@ -163,7 +186,7 b' class BaseRedisBackend(redis_backend.Red'
         self.reader_client = self.writer_client
 
     def list_keys(self, prefix=''):
-        prefix = '{}:{}*'.format(self.key_prefix, prefix)
+        prefix = f'{self.key_prefix}:{prefix}*'
         return self.reader_client.keys(prefix)
 
     def get_store(self):
@@ -171,7 +194,7 b' class BaseRedisBackend(redis_backend.Red'
 
     def get_mutex(self, key):
         if self.distributed_lock:
-            lock_key = '_lock_{0}'.format(safe_str(key))
+            lock_key = f'_lock_{safe_str(key)}'
             return get_mutex_lock(
                 self.writer_client, lock_key,
                 self._lock_timeout,
@@ -208,10 +231,10 b' def get_mutex_lock(client, lock_key, loc'
             )
 
         def __repr__(self):
-            return "{}:{}".format(self.__class__.__name__, lock_key)
+            return f"{self.__class__.__name__}:{lock_key}"
 
         def __str__(self):
-            return "{}:{}".format(self.__class__.__name__, lock_key)
+            return f"{self.__class__.__name__}:{lock_key}"
 
         def __init__(self):
             self.lock = self.get_lock()

vcsserver/lib/rc_cache/utils.py

@@ -18,6 +18,7 b''
 import functools
 import logging
 import os
+import threading
 import time
 
 import decorator
@@ -25,6 +26,7 b' from dogpile.cache import CacheRegion'
 
 from vcsserver.lib.rc_cache import region_meta
 from vcsserver.str_utils import safe_bytes
+from vcsserver.type_utils import str2bool
 from vcsserver.utils import sha1
 
 log = logging.getLogger(__name__)
@@ -32,6 +34,9 b' 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,
@@ -144,43 +149,58 b' def backend_key_generator(backend):'
 
 
 def key_generator(backend, namespace, fn):
-    fname = fn.__name__
+    func_name = fn.__name__
 
     def generate_key(*args):
         backend_prefix = 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_prefix}:{namespace_pref}:{fname}_{arg_key}"
+        final_key = f"{backend_prefix}:{namespace_pref}:{func_name}_{arg_key}"
 
         return final_key
 
     return generate_key
 
 
-def get_or_create_region(region_name, region_namespace=None):
+def get_or_create_region(region_name, region_namespace: str = None):
     from vcsserver.lib.rc_cache.backends import FileNamespaceBackend
+
     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
-        cache_dir = region_meta.dogpile_config_defaults['cache_dir']
+
         expiration_time = region_obj.expiration_time
 
-        if not os.path.isdir(cache_dir):
-            os.makedirs(cache_dir)
+        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(
-            cache_dir, f"{region_namespace}.cache.dbm")
+            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',
@@ -195,13 +215,18 b' def get_or_create_region(region_name, re'
     return region_obj
 
 
-def clear_cache_namespace(cache_region, cache_namespace_uid, invalidate=False):
-    region = get_or_create_region(cache_region, cache_namespace_uid)
-    cache_keys = region.backend.list_keys(prefix=cache_namespace_uid)
+def clear_cache_namespace(cache_region: str | RhodeCodeCacheRegion, cache_namespace_uid: str, invalidate: bool = False, hard: bool = False):
+    if not isinstance(cache_region, RhodeCodeCacheRegion):
+        cache_region = get_or_create_region(cache_region, cache_namespace_uid)
+
+    cache_keys = cache_region.backend.list_keys(prefix=cache_namespace_uid)
     num_delete_keys = len(cache_keys)
     if invalidate:
-        region.invalidate(hard=False)
+        # 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=hard)
     else:
         if num_delete_keys:
-            region.delete_multi(cache_keys)
+            cache_region.delete_multi(cache_keys)
     return num_delete_keys