LRU Cache在Python中的实现

LRU Cache - Least Recently Used Cache 最近最久未使用缓存

今天问了同事一个问题，LRU Cache系统如何实现，同事答使用时间戳。那么使用时间戳的话，他可能的想法是在Python里的字典中实现，这样通过判断时间戳的迟早来实现LRU。

首先缓存，比如对于一个这样的函数,task(arg1, arg2)，缓存主要是缓存其参数和返回值，那么就可以直接使用字典来实现，关键在于参数是可以哈希的，也就是说参数不应该是可变的，这样就可以将参数作为key，函数返回值作为value来存储到字典之中。

那么对于LRU，LRU比较常用的在操作系统内存管理的页面置换中使用到了，就是最近使用的页面在内存中，否则就在硬盘里，为了提高页面的命中率，其英文为hits，没有命中则为misses。对于上面的这个函数来说就是要存储最近使用的参数和返回值这么一个对应关系。

时间戳字典可行吗？

对于同事提到的字典，其缓存模型应该是函数参数作为key，因为要通过key来判断是否命中，时间戳也作为key，因为要比较最近使用，比如缓存已经满了，要将最早使用的缓存清除出去，那么这时候就要比较哪个参数对应的时间是最早的。

这里有一种multi_key_dict，也就是多个键对应一个值，假设时间戳不会冲突的话，使用时间戳和参数作为键，那么要取到最早的时间戳，也需要一个排序，因为键在字典中是无序的。且不说从这么多键中找出时间戳这种键，并且还有一个问题是如果参数也是时间戳呢= = !

使用list

既然要取时间戳需要排序，那么不用时间戳，直接将对应的key也就是函数参数存储到一个Python列表中就行了，最近使用的在列表头，最早使用的在列表尾，然后用一个字典存储参数-返回值，这样就实现了一个LRU嘛。

如果有参数命中(key在字典中)，那么就把这个参数从其原始位置放到列表头，如果满了就删除最后一个元素。这样看起来可以，但是涉及性能问题。从列表中间pop和从头insert都是性能瓶颈操作，时间复杂度在O(n)。

使用链表

既然使用list在insert时候有性能问题，那么使用单链表呢，只有有头指针，我就可以在头部插入，此时时间复杂度只有O(1)。但是将参数从单链表尾部移除呢，还得从头遍历一遍到尾部，那那那就用双向循环链表好了，这样找尾部元素，就可以直接使用头指针向前得到了。

看起来好像一切OK了，但是认真想想，如果命中一个参数，那么要将其从中间某个位置弄到最开头呢？双向循环链表还是得去遍历一遍链表去查找，平均的时间复杂度也要O(n)

使用字典和链表相结合

在取缓存的时候，通过字典直接从参数取到返回值，那么可不可以将链表和字典结合起来呢，使用key能不能直接获取到key在链表的位置呢？当然可以，key对应的value存储为这个key对应的链表节点不就可以了，同时这个链表节点还存储了返回值不就刚好可以将返回值取到了嘛。

functools.lru_cache

Python中functools的lru_cache实现了LRU Cache，核心是一个双向循环链表再加一个字典。

from functools import lru_cache可以看到lru_cache的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
_CacheInfo = namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"])
def lru_cache(maxsize=128, typed=False):
    '''最近最少使用的缓存装饰器
    如果maxsize为None，则不使用LRU特性，并且缓存可以没有限制的增长
    如果typed为True，不同类型的参数将会独立的缓存，比如f(3.0)和f(3)将被看作不同的调用，并且有不同的结果。
    缓存函数的参数必须是可哈希的。
    访问命名元组(hints, misses, maxsize, currsize)来查看缓存统计数据，使用f.cache_info()来得到。使用f.cache_clear()来清除缓存数据。使用f.__wrapped__来获取底层包装函数
    '''
    if maxsize is not None and not isinstance(maxsize, int):
        raise TypeError('Expected maxsize to be an integer or None')
    def decorating_function(user_function):
        wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)
        return update_wrapper(wrapper, user_function)
    return decorating_function

首先这是一个带参数的装饰器，使用的时候应该@lru_cache()，这样会返回decorating_function，用来装饰用户的函数，其内部会返回将用户的包装后的函数，使用的是_lru_cache_wrapper:

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
def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo):
    # 所有lru缓存实例共享的常量
    sentinel = object()          # 当缓存没有命中的时候返回的对象
    make_key = _make_key         # 从函数参数来生成一个key的函数
    PREV, NEXT, KEY, RESULT = 0, 1, 2, 3   # 每一个链表节点中存储的每个元素位置
    cache = {}
    hits = misses = 0
    full = False
    cache_get = cache.get  # 绑定查询键的函数
    lock = RLock()  # 链表的更新可能不是线程安全的，对于一个函数，可能多线程调用
    root = []   # 双向循环链表的头指针
    root[:] = [root, root, None, None]  # 初始化双向循环链表
    if maxsize == 0:
        # 不使用缓存，每次调用没有命中次数都会加一
        def wrapper(*args, **kwds):
            # 重新绑定自由变量misses
            nonlocal misses
            result = user_function(*args, **kwds)
            misses += 1
            return result
    elif maxsize is None:
        # 简单的缓存，没有次序，也没有大小限制，也就是直接使用字典进行缓存
        def wrapper(*args, **kwds):
            nonlocal hits, misses
            key = make_key(args, kwds, typed)
            result = cache_get(key, sentinel)
            if result is not sentinel:
                hits += 1
                return result
            result = user_function(*args, **kwds)
            cache[key] = result
            misses += 1
            return result
    else:
        # 这种情况就是有大小限制的lru_cache，我觉得应该加一个有限制的普通缓存系统
        def wrapper(*args, **kwds):
            nonlocal root, hits, misses, full
            key = make_key(args, kwds, typed)
            with lock:
                link = cache_get(key)
                # 如果命中，将命中的链表节点放到头部，每个链表节点存储了PREV,NEXT,KEY,VALUE
                # 所以先将当前节点的前后节点接在一起
                if link is not None:
                    # 从链表中取出这个节点
                    link_prev, link_next, _key, result = link
                    link_prev[NEXT] = link_next
                    link_next[PREV] = link_prev
                    # 将链表节点放到头部,root前面
                    last = root[PREV]
                    last[NEXT] = root[PREV] = link
                    link[PREV] = last
                    link[NEXT] = root]
                    hits += 1
                    return result
                result = user_function(*args, **kwds)
                # 没有命中的话，将其放入缓存，更新操作必须加锁
                with lock:
                    if key in cache:
                        # 如果key已经在cache，说明有其它的线程已经将这个key放入缓存中了
                        pass
                    elif full:
                        # 如果缓存已经满了，则取出最久没有使用的节点将其pop出去，也就是双向循环链表中root后面的指针。
                        # 这段代码的具体过程是，将root指针作为新节点，root指针后面的节点作为root节点
                        oldroot = root
                        oldroot[KEY] = key
                        oldroot[RESULT] = result
                        # root的节点作为新节点，只需要将key，result换掉
                        root = oldroot[NEXT]
                        # 记录oldkey是为了在cache字典中删除
                        oldkey = root[KEY]
                        oldresult = root[RESULT]
                        root[KEY] = root[RESULT] = None
                        # 更新好root节点后删除字典中的key
                        del cache[oldkey]
                        cache[key] = oldroot
                    else:
                        # 如果缓存没有满，更新缓存
                        last = root[PREV]
                        # 新的链表节点在root和root前一个节点之间
                        link = [last, root, key, result]
                        last[NEXT] = root[PREV] = cache[key] = link
                        full = (len(cache) >= maxsize)
                    misses += 1
                return result
        def cache_info():
            """Report cache statistics"""
            with lock:
                # 因为以下数据可能在更新，所以加锁获取共享资源
                return _CacheInfo(hits, misses, maxsize, len(cache))
        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
        return wrapper

当缓存已经满了的时候，这种处理方式还是挺不错的，但是要是我来搞就是删除后面的节点，然后将新节点放到root前面。另外这个人是左撇子么，喜欢逆时针转啊。

OrderedDict相关

另一个使用双向循环链表的例子是OrderedDict，链表保存了key添加到字典的顺序，但其实这个字典实际用的时候，是用的C语言实现的版本：

1
2
3
4
5
try:
    from _collections import OrderedDict
except ImportError:
    # Leave the pure Python version in place.
    pass

下面是OrderedDict的源码：

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
class OrderedDict(dict):
    '''记录插入顺序的字典
    内部的self.__map字典记录了key到双向循环链表的节点，双向循环链表开始和结束都是一个sentinel元素。这个元素从来都不会被删除，它在self.__hardroot, 并且self.__hardroot在self.__root有一个弱引用代理
    往前的连接是弱引用代理，为了防止循环引用
    在self.__map中存储了硬连接到单个链表节点上，当key被删除后，这些硬连接就会消失
    '''
    def __init__(*args, **kwds):
    '''Initialize an ordered dictionary.  The signature is the same as
    regular dictionaries, but keyword arguments are not recommended because
    their insertion order is arbitrary.
    '''
    if not args:
        raise TypeError("descriptor '__init__' of 'OrderedDict' object "
                        "needs an argument")
    self, *args = args
    if len(args) > 1:
        raise TypeError('expected at most 1 arguments, got %d' % len(args))
    try:
        self.__root
    except AttributeError:
        # 这里使用的是一个限定属性的类来作为链表节点
        self.__hardroot = _Link()
        self.__root = root = _proxy(self.__hardroot)
        root.prev = root.next = root
        self.__map = {}
    self.__update(*args, **kwds)

这里_Link()是一个限定属性的类，代码如下；

1
2
class _Link(object):
    __slots__ = 'prev', 'next', 'key', '__weakref__'

_proxy是from _weakref import proxy as _proxy 对一个对象的弱引用。相对于通常的引用来说，如果一个对象有一个常规的引用，它是不会被垃圾收集器销毁的，但是如果一个对象只剩下一个弱引用，那么它可能被垃圾收集器收回。werkref

1
2
3
4
5
def proxy(p_object, callback=None):
    """
    创建一个代理对象，弱引用至object
    """
    pass

继续来看OrderedDict:

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
def __setitem__(self, key, value, dict_setitem=dict.__setitem__, proxy=_proxy, Link=_Link):
    # 当设置一个新值时，首先查看是否已经在字典中了，如果在字典中，就不需要再加入到链表中了，保持之前的次序
    # 如果没有，则加入到链表中，同样是将新节点放到root前面一个
    if key not in self:
        self.__map[key] = link = Link()
        root = self.__root
        last = root.prev
        link.prev, link.next, link.key = last, root, key
        last.next = link
        # prev连接是一个proxy
        root.prev = proxy(link)
    dict_setitem(self, key, value)
def __delitem__(self, key, dict_delitem=dict.__delitem__):
    dict_delitem(self, key)
    link = self.__map.pop(key)
    link_prev = link.prev
    link_next = link.next
    link_prev.next = link_next
    link_next.prev = link_prev
    link.prev = None
    link.next = None
def __iter__(self):
    root = self.__root
    curr = root.next
    while curr is not root:
        yield curr.key
        curr = curr.next
def __reversed__(self):
    root = self.__root
    curr = root.prev
    while curr is not root:
        yield curr.key
        curr = curr.prev
def clear(self):
    root = self.__root
    root.prev = root.next = root
    self.__map.clear()
    dict.clear(self)

其它方法就不一一说明了，但是从中可以看出，涉及到元素顺序的操作，双向循环链表的应用还是很广泛的。