ceph/src/pybind/mgr/dashboard/plugins/ttl_cache.py

   1 """
   2 This is a minimal implementation of TTL-ed lru_cache function.
   3
   4 Based on Python 3 functools and backports.functools_lru_cache.
   5 """
   6 from __future__ import absolute_import
   7
   8 from functools import wraps
   9 from collections import OrderedDict
  10 from threading import RLock
  11 from time import time
  12
  13
  14 def ttl_cache(ttl, maxsize=128, typed=False):
  15     if typed is not False:
  16         raise NotImplementedError("typed caching not supported")
  17
  18     def decorating_function(function):
  19         cache = OrderedDict()
  20         stats = [0, 0, 0]
  21         rlock = RLock()
  22         setattr(
  23             function,
  24             'cache_info',
  25             lambda:
  26             "hits={}, misses={}, expired={}, maxsize={}, currsize={}".format(
  27                 stats[0], stats[1], stats[2], maxsize, len(cache)))
  28
  29         @wraps(function)
  30         def wrapper(*args, **kwargs):
  31             key = args + tuple(kwargs.items())
  32             with rlock:
  33                 refresh = True
  34                 if key in cache:
  35                     (ret, ts) = cache[key]
  36                     del cache[key]
  37                     if time() - ts < ttl:
  38                         refresh = False
  39                         stats[0] += 1
  40                     else:
  41                         stats[2] += 1
  42
  43                 if refresh:
  44                     ret = function(*args, **kwargs)
  45                     ts = time()
  46                     if len(cache) == maxsize:
  47                         cache.popitem(last=False)
  48                     stats[1] += 1
  49
  50                 cache[key] = (ret, ts)
  51
  52             return ret
  53
  54         return wrapper
  55     return decorating_function