1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114
|
# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright (c) 2015, Enthought, Inc.
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in enthought/LICENSE.txt and may be redistributed only
# under the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
# Thanks for using Enthought open source!
#
# Author: Enthought, Inc.
# -----------------------------------------------------------------------------
from threading import RLock
try:
from collections import OrderedDict
except ImportError:
from ordereddict import OrderedDict
from traits.api import Callable, Event, HasStrictTraits, Instance, Int
class LRUCache(HasStrictTraits):
""" A least-recently used cache.
Items older than `size()` accesses are dropped from the cache.
"""
size = Int
# Called with the key and value that was dropped from the cache
cache_drop_callback = Callable
# This event contains the set of cached cell keys whenever it changes
updated = Event()
_lock = Instance(RLock, args=())
_cache = Instance(OrderedDict)
def __init__(self, size, **traits):
self.size = size
self._initialize_cache()
super(LRUCache, self).__init__(**traits)
def _initialize_cache(self):
with self._lock:
if self._cache is None:
self._cache = OrderedDict()
else:
self._cache.clear()
def _renew(self, key):
with self._lock:
r = self._cache.pop(key)
self._cache[key] = r
return r
# -------------------------------------------------------------------------
# LRUCache interface
# -------------------------------------------------------------------------
def __contains__(self, key):
with self._lock:
return key in self._cache
def __len__(self):
with self._lock:
return len(self._cache)
def __getitem__(self, key):
with self._lock:
return self._renew(key)
def __setitem__(self, key, result):
try:
dropped = None
with self._lock:
self._cache[key] = result
self._renew(key)
if self.size < len(self._cache):
dropped = self._cache.popitem(last=False)
if dropped and self.cache_drop_callback is not None:
self.cache_drop_callback(*dropped)
finally:
self.updated = self.keys()
def get(self, key, default=None):
try:
return self[key]
except KeyError:
return default
def items(self):
with self._lock:
return self._cache.items()
def keys(self):
with self._lock:
return self._cache.keys()
def values(self):
with self._lock:
return self._cache.values()
def clear(self):
with self._lock:
self._initialize_cache()
self.updated = []
|