Lru_cache（来自 functools）如何工作？

Question

特别是 在使用递归代码时 lru_cache 有很大的改进。我确实知道缓存是一个 space，它存储必须快速提供的数据并避免计算机重新计算。

functools 中的 Python lru_cache 如何在内部工作？

我正在寻找一个具体的答案，它是否像 Python 的其余部分一样使用词典？它只存储 return 值吗？

我知道 Python 很大程度上建立在字典之上，但是，我找不到这个问题的具体答案。希望有人可以为 Whosebug.

上的所有用户简化此答案

Answer 1

functools 源代码可在此处获得：https://github.com/python/cpython/blob/master/Lib/functools.py

lru_cache 使用 _lru_cache_wrapper 装饰器（python 带有参数模式的装饰器）它有一个 cache 字典 in context它在其中保存调用函数的 return 值（每个装饰函数都有自己的缓存字典）。字典键是使用参数中的 _make_key 函数生成的。在下面添加了一些大胆的评论：

# ACCORDING TO PASSED maxsize ARGUMENT _lru_cache_wrapper
# DEFINES AND RETURNS ONE OF wrapper DECORATORS

def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo):
    # Constants shared by all lru cache instances:
    sentinel = object()      # unique object used to signal cache misses

    cache = {}                                # RESULTS SAVES HERE
    cache_get = cache.get    # bound method to lookup a key or return None

    # ... maxsize is None:

    def wrapper(*args, **kwds):
        # Simple caching without ordering or size limit
        nonlocal hits, misses
        key = make_key(args, kwds, typed)     # BUILD A KEY FROM ARGUMENTS
        result = cache_get(key, sentinel)     # TRYING TO GET PREVIOUS CALLS RESULT
        if result is not sentinel:            # ALREADY CALLED WITH PASSED ARGS
            hits += 1
            return result                     # RETURN SAVED RESULT
                                              # WITHOUT ACTUALLY CALLING FUNCTION
        misses += 1
        result = user_function(*args, **kwds) # FUNCTION CALL - if cache[key] empty
        cache[key] = result                   # SAVE RESULT

        return result
    # ...

    return wrapper

Answer 2

你可以查看源代码here。

本质上它使用两个数据结构，一个字典映射函数参数到它的结果，和一个链表来跟踪你的函数调用历史。

缓存基本上是使用以下内容实现的，这是不言自明的。

cache = {}
cache_get = cache.get
....
make_key = _make_key         # build a key from the function arguments
key = make_key(args, kwds, typed)
result = cache_get(key, sentinel)

更新链表的要点是：

elif full:

    oldroot = root
    oldroot[KEY] = key
    oldroot[RESULT] = result

    # update the linked list to pop out the least recent function call information        
    root = oldroot[NEXT]
    oldkey = root[KEY]
    oldresult = root[RESULT]
    root[KEY] = root[RESULT] = None
    ......

Answer 3

Python 3.9 LRU缓存源码：https://github.com/python/cpython/blob/3.9/Lib/functools.py#L429

示例 Fib 代码

@lru_cache(maxsize=2)
def fib(n):
    if n == 0:
        return 0
    if n == 1:
        return 1
    return fib(n - 1) + fib(n - 2)

LRU 缓存装饰器检查一些基本情况，然后用包装器 _lru_cache_wrapper 包装用户函数。在包装器内部，将项目添加到缓存的逻辑，LRU 逻辑，即向循环队列添加新项目，从循环队列中删除项目。

def lru_cache(maxsize=128, typed=False):
...
    if isinstance(maxsize, int):
        # Negative maxsize is treated as 0
        if maxsize < 0:
            maxsize = 0
    elif callable(maxsize) and isinstance(typed, bool):
        # The user_function was passed in directly via the maxsize argument
        user_function, maxsize = maxsize, 128
        wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)
        wrapper.cache_parameters = lambda : {'maxsize': maxsize, 'typed': typed}
        return update_wrapper(wrapper, user_function)
    elif maxsize is not None:
        raise TypeError(
         'Expected first argument to be an integer, a callable, or None')

    def decorating_function(user_function):
        wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)
        wrapper.cache_parameters = lambda : {'maxsize': maxsize, 'typed': typed}
        return update_wrapper(wrapper, user_function)

    return decorating_function

lru_cache 标准化 maxsize(when negative)，添加 CacheInfo 细节，最后添加包装器并更新装饰器文档和其他细节。

lru_cache_wrapper

Lru Cache wrapper 的簿记变量很少。

 sentinel = object()          # unique object used to signal cache misses
 make_key = _make_key         # build a key from the function arguments
 PREV, NEXT, KEY, RESULT = 0, 1, 2, 3   # names for the link fields

 cache = {}
 hits = misses = 0
 full = False
 cache_get = cache.get    # bound method to lookup a key or return None
 cache_len = cache.__len__  # get cache size without calling len()
 lock = RLock()           # because linkedlist updates aren't threadsafe
 root = []                # root of the circular doubly linked list
 root[:] = [root, root, None, None]     # initialize by pointing to self

包装器在执行任何操作之前获取锁。
几个重要变量 - 根列表包含所有符合 maxsize 值的项目。记住 root 的重要概念是在前一个 (0) 和下一个位置 (1)
中自引用自身 (root[:] = [root, root, None, None])

三个高级别检查

第一种情况，当maxsize为0时，表示没有缓存功能，wrapper将用户函数包装起来，没有任何缓存能力。包装器增加缓存未命中计数和 return 结果。
```
 def wrapper(*args, **kwds):
     # No caching -- just a statistics update
     nonlocal misses
     misses += 1
     result = user_function(*args, **kwds)
     return result
```

第二种情况。当 maxsize 为 None 时。在节中，缓存中存储的元素数量没有限制。所以包装器检查缓存（字典）中的键。当密钥存在时，包装器 returns 值并更新缓存命中信息。当密钥丢失时，包装器使用用户传递的参数调用用户函数，更新缓存，更新缓存未命中信息，并 returns 结果。

 def wrapper(*args, **kwds):
     # Simple caching without ordering or size limit
     nonlocal hits, misses
     key = make_key(args, kwds, typed)
     result = cache_get(key, sentinel)
     if result is not sentinel:
         hits += 1
         return result
     misses += 1
     result = user_function(*args, **kwds)
     cache[key] = result
     return result

第三种情况，当maxsize为默认值（128）或用户传递的整数值。这是实际的 LRU 缓存实现。包装器中的整个代码以线程安全的方式。在执行任何操作之前，read/write/delete 从缓存中，the wrapper obtains RLock.

LRU 缓存

缓存中的值存储为四项列表（记住根）。第一项是对前一项的引用，第二项是对下一项的引用，第三项是特定函数调用的键，第四项是结果。这是 Fibonacci 函数参数 1 [[[...], [...], 1, 1], [[...], [...], 1, 1], None, None] 的实际值。 [...] 表示对 self(list) 的引用。
第一个检查是缓存命中。如果是，则缓存中的值是一个包含四个值的列表。
```
 nonlocal root, hits, misses, full
 key = make_key(args, kwds, typed)
 with lock:
     link = cache_get(key)
      if link is not None:
          # Move the link to the front of the circular queue
          print(f'Cache hit for {key}, {root}')
          link_prev, link_next, _key, result = link
          link_prev[NEXT] = link_next
          link_next[PREV] = link_prev
          last = root[PREV]
          last[NEXT] = root[PREV] = link
          link[PREV] = last
          link[NEXT] = root
          hits += 1
          return result
```
当项目已经在缓存中时，不需要检查循环队列是否已满或从缓存中弹出项目。而是更改循环队列中项目的位置。由于最近使用的项目总是在顶部，代码将最近的值移动到队列的顶部，并且前一个顶部项目成为当前项目的下一个 last[NEXT] = root[PREV] = link 和 link[PREV] = last 和 link[NEXT] = root. NEXT 和 PREV 在顶部初始化，指向列表 PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 # names for the link fields 中的适当位置。最后，增加缓存命中信息和 return 结果。
当缓存未命中时，更新未命中信息，代码检查三种情况。这三个操作都发生在获得 RLock 之后。源码中的三种情况按如下顺序——获取锁key后在缓存中找到，缓存已满，缓存可以取新项。为了演示，我们按照顺序来，当缓存未满时，缓存满，获取锁后key在缓存中可用。

缓存未满时

    ...
    else:
        # Put result in a new link at the front of the queue.
        last = root[PREV]
        link = [last, root, key, result]
        last[NEXT] = root[PREV] = cache[key] = link
        # Use the cache_len bound method instead of the len() function
        # which could potentially be wrapped in an lru_cache itself.
        full = (cache_len() >= maxsize)

当缓存未满时，准备最近的result(link = [last, root, key, result])以包含根的先前引用、根、键和计算结果。
然后将最近的结果(link)指向循环队列的顶部(root[PREV] = link)，根的上一项的next指向最近的结果(last[NEXT]=link), 并将最近的结果添加到缓存中(cache[key] = link).
最后，检查缓存是否已满(cache_len() >= maxsize and cache_len = cache.__len__ is declared in the top)并将状态设置为已满
以fib为例，当函数接收到第一个值1时，root为空，root值为[[...], [...], None, None]，将结果加入循环队列后，根值为 [[[...], [...], 1, 1], [[...], [...], 1, 1], None, None]。 previous 和 next 都指向键 1 的结果。对于下一个值 0，插入后根值为

[[[[...], [...], 1, 1], [...], 0, 0], [[...], [[...], [...], 0, 0], 1, 1], None, None]。上一个是[[[[...], [...], None, None], [...], 1, 1], [[...], [[...], [...], 1, 1], None, None], 0, 0]，下一个是[[[[...], [...], 0, 0], [...], None, None], [[...], [[...], [...], None, None], 0, 0], 1, 1]

缓存满时

    ...
    elif full:
        # Use the old root to store the new key and result.
        oldroot = root
        oldroot[KEY] = key
        oldroot[RESULT] = result
        # 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.
        root = oldroot[NEXT]
        oldkey = root[KEY]
        oldresult = root[RESULT]
        root[KEY] = root[RESULT] = None
        # Now update the cache dictionary.
        del cache[oldkey]
        # Save the potentially reentrant cache[key] assignment
        # for last, after the root and links have been put in
        # a consistent state.
        cache[key] = oldroot

当缓存已满时，使用root作为oldroot(oldroot=root)更新key和result
然后将oldroot的下一项作为新的根(root=oldroot[NEXT])，复制新的根密钥和结果(oldkey = root[KEY] and oldresult = root[RESULT])。
设置新的根键和结果为None(root[KEY] = root[RESULT] = None).
从缓存中删除旧密钥的项(del cache[oldkey])并将计算结果添加到缓存中(cache[key] = oldroot)。
对于斐波那契示例，当缓存已满且键为2时，根值为[[[[...], [...], 1, 1], [...], 0, 0], [[...], [[...], [...], 0, 0], 1, 1], None, None]，块末尾的新根为[[[[...], [...], 0, 0], [...], 2, 1], [[...], [[...], [...], 2, 1], 0, 0], None, None] 。如您所见，键 1 被删除并替换为键 2.

当key在获取锁后出现在缓存中时。

    if key in 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

当键出现在缓存中时，在获取锁后，另一个线程可能已将值入队。所以没什么可做的，包装器 return 就是结果。

最后，对结果进行编码 return。在执行缓存未命中部分之前，代码更新缓存未命中信息并调用make_key函数。

注意：我无法使嵌套列表缩进正常工作，因此答案在格式方面可能看起来有点少。

Lru_cache（来自 functools）如何工作？

How does Lru_cache (from functools) Work?

python

caching

numpy

lru

python-3.x