caches: synced with CE code

vcsserver/lib/rc_cache/__init__.py

@@ -1,86 +1,110 b''
+# -*- coding: utf-8 -*-
+
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2020 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 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(
     "dogpile.cache.rc.memory_lru", f"{module_name}.lib.rc_cache.backends",
     "LRUMemoryBackend")
 
 register_backend(
     "dogpile.cache.rc.file_namespace", f"{module_name}.lib.rc_cache.backends",
     "FileNamespaceBackend")
 
 register_backend(
     "dogpile.cache.rc.redis", f"{module_name}.lib.rc_cache.backends",
     "RedisPickleBackend")
 
 register_backend(
     "dogpile.cache.rc.redis_msgpack", f"{module_name}.lib.rc_cache.backends",
     "RedisMsgPackBackend")
 
 
 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):
     cache_dir = settings.get('cache_dir')
     if cache_dir:
         region_meta.dogpile_config_defaults['cache_dir'] = cache_dir
 
     rc_cache_data = get_default_cache_settings(settings, prefixes=['rc_cache.'])
 
     # inspect available namespaces
     avail_regions = set()
     for key in rc_cache_data.keys():
         namespace_name = key.split('.', 1)[0]
         if namespace_name in avail_regions:
             continue
 
         avail_regions.add(namespace_name)
         log.debug('dogpile: found following cache regions: %s', namespace_name)
 
         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)
 
         region_meta.dogpile_cache_regions[namespace_name] = new_region
 
 
 def includeme(config):
     configure_dogpile_cache(config.registry.settings)

vcsserver/lib/rc_cache/backends.py

@@ -1,238 +1,261 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2020 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 errno
 import fcntl
 import functools
 import logging
+import os
 import pickle
-import time
+#import time
 
+#import gevent
 import msgpack
 import redis
 
 flock_org = fcntl.flock
 from typing import Union
 
 from dogpile.cache.api import Deserializer, Serializer
 from dogpile.cache.backends import file as file_backend
 from dogpile.cache.backends import memory as memory_backend
 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
 
 log = logging.getLogger(__name__)
 
 
 class LRUMemoryBackend(memory_backend.MemoryBackend):
     key_prefix = 'lru_mem_backend'
     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]
         except KeyError:
             # we don't care if key isn't there at deletion
             pass
 
     def delete_multi(self, keys):
         for key in keys:
             self.delete(key)
 
 
 class PickleSerializer:
     serializer: Union[None, Serializer] = staticmethod(  # type: ignore
         functools.partial(pickle.dumps, protocol=pickle.HIGHEST_PROTOCOL)
     )
     deserializer: Union[None, Deserializer] = staticmethod(  # type: ignore
         functools.partial(pickle.loads)
     )
 
 
 class MsgPackSerializer(object):
     serializer: Union[None, Serializer] = staticmethod(  # type: ignore
         msgpack.packb
     )
     deserializer: Union[None, Deserializer] = staticmethod(  # type: ignore
         functools.partial(msgpack.unpackb, use_list=False)
     )
 
 
 class CustomLockFactory(FileLock):
 
     pass
 
 
 class FileNamespaceBackend(PickleSerializer, file_backend.DBMBackend):
     key_prefix = 'file_backend'
 
     def __init__(self, arguments):
         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:
             log.exception('Failed to initialize db at: %s', db_file)
             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))
 
         def cond(dbm_key: bytes):
             if not prefix:
                 return True
 
             if dbm_key.startswith(prefix):
                 return True
             return False
 
         with self._dbm_file(True) as dbm:
             try:
                 return list(filter(cond, dbm.keys()))
             except Exception:
                 log.error('Failed to fetch DBM keys from DB: %s', self.get_store())
                 raise
 
     def get_store(self):
         return self.filename
 
 
 class BaseRedisBackend(redis_backend.RedisBackend):
     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 = {}
 
         if self.url is not None:
             args.update(url=self.url)
 
         else:
             args.update(
                 host=self.host, password=self.password,
                 port=self.port, db=self.db
             )
 
         connection_pool = redis.ConnectionPool(**args)
         self.writer_client = redis.StrictRedis(
             connection_pool=connection_pool
         )
         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):
         return self.reader_client.connection_pool
 
     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,
                 auto_renewal=self._lock_auto_renewal
             )
         else:
             return None
 
 
 class RedisPickleBackend(PickleSerializer, BaseRedisBackend):
     key_prefix = 'redis_pickle_backend'
     pass
 
 
 class RedisMsgPackBackend(MsgPackSerializer, BaseRedisBackend):
     key_prefix = 'redis_msgpack_backend'
     pass
 
 
 def get_mutex_lock(client, lock_key, lock_timeout, auto_renewal=False):
     from vcsserver.lib._vendor import redis_lock
 
     class _RedisLockWrapper(object):
         """LockWrapper for redis_lock"""
 
         @classmethod
         def get_lock(cls):
             return redis_lock.Lock(
                 redis_client=client,
                 name=lock_key,
                 expire=lock_timeout,
                 auto_renewal=auto_renewal,
                 strict=True,
             )
 
         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()
             self.lock_key = lock_key
 
         def acquire(self, wait=True):
             log.debug('Trying to acquire Redis lock for key %s', self.lock_key)
             try:
                 acquired = self.lock.acquire(wait)
                 log.debug('Got lock for key %s, %s', self.lock_key, acquired)
                 return acquired
             except redis_lock.AlreadyAcquired:
                 return False
             except redis_lock.AlreadyStarted:
                 # refresh thread exists, but it also means we acquired the lock
                 return True
 
         def release(self):
             try:
                 self.lock.release()
             except redis_lock.NotAcquired:
                 pass
 
     return _RedisLockWrapper()

vcsserver/lib/rc_cache/utils.py

@@ -1,207 +1,232 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2020 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.str_utils import safe_bytes
+from vcsserver.type_utils import str2bool
 from vcsserver.utils import sha1
 
 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 function_key_generator is None:
             function_key_generator = self.function_key_generator
 
         def get_or_create_for_user_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 = 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 backend_key_generator(backend):
     """
     Special wrapper that also sends over the backend to the key generator
     """
     def wrapper(namespace, fn):
         return key_generator(backend, namespace, fn)
     return wrapper
 
 
 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',
             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
 
     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