"""
Module containing (e.g. LRU) cache functionality for binary_c-python.
We use cachetools when possible because this allows us to set up the
cache of the appropriate size for the task in the grid_options dict.
Please see the LRU_* options in there.
"""
import cachetools
import curses
import functools
import importlib
import inspect
import time
class cache():
class NullCache(cachetools.Cache):
"""
A cachetools cache object that does as little as possible and never matches.
"""
def __init__(self, *args, **kwargs):
return None
def popitem(self):
return # do nothing
def __getitem__(self, key):
return self.__missing__(key)
def __setitem__(self, key, value):
return
def __delitem__(self, key):
return
def __init__(self, **kwargs):
# don't do anything: we just inherit from this class
return
def setup_function_cache(self,vb=False,type=None):
"""
Function to wrap binary_c-python's functions in function cache.
The functions listed in self.grid_options['function_cache_functions'] are
given caches of size self.grid_options['function_cache_size'][func]
Args: None
"""
# choose a type of cache
# add our custom NullCache to the cachetools selection
setattr(cachetools,'NullCache',self.NullCache)
if not self.grid_options["function_cache"]:
# no function cache: set all to NullCache
for func in self.grid_options['function_cache_functions'].keys():
self.function_cache[func] = cachetype(NullCache)
for func in self.grid_options['function_cache_functions'].keys():
(maxsize,cachetype,testargs) = self.grid_options['function_cache_functions'].get(func)
# which cache should we use?
if type:
# use type passed in, if given
usecachetype = type
elif not self.grid_options["function_cache"]:
# function cache is disabled, use NoCache
usecachetype = "NoCache"
else:
if cachetype == None:
# use the default type
usecachetype = self.grid_options["function_cache_default_type"]
else:
# use type passed in
usecachetype = cachetype
if vb:
print("Setup cache for func {func} : maxsize={maxsize}, cachetype={cachetype} -> use {usecachetype}".format(
func=func,
maxsize=maxsize,
cachetype=cachetype,
usecachetype=usecachetype,
))
if usecachetype == 'TTLCache':
extra_cacheargs = [self.grid_options["function_cache_TTL"]]
else:
extra_cacheargs = []
# detect if the function is already wrapped
x = func.split('.')
modulename = 'binarycpython.utils.' + x[0]
module = importlib.import_module(modulename)
_method = eval('module.{}.{}'.format(x[0],x[1]))
_wrapped = getattr(_method,"__wrapped__",False)
# if function is wrapped...
if _wrapped and id(_method) != id(_wrapped):
# save the wrapped function (this calls the cache)
if not (func in self.cached_function_cache):
self.cached_function_cache[func] = _method
self.original_function_cache[func] = _wrapped
if usecachetype == 'NoCache':
# unwrap if we're after NoCache
_code = 'module.{}.{} = _wrapped'.format(x[0],x[1])
exec(_code)
else:
# function isn't wrapped, which means it was previously
# unwrapped, so rewrap it if not using NoCache
if usecachetype != 'NoCache':
_code = 'module.{}.{} = self.cached_function_cache["{}"]'.format(x[0],x[1],func)
exec(_code)
# check we're not still wrapped
_method = eval('module' + '.' + x[0] + '.' + x[1])
_wrapped = getattr(_method,"__wrapped__",False)
# if NoCache (explicity use no cache), just use NullCache
# (it's never actually set)
if usecachetype == "NoCache":
cachetools_func = getattr(cachetools,"NullCache")
else:
cachetools_func = getattr(cachetools,usecachetype)
if maxsize == 0:
maxsize = self.grid_options['function_cache_default_maxsize']
if vb:
print("Make function cache for func {func}, maxsize {maxsize}".format(
func=func,
maxsize=maxsize
))
# set up cache function args
if maxsize is None:
args = [2]
else:
args = [maxsize]
args += extra_cacheargs
# clear any existing cache
if func in self.caches:
try:
self.caches[func].cache_clear()
except:
pass
del self.caches[func]
# set up new cache using the appropriate cachetools function
if usecachetype != "NoCache":
self.caches[func] = cachetools_func(*args)
def test_caches(self,dt=5.0):
"""
Function to test cache speeds of the functions that binary_c-python automatically caches.
Args:
dt (default 5) in seconds the length of each test. Long is more accurate, but takes longer (obviously).
"""
cachetypes = ("NullCache","FIFOCache","LRUCache","TTLCache","NoCache")
functions = self.grid_options['function_cache_functions'].keys()
maxsizes = (0,1,2,4,8,16,32,64,128,256)
stdscr = curses.initscr()
#curses.noecho()
stdscr.clear()
width = len(maxsizes) * 9
height = (len(cachetypes) + 2) * len(functions)
def _coord(n,x,y,xoffset,yoffset):
# map n,x,y,xoffset,yoffset to screen coords
_x = (x + xoffset) * 9
_y = (len(cachetypes) + 4) * n + (y + yoffset)
return (_x,_y)
def _print(n,x,y,xoffset,yoffset,str):
# print str at n,x,y,offset
_c = _coord(n,x,y,xoffset,yoffset)
stdscr.addstr(_c[1],_c[0],str)
stdscr.refresh()
self.grid_options['function_cache'] = True
for n,func in enumerate(functions):
_print(n,0,0,0,0,
"Cache speed test of function {func}".format(func=func))
for x,maxsize in enumerate(maxsizes):
_print(n,x,0,2,1,str(maxsize))
best = 0
best_type = None
best_maxsize = None
for y,type in enumerate(cachetypes):
_print(n,0,y,0,2,type)
self.grid_options['function_cache_default_type'] = type
self.setup_function_cache()
(maxsize,cachetype,testargs) = self.grid_options['function_cache_functions'].get(func)
x = func.split('.')
modulename = 'binarycpython.utils.' + x[0]
module = importlib.import_module(modulename)
_method = eval('module.{}.{}'.format(x[0],x[1]))
if testargs:
for x,maxsize in enumerate(maxsizes):
if type == 'NoCache' and maxsize > 0:
continue
t0 = time.time()
count = 0
while time.time() - t0 < dt:
_method(self,*testargs)
count += 1
if count < 99999:
_print(n,x,y,2,2,"{:d}".format(count))
else:
_print(n,x,y,2,2,"{:.2e}".format(float(count)))
if count > best:
best = count
best_type = type
best_maxsize = maxsize
_print(n,0,len(cachetypes)+2,0,0,"Best cache type {type} with maxsize {maxsize}".format(type=best_type,maxsize=best_maxsize))
stdscr.refresh()
stdscr.getkey()
stdscr.keypad(0)
curses.echo()
print("\npress any key to exit\n")
curses.endwin()