################################################################################ ### The following code gets implemented in C ################################### ################################################################################ # In C, make this a static function def make_key(args, kwds, typed, kwd_mark = (object(),)): # build a cache key from positional and keyword args key = args if kwds: sorted_items = tuple(sorted(kwds.items())) key += kwd_mark + sorted_items if typed: key += tuple([type(v) for v in args]) if kwds: key += tuple([type(v) for k, v in sorted_items]) return key class c_lru_cache: def __init__(self, maxsize, typed, cache_info_type, user_function): ''' Create a cached callable that wraps another function. maxsize: 0 for no caching None for unlimited cache size n for a bounded cache typed: False cache f(3) and f(3.0) as identical calls True cache f(3) and f(3.0) as distinct calls cache_info_type namedtuple class with the fields: hits misses currsize maxsize user_function the function being cached All attributes should be regarded as private. The API should only be accessed by its methods (which as all public). ''' self.cache = {} self.hits = 0 # int or PyInt_Object self.misses = 0 # int of PyInt_Object self.maxsize = maxsize # py_ssize_t self.typed = typed self.user_function = user_function self.cache_info_type = cache_info_type self.root = root = [] root[:] = [root, root, None, None] def __call__(self, *args, **kwds): # Names for the link fields # In C, these are constants created using #def PREV = 0 NEXT = 1 KEY = 2 RESULT = 3 # In C replace the use of cache.get(key, sentinel) # with a NULL check on a call to PyDict_GetItem() sentinel = object() if self.maxsize == 0: # no caching, just a statistics update after a successful call result = self.user_function(*args, **kwds) self.misses += 1 return result elif self.maxsize is None: # simple caching without ordering or size limit key = make_key(args, kwds, self.typed) result = self.cache.get(key, sentinel) if result is not sentinel: self.hits += 1 return result result = self.user_function(*args, **kwds) self.cache[key] = result self.misses += 1 return result else: # size limited caching that tracks accesses by recency # Critical section (do all these operations together while holding the GIL) key = make_key(args, kwds, self.typed) link = self.cache.get(key) if link is not None: # move the link to the front of the circular queue root = self.root link_prev, link_next, key, result = link link_prev[NEXT] = link_next link_next[PREV] = link_prev last = root[PREV] last[NEXT] = root[PREV] = link link[PREV] = last link[NEXT] = root self.hits += 1 return result # End critical section result = self.user_function(*args, **kwds) # Critical section (do all these operations together while holding the GIL) root = self.root if len(self.cache) < self.maxsize: # put result in a new link at the front of the queue last = root[PREV] link = [last, root, key, result] self.cache[key] = last[NEXT] = root[PREV] = link else: # use root to store the new key and result root[KEY] = key root[RESULT] = result self.cache[key] = root # empty the oldest link and make it the new root self.root = root = root[NEXT] del self.cache[root[KEY]] root[KEY] = root[RESULT] = None self.misses += 1 # End critical section return result def cache_info(self): """Report cache statistics""" return self.cache_info_type(self.hits, self.misses, self.maxsize, len(self.cache)) def cache_clear(self): """Clear the cache and cache statistics""" self.cache.clear() # Implement the above as: do {PyDict_Clear(d);} while (len(d) > 0) # This eliminates a race condition so that we know the cache is empty # when we're holding the GIL and doing the following critical section. root = self.root root[:] = [root, root, None, None] self.hits = self.misses = 0 ################################################################################ ### The following code gets implemented in pure Python ######################### ################################################################################ from collections import namedtuple from functools import update_wrapper _CacheInfo = namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"]) def lru_cache(maxsize=100, typed=False): """Least-recently-used cache decorator. If *maxsize* is set to None, the LRU features are disabled and the cache can grow without bound. If *typed* is True, arguments of different types will be cached separately. For example, f(3.0) and f(3) will be treated as distinct calls with distinct results. Arguments to the cached function must be hashable. View the cache statistics named tuple (hits, misses, maxsize, currsize) with f.cache_info(). Clear the cache and statistics with f.cache_clear(). Access the underlying function with f.__wrapped__. See: http://en.wikipedia.org/wiki/Cache_algorithms#Least_Recently_Used """ def decorating_function(user_function): wrapper = c_lru_cache(maxsize, typed, _CacheInfo, user_function) return update_wrapper(wrapper, user_function) return decorating_function