Created
February 6, 2025 13:05
-
-
Save JamesParrott/6a479e9ddad6e515430d695ba436b49b to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def lru_cache(maxsize=128, typed=False, initial_data = None): | |
"""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(decimal.Decimal("3.0")) and f(3.0) will be treated as | |
distinct calls with distinct results. Some types such as str and int may | |
be cached separately even when typed is false. | |
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: https://en.wikipedia.org/wiki/Cache_replacement_policies#Least_recently_used_(LRU) | |
""" | |
# Users should only access the lru_cache through its public API: | |
# cache_info, cache_clear, and f.__wrapped__ | |
# The internals of the lru_cache are encapsulated for thread safety and | |
# to allow the implementation to change (including a possible C version). | |
if isinstance(maxsize, int): | |
# Negative maxsize is treated as 0 | |
if maxsize < 0: | |
maxsize = 0 | |
elif callable(maxsize) and isinstance(typed, bool): | |
# The user_function was passed in directly via the maxsize argument | |
user_function, maxsize = maxsize, 128 | |
wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo, initial_data) | |
wrapper.cache_parameters = lambda : {'maxsize': maxsize, 'typed': typed} | |
return update_wrapper(wrapper, user_function) | |
elif maxsize is not None: | |
raise TypeError( | |
'Expected first argument to be an integer, a callable, or None') | |
def decorating_function(user_function): | |
wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo, initial_data) | |
wrapper.cache_parameters = lambda : {'maxsize': maxsize, 'typed': typed} | |
return update_wrapper(wrapper, user_function) | |
return decorating_function | |
def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo, initial_data): | |
# Constants shared by all lru cache instances: | |
sentinel = object() # unique object used to signal cache misses | |
make_key = _make_key # build a key from the function arguments | |
PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 # names for the link fields | |
# Is this better expressed as list[tuple[ParamSpec, Any]] ? | |
# initial_data: list[tuple[tuple[Any,dict[str,Any]],Any] | |
if initial_data is None: | |
initial_data = [] | |
cache = {} | |
hits = misses = 0 | |
full = False | |
cache_get = cache.get # bound method to lookup a key or return None | |
cache_len = cache.__len__ # get cache size without calling len() | |
lock = RLock() # because linkedlist updates aren't threadsafe | |
root = [] # root of the circular doubly linked list | |
root[:] = [root, root, None, None] # initialize by pointing to self | |
if maxsize == 0: | |
def cache_put(returned_value, args, kwds): | |
# It could be argued this should raise an error. | |
# But making it a no-op makes it is possible to | |
# set maxsize=0 to disable the cache for | |
# experimentation and debugging. | |
pass | |
def wrapper(*args, **kwds): | |
# No caching -- just a statistics update | |
nonlocal misses | |
misses += 1 | |
result = user_function(*args, **kwds) | |
return result | |
elif maxsize is None: | |
def cache_put(returned_value, args, kwds): | |
# The call signature is a little more elaborate | |
# than wrapper and user_function | |
# (args and kwds are not unpacked with * and **). | |
# This means functions that have their own arg named | |
# "returned_value" can still be wrapped with lru_cache | |
cache[key] = result | |
def wrapper(*args, **kwds): | |
# Simple caching without ordering or size limit | |
nonlocal hits, misses | |
key = make_key(args, kwds, typed) | |
result = cache_get(key, sentinel) | |
if result is not sentinel: | |
hits += 1 | |
return result | |
misses += 1 | |
result = user_function(*args, **kwds) | |
cache_put(result, args, kwds) | |
return result | |
else: | |
def cache_put(returned_value, args, kwds): | |
# Put returned_value in the finite cache, expiring the | |
# LRU entry and updating the LRU order. | |
nonlocal root, hits, misses, full | |
with lock: | |
if key in cache: | |
# Getting here means that this same key was added to the | |
# cache while the lock was released. Since the link | |
# update is already done, we need only return the | |
# computed result and update the count of misses. | |
pass | |
elif full: | |
# Use the old root to store the new key and result. | |
oldroot = root | |
oldroot[KEY] = key | |
oldroot[RESULT] = result | |
# Empty the oldest link and make it the new root. | |
# Keep a reference to the old key and old result to | |
# prevent their ref counts from going to zero during the | |
# update. That will prevent potentially arbitrary object | |
# clean-up code (i.e. __del__) from running while we're | |
# still adjusting the links. | |
root = oldroot[NEXT] | |
oldkey = root[KEY] | |
oldresult = root[RESULT] | |
root[KEY] = root[RESULT] = None | |
# Now update the cache dictionary. | |
del cache[oldkey] | |
# Save the potentially reentrant cache[key] assignment | |
# for last, after the root and links have been put in | |
# a consistent state. | |
cache[key] = oldroot | |
else: | |
# Put result in a new link at the front of the queue. | |
last = root[PREV] | |
link = [last, root, key, result] | |
last[NEXT] = root[PREV] = cache[key] = link | |
# Use the cache_len bound method instead of the len() function | |
# which could potentially be wrapped in an lru_cache itself. | |
full = (cache_len() >= maxsize) | |
def wrapper(*args, **kwds): | |
# Size limited caching that tracks accesses by recency | |
nonlocal root, hits, misses, full | |
key = make_key(args, kwds, typed) | |
with lock: | |
link = cache_get(key) | |
if link is not None: | |
# Move the link to the front of the circular queue | |
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 | |
hits += 1 | |
return result | |
misses += 1 | |
result = user_function(*args, **kwds) | |
cache_put(result, args, kwds, strict = False) | |
return result | |
for [(args, kwds), returned_value] in initial_data: | |
cache_put( | |
returned_value=returned_value, | |
args = args, | |
kwds = kwds, | |
) | |
def cache_info(): | |
"""Report cache statistics""" | |
with lock: | |
return _CacheInfo(hits, misses, maxsize, cache_len()) | |
def cache_clear(): | |
"""Clear the cache and cache statistics""" | |
nonlocal hits, misses, full | |
with lock: | |
cache.clear() | |
root[:] = [root, root, None, None] | |
hits = misses = 0 | |
full = False | |
wrapper.cache_info = cache_info | |
wrapper.cache_clear = cache_clear | |
wrapper.cache_put = cache_put | |
return wrapper |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Linted but untested.