Add a new insideout mode

Suor · Suor · commit ba86b3a23542 · 2023-02-25T12:33:32.000+07:00
The idea is that instead of saving all dependent cache keys in conj set we put a simple random stamp in those and store a checksum of all related conj stamps along with cache data. This makes cache reads more complicated - `MGET` key + conj keys, validate stamps checksum. However, we no longer need to store potentially big conj sets and invalidation becomes faster, including model level invalidation. It also removes strong link between conj and cache keys, i.e. loss of conj keys no longer leads to a stale cache, instead we will simply drop the key on next read. This opens easier way for maxmemory and cluster. So: - more friendly to `maxmemory`, even assumes that, see #143 - eliminates issues with big conj sets and long invalidation, see #340, - `reapconjs` is not needed with it, see #323, #434 Followups: - docs - remove `CACHEOPS_LRU` as it's superseeded by this generally - make insideout default or even drop the old ways?
diff --git a/README.rst b/README.rst
@@ -75,8 +75,7 @@ Setup redis connection and enable caching for desired models:
         ...                                  # everything else is passed to Sentinel()
     }
 
-    # To use your own redis client class,
-    # should be compatible or subclass cacheops.redis.CacheopsRedis
+    # Use your own redis client class, should be compatible or subclass redis.StrictRedis
     CACHEOPS_CLIENT_CLASS = 'your.redis.ClientClass'
 
     CACHEOPS = {
diff --git a/bench.py b/bench.py
@@ -16,7 +16,7 @@ def run_benchmarks(tests):
         if 'h' in flags:
             print(HEADER_TEMPLATE % name)
         time = bench_test(test)
-        print('%-18s time: %.2fms' % (name, time * 1000))
+        print('%-18s time: %.3fms' % (name, time * 1000))
 
 def bench_test(test):
     prepared = None
@@ -81,6 +81,9 @@ def bench_once(test, prepared=None):
     db_name = connection.creation.create_test_db(verbosity=verbosity, autoclobber=not interactive)
     call_command('loaddata', *fixtures, **{'verbosity': verbosity})
 
+    from cacheops.redis import redis_client
+    redis_client.flushdb()
+
     from tests.bench import TESTS  # import is here because it executes queries
     if selector:
         tests = [(name, test) for name, test in TESTS if select(name)]
diff --git a/cacheops/conf.py b/cacheops/conf.py
@@ -16,6 +16,7 @@ class Defaults:
     CACHEOPS = {}
     CACHEOPS_PREFIX = lambda query: ''
     CACHEOPS_LRU = False
+    CACHEOPS_INSIDEOUT = False
     CACHEOPS_CLIENT_CLASS = None
     CACHEOPS_DEGRADE_ON_FAILURE = False
     CACHEOPS_SENTINEL = {}
diff --git a/cacheops/getset.py b/cacheops/getset.py
@@ -0,0 +1,143 @@
+from contextlib import contextmanager
+import hashlib
+import json
+
+from .conf import settings
+from .redis import redis_client, handle_connection_failure, load_script
+from .transaction import transaction_states
+
+
+LOCK_TIMEOUT = 60
+
+
+@handle_connection_failure
+def cache_thing(prefix, cache_key, data, cond_dnfs, timeout, dbs=(), precall_key='',
+                expected_checksum=''):
+    """
+    Writes data to cache and creates appropriate invalidators.
+
+    If precall_key is not the empty string, the data will only be cached if the
+    precall_key is set to avoid caching stale data.
+
+    If expected_checksum is set and does not match the actual one then cache won't be written.
+    """
+    # Could have changed after last check, sometimes superficially
+    if transaction_states.is_dirty(dbs):
+        return
+
+    if settings.CACHEOPS_INSIDEOUT:
+        schemes = dnfs_to_schemes(cond_dnfs)
+        conj_keys = dnfs_to_conj_keys(prefix, cond_dnfs)
+        return load_script('cache_thing_insideout')(
+            keys=[prefix, cache_key],
+            args=[
+                settings.CACHEOPS_SERIALIZER.dumps(data),
+                json.dumps(schemes),
+                json.dumps(conj_keys),
+                timeout,
+                expected_checksum,
+            ]
+        )
+    else:
+        if prefix and precall_key == "":
+            precall_key = prefix
+        load_script('cache_thing', settings.CACHEOPS_LRU)(
+            keys=[prefix, cache_key, precall_key],
+            args=[
+                settings.CACHEOPS_SERIALIZER.dumps(data),
+                json.dumps(cond_dnfs, default=str),
+                timeout
+            ]
+        )
+
+
+@contextmanager
+def getting(key, cond_dnfs, prefix, lock=False):
+    if not lock:
+        yield _read(key, cond_dnfs, prefix)
+    else:
+        locked = False
+        try:
+            data = _get_or_lock(key, cond_dnfs, prefix)
+            locked = data is None
+            yield data
+        finally:
+            if locked:
+                _release_lock(key)
+
+
+@handle_connection_failure
+def _read(key, cond_dnfs, prefix):
+    if not settings.CACHEOPS_INSIDEOUT:
+        return redis_client.get(key)
+
+    conj_keys = dnfs_to_conj_keys(prefix, cond_dnfs)
+    coded, *stamps = redis_client.mget(key, *conj_keys)
+    if coded is None or coded == b'LOCK':
+        return coded
+
+    if None in stamps:
+        redis_client.unlink(key)
+        return
+
+    stamp_checksum, data = coded.split(b':', 1)
+    if stamp_checksum.decode() != join_stamps(stamps):
+        redis_client.unlink(key)
+        return None
+
+    return data
+
+
+@handle_connection_failure
+def _get_or_lock(key, cond_dnfs, prefix):
+    _lock = redis_client.register_script("""
+        local locked = redis.call('set', KEYS[1], 'LOCK', 'nx', 'ex', ARGV[1])
+        if locked then
+            redis.call('del', KEYS[2])
+        end
+        return locked
+    """)
+    signal_key = key + ':signal'
+
+    while True:
+        data = _read(key, cond_dnfs, prefix)
+        if data is None:
+            if _lock(keys=[key, signal_key], args=[LOCK_TIMEOUT]):
+                return None
+        elif data != b'LOCK':
+            return data
+
+        # No data and not locked, wait
+        redis_client.brpoplpush(signal_key, signal_key, timeout=LOCK_TIMEOUT)
+
+
+@handle_connection_failure
+def _release_lock(key):
+    _unlock = redis_client.register_script("""
+        if redis.call('get', KEYS[1]) == 'LOCK' then
+            redis.call('del', KEYS[1])
+        end
+        redis.call('lpush', KEYS[2], 1)
+        redis.call('expire', KEYS[2], 1)
+    """)
+    signal_key = key + ':signal'
+    _unlock(keys=[key, signal_key])
+
+
+# Key manipulation helpers
+
+def join_stamps(stamps):
+    return hashlib.sha1(b' '.join(stamps)).hexdigest()
+
+
+def dnfs_to_conj_keys(prefix, cond_dnfs):
+    def _conj_cache_key(table, conj):
+        conj_str = '&'.join(f'{field}={val}' for field, val in sorted(conj.items()))
+        return f'{prefix}conj:{table}:{conj_str}'
+
+    return [_conj_cache_key(table, conj) for table, disj in cond_dnfs.items()
+                                         for conj in disj]
+
+def dnfs_to_schemes(cond_dnfs):
+    return {table: [",".join(sorted(conj)) for conj in disj]
+            for table, disj in cond_dnfs.items() if disj}
diff --git a/cacheops/invalidation.py b/cacheops/invalidation.py
@@ -1,6 +1,6 @@
 import json
 import threading
-from funcy import memoize, post_processing, ContextDecorator, decorator
+from funcy import memoize, post_processing, ContextDecorator, decorator, walk_values
 from django.db import DEFAULT_DB_ALIAS
 from django.db.models.expressions import F, Expression
 
@@ -27,12 +27,17 @@ def skip_on_no_invalidation(call):
 def invalidate_dict(model, obj_dict, using=DEFAULT_DB_ALIAS):
     if no_invalidation.active or not settings.CACHEOPS_ENABLED:
         return
+
     model = model._meta.concrete_model
     prefix = get_prefix(_cond_dnfs=[(model._meta.db_table, list(obj_dict.items()))], dbs=[using])
-    load_script('invalidate')(keys=[prefix], args=[
-        model._meta.db_table,
-        json.dumps(obj_dict, default=str)
-    ])
+
+    if settings.CACHEOPS_INSIDEOUT:
+        script = 'invalidate_insideout'
+        serialized_dict = json.dumps(walk_values(str, obj_dict))
+    else:
+        script = 'invalidate'
+        serialized_dict = json.dumps(obj_dict, default=str)
+    load_script(script)(keys=[prefix], args=[model._meta.db_table, serialized_dict])
     cache_invalidated.send(sender=model, obj_dict=obj_dict)
 
 
@@ -60,9 +65,12 @@ def invalidate_model(model, using=DEFAULT_DB_ALIAS):
     prefix = get_prefix(tables=[model._meta.db_table], dbs=[using])
     conjs_keys = redis_client.keys('%sconj:%s:*' % (prefix, model._meta.db_table))
     if conjs_keys:
-        cache_keys = redis_client.sunion(conjs_keys)
-        keys = list(cache_keys) + conjs_keys
-        redis_client.unlink(*keys)
+        if settings.CACHEOPS_INSIDEOUT:
+            redis_client.unlink(*conjs_keys)
+        else:
+            cache_keys = redis_client.sunion(conjs_keys)
+            keys = list(cache_keys) + conjs_keys
+            redis_client.unlink(*keys)
     cache_invalidated.send(sender=model, obj_dict=None)
 
 
diff --git a/cacheops/lua/cache_thing_insideout.lua b/cacheops/lua/cache_thing_insideout.lua
@@ -0,0 +1,35 @@
+local prefix = KEYS[1]
+local key = KEYS[2]
+local data = ARGV[1]
+local schemes = cjson.decode(ARGV[2])
+local conj_keys = cjson.decode(ARGV[3])
+local timeout = tonumber(ARGV[4])
+local expected_checksum = ARGV[5]
+
+-- Ensure schemes are known
+for db_table, _schemes in pairs(schemes) do
+    redis.call('sadd', prefix .. 'schemes:' .. db_table, unpack(_schemes))
+end
+
+-- Fill in invalidators and collect stamps
+local stamps = {}
+local rnd = tostring(math.random())  -- A new value for empty stamps
+for _, conj_key in ipairs(conj_keys) do
+    local stamp = redis.call('set', conj_key, rnd, 'nx', 'get') or rnd
+    table.insert(stamps, stamp)
+    -- NOTE: an invalidator should live longer than any key it references.
+    --       So we update its ttl on every key if needed.
+    redis.call('expire', conj_key, timeout, 'gt')
+end
+
+-- Write data to cache along with a checksum of the stamps to see if any of them changed
+local all_stamps = table.concat(stamps, ' ')
+local stamp_checksum = redis.sha1hex(all_stamps)
+
+if expected_checksum ~= '' and stamp_checksum ~= expected_checksum then
+    -- Cached data was invalidated during the function call. The data is
+    -- stale and should not be cached.
+    return stamp_checksum -- This one is used for keep_fresh implementation
+end
+
+redis.call('set', key, stamp_checksum .. ':' .. data, 'ex', timeout)
diff --git a/cacheops/lua/invalidate_insideout.lua b/cacheops/lua/invalidate_insideout.lua
@@ -0,0 +1,20 @@
+local prefix = KEYS[1]
+local db_table = ARGV[1]
+local obj = cjson.decode(ARGV[2])
+
+local conj_cache_key = function (db_table, scheme, obj)
+    local parts = {}
+    for field in string.gmatch(scheme, "[^,]+") do
+        -- All obj values are strings, we still use tostring() in case obj does not contain field
+        table.insert(parts, field .. '=' .. tostring(obj[field]))
+    end
+
+    return prefix .. 'conj:' .. db_table .. ':' .. table.concat(parts, '&')
+end
+
+-- Drop conj keys
+local conj_keys = {}
+local schemes = redis.call('smembers', prefix .. 'schemes:' .. db_table)
+for _, scheme in ipairs(schemes) do
+    redis.call('unlink', conj_cache_key(db_table, scheme, obj))
+end
diff --git a/cacheops/query.py b/cacheops/query.py
@@ -1,5 +1,4 @@
 import sys
-import json
 import threading
 from random import random
 
@@ -24,8 +23,8 @@
 from .conf import model_profile, settings, ALL_OPS
 from .utils import monkey_mix, stamp_fields, get_cache_key, cached_view_fab, family_has_profile
 from .utils import md5
+from .getset import cache_thing, getting
 from .sharding import get_prefix
-from .redis import redis_client, handle_connection_failure, load_script
 from .tree import dnfs
 from .invalidation import invalidate_obj, invalidate_dict, skip_on_no_invalidation
 from .transaction import transaction_states
@@ -37,29 +36,6 @@
 _local_get_cache = {}
 
 
-@handle_connection_failure
-def cache_thing(prefix, cache_key, data, cond_dnfs, timeout, dbs=(), precall_key=''):
-    """
-    Writes data to cache and creates appropriate invalidators.
-
-    If precall_key is not the empty string, the data will only be cached if the
-    precall_key is set to avoid caching stale data.
-    """
-    # Could have changed after last check, sometimes superficially
-    if transaction_states.is_dirty(dbs):
-        return
-    if prefix and precall_key == "":
-        precall_key = prefix
-    load_script('cache_thing', settings.CACHEOPS_LRU)(
-        keys=[prefix, cache_key, precall_key],
-        args=[
-            settings.CACHEOPS_SERIALIZER.dumps(data),
-            json.dumps(cond_dnfs, default=str),
-            timeout
-        ]
-    )
-
-
 def cached_as(*samples, timeout=None, extra=None, lock=None, keep_fresh=False):
     """
     Caches results of a function and invalidates them same way as given queryset(s).
@@ -91,7 +67,7 @@ def _get_queryset(sample):
 
     querysets = lmap(_get_queryset, samples)
     dbs = list({qs.db for qs in querysets})
-    cond_dnfs = join_with(lcat, map(dnfs, querysets))
+    cond_dnfs = join_with(lcat, map(dnfs, querysets))  # TODO: use cached version?
     qs_keys = [qs._cache_key(prefix=False) for qs in querysets]
     if timeout is None:
         timeout = min(qs._cacheprofile['timeout'] for qs in querysets)
@@ -108,12 +84,22 @@ def wrapper(*args, **kwargs):
             extra_val = extra(*args, **kwargs) if callable(extra) else extra
             cache_key = prefix + 'as:' + get_cache_key(func, args, kwargs, qs_keys, extra_val)
 
-            with redis_client.getting(cache_key, lock=lock) as cache_data:
+            with getting(cache_key, cond_dnfs, prefix, lock=lock) as cache_data:
                 cache_read.send(sender=None, func=func, hit=cache_data is not None)
                 if cache_data is not None:
                     return settings.CACHEOPS_SERIALIZER.loads(cache_data)
                 else:
-                    if keep_fresh:
+                    precall_key = ''
+                    expected_checksum = ''
+                    if keep_fresh and settings.CACHEOPS_INSIDEOUT:
+                        # The conj stamps should not be dropped while we calculate the function.
+                        # But being filled in concurrently is a normal concurrent cache write.
+                        # However, if they are filled in and then dropped, we cannot detect that.
+                        # Unless we fill them ourselves and get expected checksum now. We also need
+                        # to fill in schemes, so we just reuse the cache_thing().
+                        expected_checksum = cache_thing(prefix, cache_key, '', cond_dnfs, timeout,
+                                                        dbs=dbs, expected_checksum='never match')
+                    elif keep_fresh:
                         # We call this "asp" for "as precall" because this key is
                         # cached before the actual function is called. We randomize
                         # the key to prevent falsely thinking the key was not
@@ -126,12 +112,10 @@ def wrapper(*args, **kwargs):
                         # only if it remains valid before, during, and after the
                         # call, the result can be cached and returned.
                         cache_thing(prefix, precall_key, 'PRECALL', cond_dnfs, timeout, dbs=dbs)
-                    else:
-                        precall_key = ''
 
                     result = func(*args, **kwargs)
                     cache_thing(prefix, cache_key, result, cond_dnfs, timeout, dbs=dbs,
-                                precall_key=precall_key)
+                                precall_key=precall_key, expected_checksum=expected_checksum)
                     return result
 
         return wrapper
@@ -275,7 +259,7 @@ def _fetch_all(self):
         cache_key = self._cache_key()
         lock = self._cacheprofile['lock']
 
-        with redis_client.getting(cache_key, lock=lock) as cache_data:
+        with getting(cache_key, self._cond_dnfs, self._prefix, lock=lock) as cache_data:
             cache_read.send(sender=self.model, func=None, hit=cache_data is not None)
             if cache_data is not None:
                 self._result_cache = settings.CACHEOPS_SERIALIZER.loads(cache_data)
diff --git a/cacheops/redis.py b/cacheops/redis.py
diff --git a/tests/settings.py b/tests/settings.py
diff --git a/tests/tests.py b/tests/tests.py
diff --git a/tox.ini b/tox.ini

Original file line number	Diff line number	Diff line change
`@@ -75,8 +75,7 @@ Setup redis connection and enable caching for desired models:`
`75`	`75`	`... # everything else is passed to Sentinel()`
`76`	`76`	`}`
`77`	`77`
`78`		`- # To use your own redis client class,`
`79`		`- # should be compatible or subclass cacheops.redis.CacheopsRedis`
	`78`	`+ # Use your own redis client class, should be compatible or subclass redis.StrictRedis`
`80`	`79`	`CACHEOPS_CLIENT_CLASS = 'your.redis.ClientClass'`
`81`	`80`
`82`	`81`	`CACHEOPS = {`