"""LRU cache adaptation, decoupled from functools.lru_cache internals so it can be reused directly."""
from typing import Generic, Optional, TypeVar
_K = TypeVar('_K') #:
_V = TypeVar('_V') #:
[docs]
class LRUCache(Generic[_K, _V]):
"""Constants shares by all lru cache instances.
Adapted from functools.lru_cache.
Attributes:
sentinel: Unique object used to signal cache misses.
PREV: Name for the link field 0.
NEXT: Name for the link field 1.
KEY: Name for the link field 2.
RESULT: Name for the link field 3.
"""
sentinel = object()
PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 # names for the link fields
def __init__(self, maxsize: Optional[int] = None):
from _thread import RLock
self.cache = {}
self.hits = self.misses = 0
self.full = False
self.cache_get = self.cache.get # bound method to lookup a key or return None
self.cache_len = self.cache.__len__ # get cache size without calling len()
self.lock = RLock() # because linkedlist updates aren't threadsafe
self.root = [] # root of the circular doubly linked list
self.root[:] = [self.root, self.root, None, None] # initialize by pointing to self
self.maxsize = maxsize
[docs]
def __contains__(self, item: _K) -> bool:
with self.lock:
link = self.cache_get(item)
if link is not None:
self.hits += 1
return True
self.misses += 1
return False
[docs]
def __getitem__(self, item: _K) -> _V:
with self.lock:
link = self.cache_get(item)
if link is not None:
# Move the link to the front of the circular queue
link_prev, link_next, _key, result = link
link_prev[LRUCache.NEXT] = link_next
link_next[LRUCache.PREV] = link_prev
last = self.root[LRUCache.PREV]
last[LRUCache.NEXT] = self.root[LRUCache.PREV] = link
link[LRUCache.PREV] = last
link[LRUCache.NEXT] = self.root
return result
[docs]
def __setitem__(self, key: _K, value: _V):
with self.lock:
if key in self.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 self.full:
# Use the old root to store the new key and result.
oldroot = self.root
oldroot[LRUCache.KEY] = key
oldroot[LRUCache.RESULT] = value
# 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.
self.root = oldroot[LRUCache.NEXT]
oldkey = self.root[LRUCache.KEY]
_oldresult = self.root[LRUCache.RESULT] # noqa: F841
self.root[LRUCache.KEY] = self.root[LRUCache.RESULT] = None
# Now update the cache dictionary.
del self.cache[oldkey]
# Save the potentially reentrant cache[key] assignment
# for last, after the root and links have been put in
# a consistent state.
self.cache[key] = oldroot
else:
# Put result in a new link at the front of the queue.
last = self.root[LRUCache.PREV]
link = [last, self.root, key, value]
last[LRUCache.NEXT] = self.root[LRUCache.PREV] = self.cache[key] = link
# Use the cache_len bound method instead of the len() function
# which could potentially be wrapped in an lru_cache itself.
if self.maxsize is not None:
self.full = (self.cache_len() >= self.maxsize)
[docs]
def cache_info(self):
"""Report cache statistics."""
with self.lock:
from collections import namedtuple
return namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"])(
self.hits, self.misses, self.maxsize, self.cache_len())
[docs]
def cache_clear(self):
"""Clear the cache and cache statistics."""
with self.lock:
self.cache.clear()
self.root[:] = [self.root, self.root, None, None]
self.hits = self.misses = 0
self.full = False