caches: new updated rc_cache with archive cache module and python3 changes

rhodecode/lib/rc_cache/archive_cache.py

@@ -0,0 +1,88 b''
+# Copyright (C) 2015-2020 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+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
+    settings = config.get_settings()
+    get_archival_cache_store(settings)

rhodecode/lib/rc_cache/__init__.py

@@ -1,89 +1,120 b''
-
-
 # Copyright (C) 2015-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
+import threading
+
 from dogpile.cache import register_backend
+
+from . import region_meta
+from .utils import (
+    ActiveRegionCache,
+    FreshRegionCache,
+    InvalidationContext,
+    backend_key_generator,
+    clear_cache_namespace,
+    get_default_cache_settings,
+    get_or_create_region,
+    make_region,
+    str2bool,
+)
+
 module_name = 'rhodecode'
 
 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 (
-    get_default_cache_settings, backend_key_generator, get_or_create_region,
-    clear_cache_namespace, make_region, InvalidationContext,
-    FreshRegionCache, ActiveRegionCache
-)
-
 
 FILE_TREE_CACHE_VER = 'v4'
 LICENSE_CACHE_VER = 'v2'
 
 
+CLEAR_DELETE = 'delete'
+CLEAR_INVALIDATE = 'invalidate'
+
+
+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)

rhodecode/lib/rc_cache/backends.py

@@ -1,273 +1,299 b''
-
 # Copyright (C) 2015-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
-import time
 import errno
+import fcntl
+import functools
 import logging
-import functools
+import os
+import pickle
+import time
 
+import gevent
 import msgpack
 import redis
-import gevent
-import pickle
-import fcntl
+
 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 file as file_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 dogpile.cache.api import Serializer, Deserializer
-
-from pyramid.settings import asbool
 
 from rhodecode.lib.memory_lru_dict import LRUDict, LRUDictDebug
-from rhodecode.lib.str_utils import safe_str, safe_bytes
-
+from rhodecode.lib.str_utils import safe_bytes, safe_str
+from rhodecode.lib.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)
-        super(LRUMemoryBackend, self).__init__(arguments)
+        arguments['cache_dict'] = LRUDictClass(self.max_size)
+        super().__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
+    serializer: None | Serializer = staticmethod(  # type: ignore
         functools.partial(pickle.dumps, protocol=pickle.HIGHEST_PROTOCOL)
     )
-    deserializer: Union[None, Deserializer] = staticmethod(  # type: ignore
+    deserializer: None | Deserializer = staticmethod(  # type: ignore
         functools.partial(pickle.loads)
     )
 
 
 class MsgPackSerializer(object):
-    serializer: Union[None, Serializer] = staticmethod(  # type: ignore
+    serializer: None | Serializer = staticmethod(  # type: ignore
         msgpack.packb
     )
-    deserializer: Union[None, Deserializer] = staticmethod(  # type: ignore
+    deserializer: None | Deserializer = staticmethod(  # type: ignore
         functools.partial(msgpack.unpackb, use_list=False)
     )
 
 
 class CustomLockFactory(FileLock):
 
     @memoized_property
     def _module(self):
 
         def gevent_flock(fd, operation):
             """
             Gevent compatible flock
             """
             # set non-blocking, this will cause an exception if we cannot acquire a lock
             operation |= fcntl.LOCK_NB
             start_lock_time = time.time()
             timeout = 60 * 15  # 15min
             while True:
                 try:
                     flock_org(fd, operation)
                     # lock has been acquired
                     break
                 except (OSError, IOError) as e:
                     # raise on other errors than Resource temporarily unavailable
                     if e.errno != errno.EAGAIN:
                         raise
                     elif (time.time() - start_lock_time) > timeout:
                         # waited to much time on a lock, better fail than loop for ever
                         log.error('Failed to acquire lock on `%s` after waiting %ss',
                                   self.filename, timeout)
                         raise
                     wait_timeout = 0.03
                     log.debug('Failed to acquire lock on `%s`, retry in %ss',
                               self.filename, wait_timeout)
                     gevent.sleep(wait_timeout)
 
         fcntl.flock = gevent_flock
         return fcntl
 
 
 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)
+            super().__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 _get_keys_pattern(self, prefix: bytes = b''):
+        return b'%b:%b' % (safe_bytes(self.key_prefix), safe_bytes(prefix))
 
     def list_keys(self, prefix: bytes = b''):
-        prefix = b'%b:%b' % (safe_bytes(self.key_prefix), safe_bytes(prefix))
+        prefix = self._get_keys_pattern(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):
-        super(BaseRedisBackend, self).__init__(arguments)
+        self.db_conn = arguments.get('host', '') or arguments.get('url', '') or 'redis-host'
+        super().__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)
+    def _get_keys_pattern(self, prefix: bytes = b''):
+        return b'%b:%b*' % (safe_bytes(self.key_prefix), safe_bytes(prefix))
+
+    def list_keys(self, prefix: bytes = b''):
+        prefix = self._get_keys_pattern(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 rhodecode.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()

rhodecode/lib/rc_cache/cache_key_meta.py

@@ -1,70 +1,71 b''
 
 
 # Copyright (C) 2015-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
-import os
-import sys
 import atexit
 import logging
+import os
 import signal
+import sys
+
 import rhodecode
 
 log = logging.getLogger(__name__)
 
 cache_keys_by_pid = set()
 
 
 def sigHandler(signo, frame):
     """
     signals trigger sys.exit so there's a single handler to cleanup the code.
     """
     if rhodecode.is_test:
         return
 
     sys.exit(0)
 
 
 def free_cache_keys(*args):
-    from rhodecode.model.db import Session, CacheKey
+    from rhodecode.model.db import CacheKey, Session
 
     if rhodecode.is_test:
         return
 
     ssh_cmd = os.environ.get('RC_CMD_SSH_WRAPPER')
     if ssh_cmd:
         return
 
     if cache_keys_by_pid:
         try:
             for cache_proc in set(cache_keys_by_pid):
                 like_expression = '{}%'.format(cache_proc)
                 qry = CacheKey.query().filter(CacheKey.cache_key.like(like_expression))
                 count = qry.count()
                 log.info('Clearing %s: %s cache keys, total: %s', cache_proc, len(cache_keys_by_pid), count)
                 qry.delete(synchronize_session='fetch')
                 cache_keys_by_pid.remove(cache_proc)
             Session().commit()
         except Exception:
             log.exception('Failed to clear keys, exiting gracefully')
 
 atexit.register(free_cache_keys)
 
 signal.signal(signal.SIGTERM, sigHandler)
 signal.signal(signal.SIGINT, sigHandler)
 

rhodecode/lib/rc_cache/region_meta.py

@@ -1,28 +1,27 b''
-
-
 # Copyright (C) 2015-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
+
 import os
 import tempfile
 
 dogpile_config_defaults = {
     'cache_dir': os.path.join(tempfile.gettempdir(), 'rc_cache')
 }
 
 # GLOBAL TO STORE ALL REGISTERED REGIONS
 dogpile_cache_regions = {}

rhodecode/lib/rc_cache/utils.py

@@ -1,370 +1,405 b''
 # Copyright (C) 2015-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
-import os
-import time
-import logging
+
 import functools
-import decorator
+import logging
+import os
 import threading
+import time
 
+import decorator
 from dogpile.cache import CacheRegion
 
 import rhodecode
 from rhodecode.lib.hash_utils import sha1
+from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.lib.type_utils import str2bool
-from rhodecode.lib.str_utils import safe_bytes
 
-from rhodecode.lib.rc_cache import cache_key_meta
-from rhodecode.lib.rc_cache import region_meta
+from . import region_meta, cache_key_meta
 
 log = logging.getLogger(__name__)
 
 
 def isCython(func):
     """
     Private helper that checks if a function is a cython function.
     """
     return func.__class__.__name__ == 'cython_function_or_method'
 
 
 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 than should_cache_fn
+        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(key_generator, user_func, *arg, **kw):
+        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 = key_generator(*arg, **kw)
+            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 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__
+def custom_key_generator(backend, namespace, fn):
+    func_name = fn.__name__
 
     def generate_key(*args):
-        backend_prefix = getattr(backend, 'key_prefix', None) or 'backend_prefix'
+        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 = "{}:{}:{}_{}".format(backend_prefix, namespace_pref, fname, arg_key)
+        final_key = f"{backend_pref}:{namespace_pref}:{func_name}_{arg_key}"
 
         return final_key
 
     return generate_key
 
 
-def get_or_create_region(region_name, region_namespace=None):
-    from rhodecode.lib.rc_cache.backends import FileNamespaceBackend
+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, 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:
-        raise EnvironmentError(
-            'Region `{}` not in configured: {}.'.format(
-                region_name, list(region_meta.dogpile_cache_regions.keys())))
+        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}'
 
-    region_uid_name = '{}:{}'.format(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, "{}.cache.dbm".format(region_namespace))
+            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, 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)
-    num_delete_keys = len(cache_keys)
-    if invalidate:
-        region.invalidate(hard=False)
-    else:
-        if num_delete_keys:
-            region.delete_multi(cache_keys)
-    return num_delete_keys
+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
 
 
 class ActiveRegionCache(object):
     def __init__(self, context, cache_data):
         self.context = context
         self.cache_data = cache_data
 
     def should_invalidate(self):
         return False
 
 
 class FreshRegionCache(object):
     def __init__(self, context, cache_data):
         self.context = context
         self.cache_data = cache_data
 
     def should_invalidate(self):
         return True
 
 
 class InvalidationContext(object):
     """
     usage::
 
         from rhodecode.lib import rc_cache
 
         cache_namespace_uid = CacheKey.SOME_NAMESPACE.format(1)
         region = rc_cache.get_or_create_region('cache_perms', cache_namespace_uid)
 
         @region.conditional_cache_on_arguments(namespace=cache_namespace_uid, condition=True)
         def heavy_compute(cache_name, param1, param2):
             print('COMPUTE {}, {}, {}'.format(cache_name, param1, param2))
 
         # invalidation namespace is shared namespace key for all process caches
         # we use it to send a global signal
         invalidation_namespace = 'repo_cache:1'
 
         inv_context_manager = rc_cache.InvalidationContext(
             uid=cache_namespace_uid, invalidation_namespace=invalidation_namespace)
         with inv_context_manager as invalidation_context:
             args = ('one', 'two')
             # re-compute and store cache if we get invalidate signal
             if invalidation_context.should_invalidate():
                 result = heavy_compute.refresh(*args)
             else:
                 result = heavy_compute(*args)
 
             compute_time = inv_context_manager.compute_time
             log.debug('result computed in %.4fs', compute_time)
 
         # To send global invalidation signal, simply run
         CacheKey.set_invalidate(invalidation_namespace)
 
     """
 
     def __repr__(self):
         return f'<InvalidationContext:{self.cache_key}[{self.uid}]>'
 
     def __init__(self, uid, invalidation_namespace='',
                  raise_exception=False, thread_scoped=None):
         self.uid = uid
         self.invalidation_namespace = invalidation_namespace
         self.raise_exception = raise_exception
         self.proc_id = rhodecode.CONFIG.get('instance_id') or 'DEFAULT'
         self.thread_id = 'global'
 
         if thread_scoped is None:
             # if we set "default" we can override this via .ini settings
-            thread_scoped = str2bool(rhodecode.CONFIG.get('cache_thread_scoped'))
+            thread_scoped = rhodecode.ConfigGet().get_bool('cache_thread_scoped')
 
         # Append the thread id to the cache key if this invalidation context
         # should be scoped to the current thread.
         if thread_scoped is True:
             self.thread_id = threading.current_thread().ident
 
         self.cache_key = compute_key_from_params(uid)
         self.cache_key = 'proc:{}|thread:{}|params:{}'.format(
             self.proc_id, self.thread_id, self.cache_key)
         self.proc_key = 'proc:{}'.format(self.proc_id)
         self.compute_time = 0
 
     def get_or_create_cache_obj(self, cache_type, invalidation_namespace=''):
         from rhodecode.model.db import CacheKey
 
         invalidation_namespace = invalidation_namespace or self.invalidation_namespace
         # fetch all cache keys for this namespace and convert them to a map to find if we
         # have specific cache_key object registered. We do this because we want to have
         # all consistent cache_state_uid for newly registered objects
         cache_obj_map = CacheKey.get_namespace_map(invalidation_namespace)
         cache_obj = cache_obj_map.get(self.cache_key)
         log.debug('Fetched cache obj %s using %s cache key.', cache_obj, self.cache_key)
 
         if not cache_obj:
             new_cache_args = invalidation_namespace
             first_cache_obj = next(iter(cache_obj_map.values())) if cache_obj_map else None
             cache_state_uid = None
             if first_cache_obj:
                 cache_state_uid = first_cache_obj.cache_state_uid
             cache_obj = CacheKey(self.cache_key, cache_args=new_cache_args,
                                  cache_state_uid=cache_state_uid)
             cache_key_meta.cache_keys_by_pid.add(self.proc_key)
 
         return cache_obj
 
     def __enter__(self):
         """
         Test if current object is valid, and return CacheRegion function
         that does invalidation and calculation
         """
         log.debug('Entering cache invalidation check context: %s', self.invalidation_namespace)
         # register or get a new key based on uid
         self.cache_obj = self.get_or_create_cache_obj(cache_type=self.uid)
         cache_data = self.cache_obj.get_dict()
         self._start_time = time.time()
         if self.cache_obj.cache_active:
             # means our cache obj is existing and marked as it's
             # cache is not outdated, we return ActiveRegionCache
             self.skip_cache_active_change = True
 
             return ActiveRegionCache(context=self, cache_data=cache_data)
 
         # the key is either not existing or set to False, we return
         # the real invalidator which re-computes value. We additionally set
         # the flag to actually update the Database objects
         self.skip_cache_active_change = False
         return FreshRegionCache(context=self, cache_data=cache_data)
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        from rhodecode.model.db import Session, IntegrityError
+        from rhodecode.model.db import IntegrityError, Session
 
         # save compute time
         self.compute_time = time.time() - self._start_time
 
         if self.skip_cache_active_change:
             return
 
         try:
             self.cache_obj.cache_active = True
             Session().add(self.cache_obj)
             Session().commit()
         except IntegrityError:
             # if we catch integrity error, it means we inserted this object
             # assumption is that's really an edge race-condition case and
             # it's safe is to skip it
             Session().rollback()
         except Exception:
             log.exception('Failed to commit on cache key update')
             Session().rollback()
             if self.raise_exception:
                 raise