Python functools lru_cache with instance methods: 释放对象
Python functools lru_cache with instance methods: 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 (a 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_caches 的推荐方法是什么？
我知道有两个解决方法：
Use per instance caches or （虽然这可能会导致错误的结果）
这不是最干净的解决方案，但对程序员来说是完全透明的：
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。
pip install methodtools 安装 https://pypi.org/project/methodtools/
然后您的代码只需将 functools 替换为 methodtools 即可工作。
from methodtools import lru_cache
class Foo:
@lru_cache(maxsize=16)
def cached_method(self, x):
return x + 5
当然gc测试也是returns0。
python 3.8 在 functools 模块中引入了 cached_property 装饰器。
测试时它似乎不保留实例。
如果您不想更新到 python 3.8，您可以使用 source code。
您只需要导入 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
何时使用它
由于 weakrefs 增加了一些开销，您只希望在实例很大并且应用程序不能等待较旧的未使用调用从缓存中老化时使用它。
为什么这样更好
与其他答案不同，我们只有一个 class 缓存，而不是每个实例一个。如果您想从最近最少使用的算法中获益，这一点很重要。每个方法使用一个缓存，您可以设置 maxsize 以便无论活动实例的数量如何，总内存使用量都是有限的。
处理可变属性
如果方法中使用的任何属性是可变的，请务必添加 _eq_() 和 _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)
这个问题的一个更简单的解决方案是在构造函数中而不是在 class 定义中声明缓存：
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。
这种方法的另一个好处是，即使您的 classes 是可变的，缓存也是正确的。并且缓存键更明确，因为相关数据在缓存函数的签名中。
为了使示例更加真实，我们假设 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()
现在将实施移到 class 之外：
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()
解决方案
下面是 drop-in 的小 drop-in 替换（和包装器）lru_cache，它将 LRU 缓存放在实例（对象）上而不是 class.
总结
替换结合了 lru_cache 和 cached_property。它使用 cached_property 在第一次访问时将缓存的方法存储在实例上；这样 lru_cache 跟随对象，作为奖励，它可以用于不可散列的对象，如 non-frozen dataclass.
如何使用
使用 @instance_lru_cache 而不是 @lru_cache 来装饰方法，一切就绪。支持装饰器参数，例如@instance_lru_cache(maxsize=None)
与其他答案的比较
结果与 and , but with a simple decorator syntax. Compared to the answer provided by 提供的答案相当，此装饰器是类型提示的，不需要 third-party 库（结果相当）。
这个答案与 提供的答案不同，因为缓存现在存储在实例上（这意味着最大大小是按实例定义的，而不是按 class 定义的）并且它适用于方法无法散列的对象。
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 方法，它是一个 class 方法，除了 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
如何在 类 中使用 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 (a 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_caches 的推荐方法是什么？
我知道有两个解决方法：
Use per instance caches or
这不是最干净的解决方案，但对程序员来说是完全透明的：
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。
pip install methodtools 安装 https://pypi.org/project/methodtools/
然后您的代码只需将 functools 替换为 methodtools 即可工作。
from methodtools import lru_cache
class Foo:
@lru_cache(maxsize=16)
def cached_method(self, x):
return x + 5
当然gc测试也是returns0。
python 3.8 在 functools 模块中引入了 cached_property 装饰器。
测试时它似乎不保留实例。
如果您不想更新到 python 3.8，您可以使用 source code。
您只需要导入 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
何时使用它
由于 weakrefs 增加了一些开销，您只希望在实例很大并且应用程序不能等待较旧的未使用调用从缓存中老化时使用它。
为什么这样更好
与其他答案不同，我们只有一个 class 缓存，而不是每个实例一个。如果您想从最近最少使用的算法中获益，这一点很重要。每个方法使用一个缓存，您可以设置 maxsize 以便无论活动实例的数量如何，总内存使用量都是有限的。
处理可变属性
如果方法中使用的任何属性是可变的，请务必添加 _eq_() 和 _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)
这个问题的一个更简单的解决方案是在构造函数中而不是在 class 定义中声明缓存：
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。
这种方法的另一个好处是，即使您的 classes 是可变的，缓存也是正确的。并且缓存键更明确，因为相关数据在缓存函数的签名中。
为了使示例更加真实，我们假设 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()
现在将实施移到 class 之外：
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()
解决方案
下面是 drop-in 的小 drop-in 替换（和包装器）lru_cache，它将 LRU 缓存放在实例（对象）上而不是 class.
总结
替换结合了 lru_cache 和 cached_property。它使用 cached_property 在第一次访问时将缓存的方法存储在实例上；这样 lru_cache 跟随对象，作为奖励，它可以用于不可散列的对象，如 non-frozen dataclass.
如何使用
使用 @instance_lru_cache 而不是 @lru_cache 来装饰方法，一切就绪。支持装饰器参数，例如@instance_lru_cache(maxsize=None)
与其他答案的比较
结果与
这个答案与
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 方法，它是一个 class 方法，除了 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