显然,在Python中,快速搜索可以得到数百万种memoization装饰器的实现和风味。然而,我对一种我无法找到的风味感兴趣。我希望它能够使存储值的缓存具有固定容量。当添加新元素时,如果达到容量,则删除最旧的值并用最新的值替换它。
我的担忧是,如果我使用memoization来存储大量元素,那么程序将因为缺乏内存而崩溃。(我不知道这种担忧在实践中是否合理。)如果缓存的大小是固定的,那么内存错误就不是一个问题了。而且,我处理的许多问题会随着程序的执行而改变,因此最初缓存的值看起来会与后来缓存的值非常不同(并且很少再次出现)。这就是为什么我希望最老的东西被最新的东西替换。
我发现了OrderedDict类和一个示例,展示了如何对其进行子类化以指定最大大小。我想使用它作为我的缓存,而不是普通的dict。问题是,我需要memoize装饰器接受一个名为maxlen的参数,默认值为None。如果它是None,则缓存是无限的,并像往常一样运行。任何其他值都用作缓存的大小。
我希望它能像以下这样工作:
以下是我目前的代码,但我不知道如何将参数传递到装饰器中,同时使其在“裸”和带有参数的情况下都能正常工作。
我的担忧是,如果我使用memoization来存储大量元素,那么程序将因为缺乏内存而崩溃。(我不知道这种担忧在实践中是否合理。)如果缓存的大小是固定的,那么内存错误就不是一个问题了。而且,我处理的许多问题会随着程序的执行而改变,因此最初缓存的值看起来会与后来缓存的值非常不同(并且很少再次出现)。这就是为什么我希望最老的东西被最新的东西替换。
我发现了OrderedDict类和一个示例,展示了如何对其进行子类化以指定最大大小。我想使用它作为我的缓存,而不是普通的dict。问题是,我需要memoize装饰器接受一个名为maxlen的参数,默认值为None。如果它是None,则缓存是无限的,并像往常一样运行。任何其他值都用作缓存的大小。
我希望它能像以下这样工作:
@memoize
def some_function(spam, eggs):
# This would use the boundless cache.
pass
并且
@memoize(200) # or @memoize(maxlen=200)
def some_function(spam, eggs):
# This would use the bounded cache of size 200.
pass
以下是我目前的代码,但我不知道如何将参数传递到装饰器中,同时使其在“裸”和带有参数的情况下都能正常工作。
import collections
import functools
class BoundedOrderedDict(collections.OrderedDict):
def __init__(self, *args, **kwds):
self.maxlen = kwds.pop("maxlen", None)
collections.OrderedDict.__init__(self, *args, **kwds)
self._checklen()
def __setitem__(self, key, value):
collections.OrderedDict.__setitem__(self, key, value)
self._checklen()
def _checklen(self):
if self.maxlen is not None:
while len(self) > self.maxlen:
self.popitem(last=False)
def memoize(function):
cache = BoundedOrderedDict() # I want this to take maxlen as an argument
@functools.wraps(function)
def memo_target(*args):
lookup_value = args
if lookup_value not in cache:
cache[lookup_value] = function(*args)
return cache[lookup_value]
return memo_target
@memoize
def fib(n):
if n < 2: return 1
return fib(n-1) + fib(n-2)
if __name__ == '__main__':
x = fib(50)
print(x)
编辑:根据Ben的建议,我创建了以下装饰器,我相信它按照我想象的方式工作。对于我来说,能够使用这些被修饰过的函数与multiprocessing
很重要,在过去这曾经是个问题。但是,对这段代码进行快速测试时,即使将工作分配给一组线程池,它似乎也能正确地工作。
def memoize(func=None, maxlen=None):
if func:
cache = BoundedOrderedDict(maxlen=maxlen)
@functools.wraps(func)
def memo_target(*args):
lookup_value = args
if lookup_value not in cache:
cache[lookup_value] = func(*args)
return cache[lookup_value]
return memo_target
else:
def memoize_factory(func):
return memoize(func, maxlen=maxlen)
return memoize_factory