ロックフリーの並行LRUキャッシュを実装する

```python import threading import time from collections import OrderedDict class Node: def __init__(self, key, value): self.key = key self.value = value self.prev = None self.next = None class LRUCache: def __init__(self, capacity: int): if capacity <= 0: raise ValueError("容量は正の整数である必要があります。") self.capacity = capacity self.cache = {} self.head = Node(None, None) # ダミーヘッド self.tail = Node(None, None) # ダミーテール self.head.next = self.tail self.tail.prev = self.head self.lock = threading.RLock() # この例では簡潔さのためにRLockを使用していますが、よりきめ細かなアプローチも可能です。 def _remove_node(self, node): prev_node = node.prev next_node = node.next prev_node.next = next_node next_node.prev = prev_node def _add_to_front(self, node): node.next = self.head.next node.prev = self.head self.head.next.prev = node self.head.next = node def get(self, key): with self.lock: if key in self.cache: node = self.cache[key] self._remove_node(node) self._add_to_front(node) return node.value return None def put(self, key, value): with self.lock: if key in self.cache: node = self.cache[key] node.value = value self._remove_node(node) self._add_to_front(node) else: if len(self.cache) >= self.capacity: # LRUアイテムを削除 lru_node = self.tail.prev self._remove_node(lru_node) del self.cache[lru_node.key] new_node = Node(key, value) self.cache[key] = new_node self._add_to_front(new_node) def delete(self, key): with self.lock: if key in self.cache: node = self.cache[key] self._remove_node(node) del self.cache[key] return True return False def __len__(self): with self.lock: return len(self.cache) def __str__(self): with self.lock: items = [] current = self.head.next while current != self.tail: items.append(f"({current.key}: {current.value})") current = current.next return " -> ".join(items) # --- 並行処理戦略の説明 --- # この実装では、すべての操作（get、put、delete）に単一のthreading.RLockを使用しています。 # これによりスレッドセーフティは保証されますが、プロンプトで要求されているように競合を最小限に抑えることはできません。 # より高度で、きめ細かなアプローチとしては、キャッシュをセグメント化したり、 # ロックフリーデータ構造を使用したりすることが考えられます。例えば、キャッシュエントリのサブセットを保護する複数のロック（ストライピングロック）を使用したり、 # 読み取り/書き込みロックを使用して複数のリーダーがキャッシュの異なる部分に同時にアクセスできるようにしたりすることができます。 # 現在のRLockは、一度に1つのスレッドのみがキャッシュの内部状態（辞書と双方向連結リスト）を変更できることを保証します。 # これにより、複数のスレッドが同じノードを更新しようとしたり、ノードを追加/削除したり、エントリを同時に削除したりして、データ破損やLRU順序の間違いにつながる競合状態を防ぎます。 # RLock（再入可能ロック）の使用は、スレッドがロックを複数回取得できるため、複雑なシナリオで役立つことがありますが、この特定の用途では標準的なLockと比較して競合特性を根本的に変更するものではありません。 # --- デモンストレーション --- def worker(cache, operations, thread_id): for op, key, value in operations: if op == 'get': cache.get(key) elif op == 'put': cache.put(key, value) elif op == 'delete': cache.delete(key) # いくらかの作業をシミュレートし、他のスレッドが実行できるようにする time.sleep(0.001) def run_demonstration(): print("LRUキャッシュデモンストレーションを実行中...") capacity = 5 cache = LRUCache(capacity) num_threads = 10 ops_per_thread = 50 all_operations = [] for i in range(num_threads): thread_ops = [] for j in range(ops_per_thread): op_type = ['get', 'put', 'delete'][j % 3] # 操作タイプをローテーション key = (i * ops_per_thread + j) % (capacity * 2) # キャッシュにあるかもしれないし、ないかもしれないキー value = f"val_{i}_{j}" thread_ops.append((op_type, key, value)) all_operations.append(thread_ops) threads = [] for i in range(num_threads): thread = threading.Thread(target=worker, args=(cache, all_operations[i], i)) threads.append(thread) thread.start() for thread in threads: thread.join() print(f"デモンストレーション完了。最終キャッシュサイズ: {len(cache)}") # キャッシュが容量を超えないことをアサートする assert len(cache) <= capacity, f"キャッシュ容量を超過しました！現在のサイズ: {len(cache)}, 容量: {capacity}" print("アサート成功：キャッシュ容量は維持されています。") # 基本的な整合性チェック print("基本的な整合性チェックを実行中...") # これは非常に基本的なチェックです。より堅牢なテストでは、期待される状態を追跡する必要があります。 # とりあえず、サイズが負にならない、アクセス中に例外が発生しないことを確認します。 try: # 投入された可能性が高いキーをいくつかアクセスしてみる for i in range(min(5, capacity)): key_to_check = (i * ops_per_thread) % (capacity * 2) cache.get(key_to_check) print("基本的な整合性チェックに成功しました。") except Exception as e: print(f"整合性チェックが例外で失敗しました: {e}") assert False, "キャッシュ整合性チェックに失敗しました。" if __name__ == "__main__": run_demonstration() ```

"""競合を最小化した並行 LRU キャッシュ（厳密な LRU）。 並行性戦略（概要）: - キャッシュは、key->Node のために dict を使用し、グローバルな LRU 順序のために双方向リンクリストを使用します。 - キーと値の操作における競合を減らすために、キー空間は複数のロックにまたがってストライプ化されています。 特定のキーに対する更新（マッピングおよびノードフィールドの挿入/更新/削除/読み取り）は、 そのキーのストライプロックのみを使用します。 - *厳密なグローバル LRU セマンティクス* を保つために、LRU リストのすべての変更は 専用の lru_lock によって直列化されます。これにより、リストを破損させうる競合を回避し、 「touch」操作に単一の全順序が存在することを保証します。 - エビクションには協調が必要です: lru_lock の下で現在の末尾（最も最近使われていないもの）を選択し、 次にそのキーのストライプロックを取得して、それが依然として同じノードであることを確認してから マップから削除します。これにより、並行更新されたキーを誤ってエビクトすることを防ぎます。 これが正しい理由: - LRU リストは lru_lock の下でのみ変更されるため、破損することがなく、各 touch （既存キーに対する get/put）は完全に順序付けられ、厳密な LRU が実現されます。 - マップエントリとノードの値はそのキー用のストライプロックによって保護されるため、同じキーに対する 並行した更新/削除によって状態が破損することを防ぎます。 - エビクションは、lru_lock の下でエビクションが決定された瞬間の実際の末尾に基づきます。候補ノードが 並行した touch/delete によって変化した場合、エビクションは再試行されます。 注: Python は複雑な共有構造に対する真のロックフリープリミティブを提供していません。この 実装では、関心事を分離することにより、*すべて* を単一のグローバルロックで囲むことを避けています: - キーレベルの操作はストライプ化されています。 - リスト順序の維持のみに単一のロックを使用します。 デモは __main__ に含まれています。 """ from __future__ import annotations import threading import time import random from dataclasses import dataclass from typing import Any, Optional, Dict, List @dataclass class _Node: key: Any value: Any prev: Optional["_Node"] = None next: Optional["_Node"] = None alive: bool = True # 削除/エビクト時に False に設定される class ConcurrentLRUCache: def __init__(self, capacity: int, stripes: int = 64): if capacity <= 0: raise ValueError("capacity は正の値でなければなりません") if stripes <= 0: raise ValueError("stripes は正の値でなければなりません") self.capacity = int(capacity) self._map: Dict[Any, _Node] = {} # 双方向リンクリストの番兵 head/tail（MRU は head.next、LRU は tail.prev） self._head = _Node("<HEAD>", None) self._tail = _Node("<TAIL>", None) self._head.next = self._tail self._tail.prev = self._head # リスト操作とサイズ/エビクション判定を直列化する self._lru_lock = threading.Lock() # キーレベルの map/node 操作用ストライプロック self._stripes_n = int(stripes) self._stripes: List[threading.Lock] = [threading.Lock() for _ in range(self._stripes_n)] # サイズは lru_lock の下で追跡する self._size = 0 def _stripe_lock(self, key: Any) -> threading.Lock: return self._stripes[hash(key) % self._stripes_n] # ----- 双方向リンクリストのヘルパー（self._lru_lock を保持していることが必要） ----- def _unlink(self, node: _Node) -> None: p = node.prev n = node.next if p is not None: p.next = n if n is not None: n.prev = p node.prev = None node.next = None def _link_front(self, node: _Node) -> None: first = self._head.next node.prev = self._head node.next = first self._head.next = node if first is not None: first.prev = node def _move_to_front(self, node: _Node) -> None: # node はリスト内に存在していなければならない self._unlink(node) self._link_front(node) def _current_lru_node(self) -> Optional[_Node]: # 実際の LRU ノード（tail.prev）を返す。空なら None n = self._tail.prev if n is None or n is self._head: return None return n # ----- 公開 API ----- def get(self, key: Any) -> Any: lock = self._stripe_lock(key) with lock: node = self._map.get(key) if node is None or not node.alive: return None value = node.value # 先頭への移動にはリストロックが必要。競合を減らすため、ストライプロックの外で行う。 with self._lru_lock: # 削除/置換されたノードを移動しないよう、ストライプロック下で所属を再確認する。 with lock: cur = self._map.get(key) if cur is not node or cur is None or not cur.alive: return value # 読み取り時点では value は有効だった。競合により touch は失われた。 # lru_lock の下では、このノードを安全に移動できる。 self._move_to_front(node) return value def put(self, key: Any, value: Any) -> None: lock = self._stripe_lock(key) # まず、キーが存在するかを確認し、ストライプロック下で更新する。 with lock: node = self._map.get(key) if node is not None and node.alive: node.value = value # lru_lock の下で touch（先頭へ移動）。 # ストライプロックは素早く解放する？ 短く保てる: 更新は完了済み。 pass else: node = None if node is not None: with self._lru_lock: # 依然として存在し、同じノードであることを保証する with lock: cur = self._map.get(key) if cur is node and cur is not None and cur.alive: self._move_to_front(node) return # 新しいノードを挿入 new_node = _Node(key, value) with self._lru_lock: # 一貫した挿入+リンクを保証するため、lru_lock を保持したままストライプロックを取得する。 with lock: existing = self._map.get(key) if existing is not None and existing.alive: # 誰かが並行して挿入した。その代わりにそれを更新する。 existing.value = value # touch する self._move_to_front(existing) return # マッピングを挿入してリンクする self._map[key] = new_node self._link_front(new_node) self._size += 1 # 容量超過中はエビクトする while self._size > self.capacity: if not self._evict_one_locked(): # 競合によりエビクションに失敗した場合は再試行する continue def _evict_one_locked(self) -> bool: """単一の LRU ノードをエビクトする。 self._lru_lock を保持していることが必要。 エビクションが行われた場合は True、再試行が必要な場合は False を返す。 """ victim = self._current_lru_node() if victim is None: return True vkey = victim.key vlock = self._stripe_lock(vkey) with vlock: cur = self._map.get(vkey) if cur is not victim or cur is None or not cur.alive: # victim が変わった、またはすでに削除されている。再試行。 return False # まず map から削除する del self._map[vkey] victim.alive = False # 次にリストから削除する（lru_lock は保持している） self._unlink(victim) self._size -= 1 return True def delete(self, key: Any) -> bool: lock = self._stripe_lock(key) # エビクション/挿入経路とのデッドロックを避けるため、先に lru_lock を取得する。 with self._lru_lock: with lock: node = self._map.get(key) if node is None or not node.alive: return False del self._map[key] node.alive = False # LRU リストから切り離す self._unlink(node) self._size -= 1 return True def __len__(self) -> int: with self._lru_lock: return self._size def _assert_invariants(self) -> None: """デバッグ/検査ルーチン: 内部整合性を検証する。""" with self._lru_lock: # リスト構造を確認し、ノード数を数える count = 0 seen = set() n = self._head.next prev = self._head while n is not None and n is not self._tail: assert n.prev is prev assert n.alive assert n.key not in seen seen.add(n.key) count += 1 prev = n n = n.next assert n is self._tail assert self._tail.prev is prev assert count == self._size # map とリストが一致していること（ベストエフォート。完全にするにはストライプロックが必要）。 # lru_lock の下でも、map はストライプロック下で変更されうるため、保守的な検査のみを行う: # すべてのリストノードは map に存在していなければならない。 for k in seen: lock = self._stripe_lock(k) with lock: m = self._map.get(k) assert m is not None and m.alive and m.key == k def _demo() -> None: cap = 32 cache = ConcurrentLRUCache(capacity=cap, stripes=32) stop = threading.Event() errors = [] keys = [f"k{i}" for i in range(64)] def worker(tid: int) -> None: rnd = random.Random(12345 + tid) try: while not stop.is_set(): op = rnd.random() k = rnd.choice(keys) if op < 0.50: cache.get(k) elif op < 0.85: cache.put(k, (tid, rnd.randint(0, 10_000))) else: cache.delete(k) # ときどきサイズが容量を超えていないことを確認する if rnd.random() < 0.01: assert len(cache) <= cap except Exception as e: errors.append(e) stop.set() threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(12)] for t in threads: t.start() # 少し実行する time.sleep(1.5) # 停止して join する stop.set() for t in threads: t.join(timeout=2.0) # ワーカーエラーなし if errors: raise errors[0] # 最終的な不変条件の検査 assert len(cache) <= cap cache._assert_invariants() # 基本的な健全性: 複数スレッドから同じキーに対して put/get を繰り返しても破損しないはず cache.put("sentinel", 1) assert cache.get("sentinel") == 1 assert cache.delete("sentinel") is True assert cache.get("sentinel") is None if __name__ == "__main__": _demo() print("デモ完了; キャッシュの不変条件は保たれています。")