''' The module provide ring cache.
Ring cache provides a cache to any objects, but with
limitation of the maximum number of the object.
Using this instead of the usual ``dictionary`` as cache,
developer may control the maximum use of the memory.
.. codeauthor:: Yoshifumi Futaana
'''
import sys
import collections
import logging
_logger = logging.getLogger(__name__)
[docs]class RingCache:
""" A ring cache
>>> rcache = RingCache(5, name="mycache")
>>> print(rcache)
<Ring cache (mycache): Size=0 / Max=5>
>>> rcache.add(30, 'thirty')
>>> print(len(rcache))
1
>>> rcache.add(20, None)
>>> rcache.add(10, None)
>>> print(len(rcache))
3
>>> print(rcache.get(10))
None
>>> rcache.add(11, 'Eleven')
>>> rcache.add(12, 'tolv')
>>> rcache.add(13, '13')
>>> rcache.add(14, '14')
>>> rcache.add(15, '15')
>>> print(len(rcache))
5
>>> print(rcache.get(13))
13
>>> print(rcache.hasKey(30))
False
>>> rcache.get(30) # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
...
KeyError:
>>> rcache.show_statistics(file=sys.stdout)
Name=mycache
Max=5
Size=5
Added=8
Popped=3
Get (total)=3
Get (success)=2
Inquired=1
>>> rcache.clearCache()
>>> print(rcache)
<Ring cache (mycache): Size=0 / Max=5>
>>> print(15 in rcache)
False
>>> rcache.add(15, 'Femton')
>>> print(rcache.get(15))
Femton
>>> rcache.show_statistics(file=sys.stdout)
Name=mycache
Max=5
Size=1
Added=1
Popped=0
Get (total)=1
Get (success)=1
Inquired=1
"""
def __init__(self, max_cache, name=None):
''' Instance a ring cache.
:param max_cache: The maximum cache to be stored.
'''
self.max = max_cache
self.cache = collections.OrderedDict()
self._name = name
if name is None:
self._name = 'RingCache:' + hex(id(self))
self._statistics = {"added": 0,
"popped": 0,
"get_total": 0,
"get_cached": 0,
"inquired": 0,
}
_logger.debug("Ring cache ``{}``: Created with size {}.".format(self._name, self.max))
[docs] def add(self, key, obj):
""" Add the object to cache
"""
if len(self.cache) == self.max:
self.cache.popitem(last=False)
self._statistics['popped'] += 1
self.cache[key] = obj
self._statistics['added'] += 1
_logger.debug("Ring cache ``{}``: Added with key ``{}``. Size {}/{}".format(self._name, key, len(self.cache), self.max))
[docs] def clearCache(self):
''' Clear the cache
'''
self.cache = collections.OrderedDict()
self._statistics = {"added": 0,
"popped": 0,
"get_total": 0,
"get_cached": 0,
"inquired": 0,
}
_logger.debug("Ring cache ``{}``: Cleared with size {}.".format(self._name, self.max))
[docs] def consistency_check(self):
pass
def __len__(self):
return len(self.cache)
[docs] def size(self):
return len(self.cache)
[docs] def cachesize(self):
return len(self.cache)
[docs] def hasKey(self, key):
""" Return if the ``key`` is in cache
:param key: Key of the data
:return: True/False if the key-data pair is/is-not in the cache.
"""
self._statistics["inquired"] += 1
return (key in self.cache)
[docs] def get(self, key):
""" Get the value corresponding to the ``key``
:param key: Key of the data
:return: The data
It raises ``KeyError`` if the key-data pair is not in the cache.
"""
self._statistics['get_total'] += 1
try:
obj = self.cache[key]
_logger.debug("Ring cache ``{}``: Key {} found.".format(self._name, key))
except:
_logger.debug("Ring cache ``{}``: Key {} not found.".format(self._name, key))
raise
self._statistics['get_cached'] += 1
return obj
[docs] def showBuffer(self, file=sys.stderr):
""" Show the buffer information
:keyword file: A file object to dump. Deafult is sys.stderr.
"""
print("Not supported", file=file)
[docs] def show_statistics(self, file=sys.stderr):
print('Name={}'.format(self._name), file=file)
print('Max={}'.format(self.max), file=file)
print('Size={}'.format(len(self)), file=file)
print('Added={}'.format(self._statistics["added"]), file=file)
print('Popped={}'.format(self._statistics["popped"]), file=file)
print('Get (total)={}'.format(self._statistics["get_total"]), file=file)
print('Get (success)={}'.format(self._statistics["get_cached"]), file=file)
print('Inquired={}'.format(self._statistics["inquired"]), file=file)
def __contains__(self, item):
return self.hasKey(item)
def __str__(self):
s = "<Ring cache ({}): Size={:d} / Max={:d}>".format(self._name, len(self), self.max)
return s
[docs]class SimpleRingCache():
''' A simple ring cache (dictionary-like).
>>> rcache = SimpleRingCache(5)
>>> rcache.add(30, 'thirty')
>>> print(len(rcache))
1
>>> rcache.add(20, None)
>>> rcache.add(10, None)
>>> print(len(rcache))
3
>>> print(rcache.get(10))
None
>>> rcache.add(11, 'Eleven')
>>> rcache.add(12, 'tolv')
>>> rcache.add(13, '13')
>>> rcache.add(14, '14')
>>> rcache.add(15, '15')
>>> print(len(rcache))
5
>>> print(rcache.get(13))
13
>>> print(rcache.hasKey(30))
False
'''
[docs] class UnfilledRingCache():
def __init__(self, max_cache):
self.max = max_cache
self.clearCache()
[docs] def clearCache(self):
self.id2key = {}
self.key2obj = {}
self.cur = 0
[docs] def add(self, key, obj):
self.consistency_check()
if key in self.key2obj:
raise RuntimeError('Key already in cache')
self.id2key[self.cur] = key
self.key2obj[key] = obj
self.cur = self.cur + 1
def __len__(self):
self.consistency_check()
return len(self.id2key)
[docs] def size(self):
return len(self)
[docs] def cachesize(self):
return self.max
[docs] def consistency_check(self):
if len(self.id2key) != len(self.key2obj):
_logger.warning('Internal data base inconsistent.'
'Data length for id2key(%d)!=key2obj(%d)' %
(len(self.id2key), len(self.key2obj)))
raise RuntimeError(
'Data base not consistent by unknown reason.')
[docs] def hasKey(self, key):
return key in self.key2obj
[docs] def get(self, key):
return self.key2obj[key]
[docs] def showBuffer(self, file=sys.stderr):
print('UnfilledBuffer with max size %d' % self.max, file=file)
print('Current size %d' % len(self), file=file)
print('Current position %d' % self.cur, file=file)
print('id2key: \n%s' % self.id2key, file=file)
print('key2obj: \n%s' % self.key2obj, file=file)
[docs] class FilledRingCache():
def __init__(self, unfilledRingCache):
self.max = unfilledRingCache.max
self.id2key = unfilledRingCache.id2key
self.key2obj = unfilledRingCache.key2obj
self.cur = 0
[docs] def size(self):
return len(self)
[docs] def cachesize(self):
return self.max
[docs] def hasKey(self, key):
return key in self.key2obj
[docs] def get(self, key):
return self.key2obj[key]
[docs] def add(self, key, obj):
self.consistency_check()
if key in self.key2obj:
raise RuntimeError('Key already in cache')
# Removing old key-value.
oldkey = self.id2key[self.cur]
self.key2obj.pop(oldkey)
# Update
self.key2obj[key] = obj
self.id2key[self.cur] = key
# Increment
self.cur = (self.cur + 1) % self.max
def __len__(self):
self.consistency_check()
return len(self.id2key)
[docs] def consistency_check(self):
if len(self.id2key) != self.max:
_logger.warning('Internal data base inconsistent.'
'Data length (%d) must equal to the maximum buffer (%d)' %
(len(self.id2key), self.max))
raise RuntimeError(
'Data base not consistent by unknown reason.')
if len(self.id2key) != len(self.key2obj):
_logger.warning('Internal data base inconsistent.'
'Data length for id2key(%d)!=key2obj(%d)' %
(len(self.id2key), len(self.key2obj)))
raise RuntimeError(
'Data base not consistent by unknown reason.')
[docs] def showBuffer(self, file=sys.stderr):
print('FilledBuffer with max size %d' % self.max, file=file)
print('Current size %d' % len(self), file=file)
print('Current position %d' % self.cur, file=file)
print('id2key: \n%s' % self.id2key, file=file)
print('key2obj: \n%s' % self.key2obj, file=file)
def __init__(self, max_cache):
''' Instance a ring cache.
@param max_cache The maximum cache to be stored.
'''
self.max = max_cache
self.cache = self.UnfilledRingCache(self.max)
[docs] def add(self, key, obj):
# If this addition will fill the buffer
if len(self.cache) == self.max - 1:
self.cache.add(key, obj)
self.cache = self.FilledRingCache(self.cache)
else:
self.cache.add(key, obj)
[docs] def clearCache(self):
''' Clear the cache
'''
self.cache = self.UnfilledRingCache(self.max)
[docs] def consistency_check(self):
self.cache.consistency_check()
def __len__(self):
return len(self.cache)
[docs] def size(self):
return self.cache.size()
[docs] def cachesize(self):
return self.cache.cachesize()
[docs] def hasKey(self, key):
""" Return if the ``key`` is in cache
:param key: Key of the data
:return: True/False if the key-data pair is/is-not in the cache.
"""
return self.cache.hasKey(key)
[docs] def get(self, key):
""" Get the value corresponding to the ``key``
:param key: Key of the data
:return: The data
It raises ``KeyError`` if the key-data pair is not in the cache.
"""
return self.cache.get(key)
[docs] def showBuffer(self, file=sys.stderr):
""" Show the buffer information
:keyword file: A file object to dump. Deafult is sys.stderr.
"""
self.cache.showBuffer(file=file)
def __contains__(self, item):
return self.hasKey(item)