使用实例方法的 Python function tools lru_cache: release object

如何在类内部使用 functools.lru_cache而不泄漏内存？

在下面的最小示例中，foo实例将不会被释放，尽管超出了作用域并且没有引用程序(除了 lru_cache)。

from functools import lru_cache
class BigClass:
pass
class Foo:
def __init__(self):
self.big = BigClass()
@lru_cache(maxsize=16)
def cached_method(self, x):
return x + 5


def fun():
foo = Foo()
print(foo.cached_method(10))
print(foo.cached_method(10)) # use cache
return 'something'


fun()

但是 foo和 foo.big(BigClass)仍然活着

import gc; gc.collect()  # collect garbage
len([obj for obj in gc.get_objects() if isinstance(obj, Foo)]) # is 1

这意味着 Foo/BigClass实例仍然驻留在内存中，即使删除 Foo(del Foo)也不会释放它们。

lru_cache到底为什么要保留这个实例呢？缓存不是使用一些散列而不是实际的对象吗？

在类中使用 lru_cache的推荐方法是什么？

我知道两个解决办法: 使用每个实例缓存 或 使缓存忽略对象(尽管这可能会导致错误的结果)

35478 次浏览
小开
最佳答案

这不是最干净的解决方案，但对程序员来说是完全透明的:

import functools
import weakref


def memoized_method(*lru_args, **lru_kwargs):
def decorator(func):
@functools.wraps(func)
def wrapped_func(self, *args, **kwargs):
# We're storing the wrapped method inside the instance. If we had
# a strong reference to self the instance would never die.
self_weak = weakref.ref(self)
@functools.wraps(func)
@functools.lru_cache(*lru_args, **lru_kwargs)
def cached_method(*args, **kwargs):
return func(self_weak(), *args, **kwargs)
setattr(self, func.__name__, cached_method)
return cached_method(*args, **kwargs)
return wrapped_func
return decorator

它采用与 lru_cache完全相同的参数，并且工作原理完全相同。然而，它从不将 self传递给 lru_cache，而是使用每个实例 lru_cache。

小开

我将为这个用例介绍 methodtools。

安装 https://pypi.org/project/methodtools/

然后，您的代码只需将 function 工具替换为 methodtools 即可工作。

from methodtools import lru_cache
class Foo:
@lru_cache(maxsize=16)
def cached_method(self, x):
return x + 5

当然，gc 测试也返回0。

小开

Python 3.8在 functools模块中引入了 cached_property装饰器。 当测试它似乎不保留的实例。

如果不想更新到 python 3.8，可以使用 源代码。 你所需要的只是导入 RLock并创建 _NOT_FOUND对象。意思是:

from threading import RLock


_NOT_FOUND = object()


class cached_property:
# https://github.com/python/cpython/blob/v3.8.0/Lib/functools.py#L930
...
小开

简单的包装解决方案

下面是一个包装器，它将保持对实例的弱引用:

import functools
import weakref


def weak_lru(maxsize=128, typed=False):
'LRU Cache decorator that keeps a weak reference to "self"'
def wrapper(func):


@functools.lru_cache(maxsize, typed)
def _func(_self, *args, **kwargs):
return func(_self(), *args, **kwargs)


@functools.wraps(func)
def inner(self, *args, **kwargs):
return _func(weakref.ref(self), *args, **kwargs)


return inner


return wrapper

例子

像这样使用它:

class Weather:
"Lookup weather information on a government website"


def __init__(self, station_id):
self.station_id = station_id


@weak_lru(maxsize=10)
def climate(self, category='average_temperature'):
print('Simulating a slow method call!')
return self.station_id + category

什么时候用

由于弱点增加了一些开销，因此只有当实例较大且应用程序不能等待较旧的未使用调用超出缓存时，才需要使用它。

为什么这样更好

与另一个答案不同，我们只有类的一个缓存，而不是每个实例一个缓存。如果您想从最近使用最少的算法中获得一些好处，这一点很重要。每个方法只有一个缓存，你可以设置 最大尺寸，这样总内存使用是有限的，不管数量是活的。

处理可变属性

如果方法中使用的任何属性是可变的，请确保添加 等式 _ () 和 _ < em > hash _ () 方法:

class Weather:
"Lookup weather information on a government website"


def __init__(self, station_id):
self.station_id = station_id


def update_station(station_id):
self.station_id = station_id


def __eq__(self, other):
return self.station_id == other.station_id


def __hash__(self):
return hash(self.station_id)
小开

这个问题的一个更简单的解决方案是在构造函数中而不是在类定义中声明缓存:

from functools import lru_cache
import gc


class BigClass:
pass
class Foo:
def __init__(self):
self.big = BigClass()
self.cached_method = lru_cache(maxsize=16)(self.cached_method)
def cached_method(self, x):
return x + 5


def fun():
foo = Foo()
print(foo.cached_method(10))
print(foo.cached_method(10)) # use cache
return 'something'
    

if __name__ == '__main__':
fun()
gc.collect()  # collect garbage
print(len([obj for obj in gc.get_objects() if isinstance(obj, Foo)]))  # is 0
小开

您可以将该方法的实现移动到一个模块全局函数，在从该方法调用它时只传递来自 self的相关数据，并在该函数上使用 @lru_cache。

这种方法的另一个好处是，即使类是可变的，缓存也是正确的。而且缓存键更加明确，因为只有相关数据在缓存函数的签名中。

为了让这个例子更加真实，我们假设 cached_method()需要来自 self.big的信息:

from dataclasses import dataclass
from functools import lru_cache


@dataclass
class BigClass:
base: int


class Foo:
def __init__(self):
self.big = BigClass(base=100)


@lru_cache(maxsize=16)  # the leak is here
def cached_method(self, x: int) -> int:
return self.big.base + x


def fun():
foo = Foo()
print(foo.cached_method(10))
print(foo.cached_method(10)) # use cache
return 'something'


fun()

现在将实现移到类之外:

from dataclasses import dataclass
from functools import lru_cache


@dataclass
class BigClass:
base: int


@lru_cache(maxsize=16)  # no leak from here
def _cached_method(base: int, x: int) -> int:
return base + x


class Foo:
def __init__(self):
self.big = BigClass(base=100)


def cached_method(self, x: int) -> int:
return _cached_method(self.big.base, x)


def fun():
foo = Foo()
print(foo.cached_method(10))
print(foo.cached_method(10)) # use cache
return 'something'


fun()
小开

解决方案

下面是 lru_cache的一个小插件替换(和包装) ，它将 LRU 缓存放在实例(对象)上而不放在类上。

摘要

替代品将 lru_cache和 cached_property结合在一起。它使用 cached_property在第一次访问时将缓存的方法存储在实例上; 这样 lru_cache就跟在对象后面，而且作为额外的好处，它可以用在非哈希对象上，比如非冻结的 dataclass。

怎么用

使用 @instance_lru_cache而不是 @lru_cache来装饰一个方法，这样就可以了

与其他答案比较

结果与 巴勃罗和 欢迎光临提供的答案相当，但使用了简单的装饰语法。与 你认识一个提供的答案相比，这个装饰符是类型提示的，并且不需要第三方库(结果是可比较的)。

这个答案与 Raymond Hettinger提供的答案不同，因为缓存现在存储在实例上(这意味着每个实例定义最大大小，而不是每个类) ，并且它适用于不可散列对象的方法。

from functools import cached_property, lru_cache, partial, update_wrapper
from typing import Callable, Optional, TypeVar, Union


T = TypeVar("T")


def instance_lru_cache(
method: Optional[Callable[..., T]] = None,
*,
maxsize: Optional[int] = 128,
typed: bool = False
) -> Union[Callable[..., T], Callable[[Callable[..., T]], Callable[..., T]]]:
"""Least-recently-used cache decorator for instance methods.


The cache follows the lifetime of an object (it is stored on the object,
not on the class) and can be used on unhashable objects. Wrapper around
functools.lru_cache.


If *maxsize* is set to None, the LRU features are disabled and the cache
can grow without bound.


If *typed* is True, arguments of different types will be cached separately.
For example, f(3.0) and f(3) will be treated as distinct calls with
distinct results.


Arguments to the cached method (other than 'self') must be hashable.


View the cache statistics named tuple (hits, misses, maxsize, currsize)
with f.cache_info().  Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.


"""


def decorator(wrapped: Callable[..., T]) -> Callable[..., T]:
def wrapper(self: object) -> Callable[..., T]:
return lru_cache(maxsize=maxsize, typed=typed)(
update_wrapper(partial(wrapped, self), wrapped)
)


return cached_property(wrapper)  # type: ignore


return decorator if method is None else decorator(method)
小开

在实例方法上使用@lru _ cache 或@cache 的问题是，self 被传递给缓存方法，尽管实际上并不需要。我不能告诉您为什么缓存 self 会导致这个问题，但是我可以给出一个我认为非常优雅的解决方案。

处理这个问题的首选方法是定义一个 dunder 方法，它是一个类方法，除了 self 之外，接受与实例方法相同的所有参数。我之所以选择这种方式，是因为它非常清晰、简约，而且不依赖于外部库。

from functools import lru_cache
class BigClass:
pass




class Foo:
def __init__(self):
self.big = BigClass()
    

@staticmethod
@lru_cache(maxsize=16)
def __cached_method__(x: int) -> int:
return x + 5


def cached_method(self, x: int) -> int:
return self.__cached_method__(x)




def fun():
foo = Foo()
print(foo.cached_method(10))
print(foo.cached_method(10)) # use cache
return 'something'


fun()

我已确认该项目是正确的垃圾收集:

import gc; gc.collect()  # collect garbage
len([obj for obj in gc.get_objects() if isinstance(obj, Foo)]) # is 0

提交答案