Flexicacheの探求

ジェレミーからのメッセージ：伝説的なダニエル・ロイ・グリーンフェルドが、私が fastcore に最近追加した flexicache という非常に新しい機能に時間を割いて詳しく調査してくれたことを嬉しく思います。これは今では私が常に使用している非常に便利な小さなツールです。ダンと私が気に入っているのと同じくらい、あなたも気に入ってくれることを願っています！

Python でコーディングする際、関数やメソッドの結果をメモリーにキャッシュしたり、時には memcached のような一時的なストレージにキャッシュしたりするためにデコレーターをよく使用します。実際、私は複数のキャッシュデコレーターを開発・作成しており、その中には Python 3.8 の @cached_property デコーターの実装に影響を与えたものも含まれています。

flexicache という名前のキャッシュデコーターは fastcore ライブラリの一部です。flexicache を使用すると、関数やメソッドの結果を柔軟にメモリー内にキャッシュすることができます。LRU キャッシング（Least Recently Used caching）の実装を持つだけでなく、デコーターの各使用において、1 つ以上のキャッシュ無効化ポリシーを使用するように構成できます。

time_policy と mtime_policy という 2 つのポリシーが、それぞれ時間とファイルの変更時刻に基づいてキャッシュを無効化するために使用されます。time_policy は指定された秒数後にキャッシュを無効化し、mtime_policy は最後にキャッシュされた時点以降にファイルが変更されている場合にキャッシュを無効化します。

試してみましょう！

基本的な使用方法

必要なライブラリをインポートする

from fastcore.xtras import flexicache, time_policy, mtime_policy

キャッシュの有効性とキャッシュの無効化を検証するために使用されるライブラリ

from random import randint

from pathlib import Path

from time import sleep

ここでは、1 から 1000 の間の数を返す単純な関数を紹介し、これがキャッシュされている様子を示します。この関数はすべての例で使用します。

def random_func(v):

return randint(1, 1000)

関数がキャッシュされていないため False をアサート

assert random_func(1) != random_func(1)

時間ポリシー (time_policy)

これが、time_policy を使用して関数をキャッシュする方法です。

@flexicache(time_policy(.1))

def random_func():

return randint(1, 1000)

関数がキャッシュされているため True をアサート

assert random_func() == random_func()

random_func の呼び出し間の時間をシミュレートするために、sleep 関数を使用しましょう。

result = random_func()

関数がキャッシュされているため True

assert result == random_func()

キャッシュが期限切れになるように .2 秒間待機

sleep(0.2)

キャッシュが期限切れになり、関数が再度呼び出されるため False をアサート

assert result != random_func()

ファイル更新時刻ポリシー (mtime_policy)

mtime_policy で試してみましょう。ファイルにタッチすることでキャッシュが無効化されるかを確認します。ここでは、このサイトの main.py ファイルをタッチする対象ファイルとして使用します。

@flexicache(mtime_policy('../../main.py'))

def random_func():

return randint(1, 1000)

関数がキャッシュされているため True をアサート

assert random_func() == random_func()

次に、Path.touch() メソッドを使用してファイルにタッチします。これにより、ファイルの更新時刻が現在の時刻に更新され、キャッシュが無効化されるはずです。

キャッシュ結果を呼び出す関数を実行

result = random_func()

assert result == random_func() # True、関数がキャッシュされているため

ファイルの更新時刻を変更し、キャッシュを無効化

Path('../../main.py').touch()

キャッシュが無効化されたため False をアサート

assert result != random_func()

複数のポリシーの使用

flexicache のユニークな機能は、同時に複数のポリシーを使用できる点です。これにより、異なるキャッシュ戦略の利点を組み合わせることが可能になります。この例では、time_policy と mtime_policy の両方を併用します。つまり、時間制限に達した場合またはファイルが変更された場合のいずれかでキャッシュが無効化されます。

両方のポリシーでキャッシュをテストする方法は、以前の例と同じです。まず time ポリシーで関数を呼び出し、次に mtime ポリシーで呼び出し、最後に両方のポリシーを組み合わせて呼び出します。また、キャッシュが無効化されるかどうかを確認するためにファイルにも触れます。

@flexicache(time_policy(.1), mtime_policy('../../main.py'))

def random_func():

return randint(1, 1000)

True、関数がキャッシュされているため

assert random_func() == random_func()

時間による無効化のテストは以前と同じです。関数を呼び出し、時間制限に達するまで待ち、その後再度呼び出してキャッシュが無効化されるかどうかを確認します。

result = random_func()

True as the function is cached

assert result == random_func()

Sleep for .2 seconds to allow cache to expire

sleep(0.2)

False as the cache has expired and the function is called again

assert result != random_func()

テストファイルのタイムスタンプが以前と同じであることを確認します。関数を呼び出してファイルをタッチし、再度呼び出すことでキャッシュが無効化されるかを確認します。

結果をキャッシュするために関数を呼び出す

result = random_func()

True as the function is cached

assert result == random_func()

ファイルの更新時刻を更新してキャッシュを無効化する

Path('../../main.py').touch()

キャッシュが無効化されたことを確認（False を期待）

assert result != random_func()

LRU キャッシングはどうでしょうか？

次に、flexicache デコレータが LRU キャッシングの代替としてどのように動作するかをテストしてみましょう。参考までに、LRU キャッシング (Least Recently Used caching) は、最も最近使用されたアイテムを追跡し、キャッシュが最大サイズに達したときに最も古く使用されたアイテムを削除するキャッシング戦略です。つまり、スペースがなくなると、最初に最新アイテムをキャッシュから取り出します。これは FIFO (First In, First Out) 戦略を使用して、キャッシュから最も古いアイテムを削除します。

flexicache を maxsize（キャッシュの最大サイズ）2 で使用します。つまり、2 回保存した後、最も古いキャッシュエントリが破棄され始めます。キャッシュ関数のエントリは引数 (v) によって識別されるため、関数に引数を追加します。

@flexicache(maxsize=2)

def random_func(v):

return randint(1, 1000)

仕組みを見てみましょう。

result1 = random_func(1)

関数がキャッシュされているため True

assert result1 == random_func(1)

関数がキャッシュされているため True

assert random_func(2) == random_func(2)

So far so good. The cache is working as expected. Now let's start evicting the first items added to the cache. We'll add a third item to the cache and see if the first one is evicted.

3 の関数がキャッシュされているため True、

ただし random_func2(1) の結果は除外されます

assert random_func(3) == random_func(3)

最初の結果がもはやキャッシュされていないため False

assert result1 != random_func(1)

timed_cache convenience wrapper

lru_cache は Python に組み込まれたデコレータで、関数の結果をキャッシュするためのシンプルな方法を提供します。これは Least Recently Used (LRU) キャッシング戦略を使用しており、引数に基づいて最も最近使用された項目を追跡し、キャッシュが最大サイズに達したときに最も最近使用されていない項目を削除します。つまり、スペースがなくなると、最新の項目からキャッシュから取り出します。

欠点はタイムアウト機能がないことです。特定の時間だけ結果をキャッシュしたい場合は、それを自分で実装する必要があります。

fastcore.xtras.timed_cache は、functools.lru_cache にタイムアウト機能を追加した flexicache の実装です。

from fastcore.xtras import timed_cache

@flexicache(time_policy(.1), maxsize=2) のショートカット

@timed_cache(.1, maxsize=2)

def random_func(v):

return randint(1, 1000)

関数がキャッシュされている場合も真となる

assert random_func(1) == random_func(1)

time_policy(.1)、maxsize=2 を指定した flexicache で、以前と同じようにタイムアウトのテストを行います。関数を呼び出し、タイムアウトに達するまで待機し、その後再度呼び出してキャッシュが無効化されたかを確認します。

キャッシュが期限切れになるのに十分な時間待つ

sleep(0.2)

キャッシュが時間無効化されるため偽となる

assert result1 != random_func(1)

最後に、LRU（Least Recently Used）キャッシュが最初にキャッシュされたアイテムを削除していることを確認します。これは上記の LRU キャッシングに関するセクションで用いたのと同じ LRU キャッシュセットのテストです。再びキャッシュに第 3 のアイテムを追加し、最初のアイテムが除外されるかを確認します。

result1 = random_func(1)

関数がキャッシュされているため真となる

assert result1 == random_func(1)

関数がキャッシュされているため真となる

assert random_func(2) == random_func(2)

3 のための関数がキャッシュされるが、random_func2(1) の結果を除外する

assert random_func(3) == random_func(3)

最初の結果はもはやキャッシュされていないため偽となる

assert result1 != random_func(1)

images/exploring-flexicache.png