tileradotorg/synapse
1
0
Fork 0
forked from MirrorHub/synapse
Code Pull requests Activity

Merge branch 'erikj/cache_deferreds' into erikj/dictionary_cache

Browse source
This commit is contained in:
Erik Johnston 2015-08-06 14:03:15 +01:00
commit 1d08bf7c17
1 changed files with 27 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
synapse/storage/_base.py
View file

@ -149,6 +149,9 @@ class Cache(object):
class CacheDescriptor(object): class CacheDescriptor(object):
""" A method decorator that applies a memoizing cache around the function. """ A method decorator that applies a memoizing cache around the function.
This caches deferreds, rather than the results themselves. Deferreds that
fail are removed from the cache.
The function is presumed to take zero or more arguments, which are used in The function is presumed to take zero or more arguments, which are used in
a tuple as the key for the cache. Hits are served directly from the cache; a tuple as the key for the cache. Hits are served directly from the cache;
misses use the function body to generate the value. misses use the function body to generate the value.
@ -195,8 +198,23 @@ class CacheDescriptor(object):
arg_dict = inspect.getcallargs(self.orig, obj, *args, **kwargs) arg_dict = inspect.getcallargs(self.orig, obj, *args, **kwargs)
keyargs = [arg_dict[arg_nm] for arg_nm in self.arg_names] keyargs = [arg_dict[arg_nm] for arg_nm in self.arg_names]
try: try:
cached_result = cache.get(*keyargs) cached_result_d = cache.get(*keyargs)
return cached_result.observe()
if DEBUG_CACHES:
@defer.inlineCallbacks
def check_result(cached_result):
actual_result = yield self.function_to_call(obj, *args, **kwargs)
if actual_result != cached_result:
logger.error(
"Stale cache entry %s%r: cached: %r, actual %r",
self.orig.__name__, keyargs,
cached_result, actual_result,
)
raise ValueError("Stale cache entry")
cached_result_d.observe().addCallback(check_result)
return cached_result_d.observe()
except KeyError: except KeyError:
# Get the sequence number of the cache before reading from the # Get the sequence number of the cache before reading from the
# database so that we can tell if the cache is invalidated # database so that we can tell if the cache is invalidated
@ -207,8 +225,14 @@ class CacheDescriptor(object):
self.function_to_call, self.function_to_call,
obj, *args, **kwargs obj, *args, **kwargs
) )
ret = ObservableDeferred(ret, consumeErrors=False)
def onErr(f):
cache.invalidate(*keyargs)
return f
ret.addErrback(onErr)
ret = ObservableDeferred(ret, consumeErrors=False)
cache.update(sequence, *(keyargs + [ret])) cache.update(sequence, *(keyargs + [ret]))
return ret.observe() return ret.observe()