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

回答:
import threading
import random
import time
from typing import Any, Optional, List

class _Node:
"""
LRU の順序付けに使用される双方向連結リスト内のノードを表します。
メモリ効率のために slots を使用します。
"""
slots = 'key', 'value', 'prev', 'next'

def __init__(self, key: str, value: Any):
    self.key = key
    self.value = value
    self.prev = None
    self.next = None

class ReadWriteLock:
"""
カスタムの読み取り・書き込みロック実装です。このロックは複数のリーダーが
同時にロックを保持できる一方で、ライターは一度に 1 つだけしか保持できません。ライターは
すべてのリーダーが終了するのを待つ必要があり、リーダーはライターが
終了するのを待つ必要があります。これはリーダー優先の実装であり、新しいリーダーは
ライターが待機中であっても読み取りロックを取得できます。

このクラスは、よりクリーンにロックを取得・解放するためのコンテキストマネージャも提供します。
"""
def __init__(self):
    self._read_lock = threading.Lock() # _readers のカウントと条件変数を保護
    self._write_lock = threading.Lock() # 実際の書き込みロック
    self._readers = 0 # アクティブなリーダー数
    self._no_readers = threading.Condition(self._read_lock) # ライターがリーダーを待つための条件

def acquire_read(self):
    """読み取りロックを取得します。複数のリーダーが同時にこのロックを保持できます。"""
    with self._read_lock:
        # 現在ライターが書き込みロックを保持している場合は待機します。
        # ライターがアクティブな間、新しいリーダーはブロックされます。
        while self._write_lock.locked():
            self._no_readers.wait()
        self._readers += 1

def release_read(self):
    """読み取りロックを解放します。"""
    with self._read_lock:
        self._readers -= 1
        # リーダーがいなくなった場合は、待機中のライターに通知します
        if self._readers == 0:
            self._no_readers.notify_all()

def acquire_write(self):
    """書き込みロックを取得します。一度に 1 つのライターのみがこのロックを保持できます。"""
    # まずメインの書き込みロックを取得して、書き込みの意図を示します
    self._write_lock.acquire()
    with self._read_lock: # 次に read_lock を取得し、アクティブなリーダーを待ちます
        # すべてのアクティブなリーダーが終了するまで待機します
        while self._readers > 0:
            self._no_readers.wait()

def release_write(self):
    """書き込みロックを解放します。"""
    with self._read_lock: # read_lock を取得して、待機中のリーダー/ライターに通知します
        self._no_readers.notify_all() # 待機中のリーダーまたはライターに通知します
    self._write_lock.release() # メインの書き込みロックを解放します

class _ReadContextManager:
    def __init__(self, lock):
        self.lock = lock
    def __enter__(self):
        self.lock.acquire_read()
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.lock.release_read()

class _WriteContextManager:
    def __init__(self, lock):
        self.lock = lock
    def __enter__(self):
        self.lock.acquire_write()
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.lock.release_write()

def read_locked(self):
    """読み取りロックを取得するためのコンテキストマネージャを返します。"""
    return self._ReadContextManager(self)

def write_locked(self):
    """書き込みロックを取得するためのコンテキストマネージャを返します。"""
    return self._WriteContextManager(self)

class LRUCache:
"""
スレッドセーフな LRU（Least Recently Used: 最近最も使われていないものを削除）キャッシュ実装です。

このキャッシュは、同時アクセスを管理するために ReadWriteLock を使用します。この設計は
キャッシュミス時の並行読み取りと `keys()` 操作を可能にすることを目指しています。
一方で、キャッシュ状態または LRU 順序を変更する操作（`put`、`delete`、
および `get` におけるキャッシュヒット）は書き込みロックを取得し、これらの重要区間に対して
原子性と一貫性を保証します。このアプローチは、すべての操作に対して単一のグローバルミューテックスを
用いる場合よりも高い並行性を提供します。特に、キャッシュミスや `keys()` 呼び出しが頻繁な
読み取り中心のワークロードで効果的です。

`get` 操作は「アップグレード」パターンを採用しています。まず読み取りロックを試みて
キーの存在を確認します。見つかった場合、読み取りロックを解放してから書き込みロックを取得し、
LRU 順序を更新して最新の値を取得します。このパターンは、ロックのアップグレード中に発生しうる
競合状態（たとえば、他のスレッドによってキーが削除または更新されるなど）を処理するように
慎重に実装されています。
"""
def __init__(self, capacity: int):
    if capacity <= 0:
        raise ValueError("容量は正の整数でなければなりません。")
    self._capacity = capacity
    self._cache = {}  # O(1) で検索できるように key -> _Node を保持
    self._size = 0

    # 双方向連結リスト用のダミーの先頭ノードと末尾ノード。
    # 実データノードは常に head と tail の間にあるため、ノードの追加/削除における
    # 境界ケースが簡略化されます。
    self._head = _Node("dummy_head", None)
    self._tail = _Node("dummy_tail", None)
    self._head.next = self._tail
    self._tail.prev = self._head

    # 並行性制御: 単一の ReadWriteLock がキャッシュ全体の状態を保護します。
    self._rw_lock = ReadWriteLock()

def _add_node(self, node: _Node):
    """head の直後（最も最近使われた位置）にノードを追加します。
    書き込みロックが保持されていることを前提とします。"""
    node.prev = self._head
    node.next = self._head.next
    self._head.next.prev = node
    self._head.next = node

def _remove_node(self, node: _Node):
    """双方向連結リストからノードを削除します。
    書き込みロックが保持されていることを前提とします。"""
    prev_node = node.prev
    next_node = node.next
    prev_node.next = next_node
    next_node.prev = prev_node

def _move_to_front(self, node: _Node):
    """既存のノードをリスト先頭（最も最近使われた位置）に移動します。
    書き込みロックが保持されていることを前提とします。"""
    self._remove_node(node)
    self._add_node(node)

def _pop_tail(self) -> _Node:
    """最も最近使われていないノード（tail の直前のノード）を削除して返します。
    書き込みロックが保持されていることを前提とします。"""
    node = self._tail.prev
    self._remove_node(node)
    return node

def get(self, key: str) -> Optional[Any]:
    """
    キーが存在する場合は、そのキーに関連付けられた値を返し、最近使われたものとしてマークします。
    キーがキャッシュ内にない場合は None を返します。
    
    この操作はまず読み取りロックを取得してキーの存在を確認します。キーが
    見つかった場合は読み取りロックを解放し、その後書き込みロックを取得して
    LRU 順序を更新し、最新の値を取得します。この「アップグレード」パターンにより、
    `get` のミスや `keys()` 操作は並行実行できますが、`get` のヒットは他の書き込みと直列化されます。
    """
    node_found_under_read_lock = None
    with self._rw_lock.read_locked():
        node_found_under_read_lock = self._cache.get(key)
        if not node_found_under_read_lock:
            return None
    
    # キーが見つかった場合は、LRU 順序を更新して最新の値を得るために書き込みロックを取得します。
    # これは「アップグレード」パターンです。ここでは競合状態が発生する可能性があります: 別スレッドが
    # 読み取りロックの解放から書き込みロック取得までの間に、そのキーを削除または更新するかもしれません。
    with self._rw_lock.write_locked():
        # そのキーが依然として存在し、かつまったく同じノードオブジェクトを参照しているかを再確認します。
        # これは競合下での正しさのために非常に重要です。別スレッドが同じキーに対して
        # `put` を実行していた場合、`_cache.get(key)` は *新しい* ノードを返します。
        # 別スレッドが `delete` を実行していた場合は `None` を返します。
        current_node = self._cache.get(key)
        if current_node is node_found_under_read_lock: # まったく同じノードオブジェクトであることを確認
            self._move_to_front(node_found_under_read_lock)
            value = current_node.value # 移動後に最新の値を取得
            return value
        else:
            # ノードは、読み取りロックと書き込みロック取得の間に、別の 'put' 操作によって
            # 削除されたか置き換えられました。ミスとして扱います。
            return None

def put(self, key: str, value: Any) -> None:
    """
    キーと値のペアを挿入または更新します。挿入後にキャッシュが容量を超えた場合は、
    最も最近使われていない項目を追い出します。これは書き込み操作であり、書き込みロックを取得します。
    """
    with self._rw_lock.write_locked():
        node = self._cache.get(key)
        if node:
            # キーが存在する場合: 値を更新して先頭へ移動
            node.value = value
            self._move_to_front(node)
        else:
            # キーが存在しない場合: 新しいノードを追加
            new_node = _Node(key, value)
            self._cache[key] = new_node
            self._add_node(new_node)
            self._size += 1

            # 容量超過時は追い出し
            if self._size > self._capacity:
                lru_node = self._pop_tail()
                del self._cache[lru_node.key]
                self._size -= 1

def delete(self, key: str) -> bool:
    """
    キャッシュからキーを削除します。キーが存在した場合は True、そうでない場合は False を返します。
    これは書き込み操作であり、書き込みロックを取得します。
    """
    with self._rw_lock.write_locked():
        node = self._cache.get(key)
        if node:
            self._remove_node(node)
            del self._cache[key]
            self._size -= 1
            return True
        return False

def keys(self) -> List[str]:
    """
    現在キャッシュ内にあるすべてのキーのリストを、最も最近使われたものから
    最も最近使われていないものの順で返します。これは読み取り操作であり、読み取りロックを取得します。
    """
    with self._rw_lock.read_locked():
        keys_list = []
        current = self._head.next
        while current is not self._tail:
            keys_list.append(current.key)
            current = current.next
        return keys_list

def run_tests():
"""
LRUCache 実装を検証するために、シングルスレッドおよびマルチスレッドの一連のテストを実行します。
"""
print("--- LRU キャッシュテスト開始 ---")

# --- シングルスレッドテスト ---
print("\n--- シングルスレッドテスト ---")
cache = LRUCache(capacity=3)
assert cache.get("a") is None, "ST: 存在しないキーの get は None を返すべき"
assert cache.keys() == [], "ST: 空のキャッシュでは keys リストは空であるべき"

cache.put("a", 1)
assert cache.get("a") == 1, "ST: put 後に存在するキーを get"
assert cache.keys() == ["a"], "ST: 1 回 put した後の keys リスト"

cache.put("b", 2)
cache.put("c", 3)
assert cache.keys() == ["c", "b", "a"], "ST: 複数回 put した後の keys リスト"
assert cache._size == 3, "ST: キャッシュサイズは 3 であるべき"

# 容量超過による追い出しのテスト
cache.put("d", 4) # 'a' は LRU なので追い出されるはず
assert cache.get("a") is None, "ST: 追い出されたキー 'a' は None であるべき"
assert cache.get("d") == 4, "ST: 新しいキー 'd' は存在するべき"
assert cache.keys() == ["d", "c", "b"], "ST: 追い出し後の keys リスト"
assert cache._size == 3, "ST: 追い出し後もキャッシュサイズは 3 のままであるべき"

# 既存キーに対する put のテスト（値更新と先頭への移動）
cache.put("b", 22) # 'b' は先頭に移動し、値も更新されるはず
assert cache.get("b") == 22, "ST: 'b' の更新後の値"
assert cache.keys() == ["b", "d", "c"], "ST: 更新と 'b' の移動後の keys リスト"

# delete のテスト
assert cache.delete("c") is True, "ST: 存在するキー 'c' を delete"
assert cache.get("c") is None, "ST: 削除されたキー 'c' は None であるべき"
assert cache.keys() == ["b", "d"], "ST: 'c' 削除後の keys リスト"
assert cache._size == 2, "ST: delete 後のキャッシュサイズ"

assert cache.delete("z") is False, "ST: 存在しないキー 'z' を delete"
assert cache.keys() == ["b", "d"], "ST: 存在しないキー削除後も keys リストは変わらない"

# 容量 1 のテスト
cache_cap1 = LRUCache(capacity=1)
cache_cap1.put("x", 10)
assert cache_cap1.get("x") == 10, "ST: 容量 1、'x' を get"
cache_cap1.put("y", 20) # 'x' は追い出されるはず
assert cache_cap1.get("x") is None, "ST: 容量 1、'x' は追い出される"
assert cache_cap1.get("y") == 20, "ST: 容量 1、'y' を get"
assert cache_cap1.keys() == ["y"], "ST: 容量 1、keys リスト"

print("シングルスレッドテストに合格しました！")

# --- マルチスレッドテスト ---
print("\n--- マルチスレッドテスト ---")
MT_CAPACITY = 10
MT_NUM_THREADS = 8
MT_OPS_PER_THREAD = 2000
MT_KEY_RANGE = 20 # キーは 0 から 19（例: "0", "1", ..., "19"）

mt_cache = LRUCache(capacity=MT_CAPACITY)
# すべてのワーカースレッドが同時に操作を開始するようにするバリア
barrier = threading.Barrier(MT_NUM_THREADS + 1) # +1 はメインスレッド分
error_flag = threading.Event() # いずれかのスレッドでエラーが発生したらセット

# いずれかのスレッドによって、これまでに一度でも正常にキャッシュへ put されたキーを追跡します。
# `get` が、本当に一度も挿入されていないキーに対して値を返さないことを検証するために使います。
global_ever_inserted_keys = set()
global_ever_inserted_keys_lock = threading.Lock()

def mt_worker(cache_instance, thread_id, barrier_instance, error_event,
              ever_inserted_keys_set, ever_inserted_keys_lock):
    """マルチスレッドテスト用のワーカー関数。"""
    barrier_instance.wait() # すべてのスレッドの準備が整い、操作開始できるまで待機

    for i in range(MT_OPS_PER_THREAD):
        if error_event.is_set():
            break # 別スレッドがすでにエラーを見つけていれば停止

        # 'get' と 'put' に偏らせてランダムに操作を選択
        op = random.choices(['get', 'put', 'delete'], weights=[0.5, 0.4, 0.1], k=1)[0]
        key = str(random.randint(0, MT_KEY_RANGE - 1))
        value = f"value_T{thread_id}_K{key}_I{i}"

        if op == 'put':
            cache_instance.put(key, value)
            with ever_inserted_keys_lock:
                ever_inserted_keys_set.add(key)
        elif op == 'get':
            val = cache_instance.get(key)
            if val is not None:
                # 値が返ってきた場合、そのキーは少なくともどこかの時点で挿入されていなければならないことを確認
                with ever_inserted_keys_lock:
                    if key not in ever_inserted_keys_set:
                        print(f"MT エラー: スレッド {thread_id} がキー '{key}' に対して値 '{val}' を取得しましたが、"
                              f"そのキーはグローバルには一度も挿入されていません。これは起きてはなりません。")
                        error_event.set()
                        break
                # 値が壊れていないことを確認（単純な書式チェック）
                if not isinstance(val, str) or not val.startswith("value_T"):
                    print(f"MT エラー: スレッド {thread_id} がキー '{key}' に対して破損した値 '{val}' を取得しました。")
                    error_event.set()
                    break
        elif op == 'delete':
            cache_instance.delete(key)
        
        # 任意: 小さなランダム sleep を導入してコンテキストスイッチと競合を増やす
        # time.sleep(random.uniform(0.00001, 0.0001))

    # print(f"スレッド {thread_id} が終了しました。") # 詳細出力が必要ならコメント解除

threads = []
for i in range(MT_NUM_THREADS):
    thread = threading.Thread(target=mt_worker,
                              args=(mt_cache, i, barrier, error_flag,
                                    global_ever_inserted_keys, global_ever_inserted_keys_lock))
    threads.append(thread)
    thread.start()

# すべてのワーカースレッドがバリアに到達し、操作を開始するのを待つ
barrier.wait()
print(f"{MT_NUM_THREADS} 個のスレッドがすべて操作を開始しました。")

# すべてのワーカースレッドの操作完了を待つ
for thread in threads:
    thread.join()

print("すべてのスレッドが完了しました。")

# マルチスレッドテストの最終アサーション
assert not error_flag.is_set(), "MT: マルチスレッド操作中にエラーが検出されました。詳細はコンソールを確認してください。"
assert mt_cache._size <= MT_CAPACITY, f"MT: キャッシュサイズ ({mt_cache._size}) が容量 ({MT_CAPACITY}) を超えました。"

# キャッシュに最終的に残っているすべてのキーが、いずれかの時点で挿入されたことを確認
final_keys = mt_cache.keys()
with global_ever_inserted_keys_lock:
    for key in final_keys:
        assert key in global_ever_inserted_keys, f"MT エラー: 最終キャッシュに、グローバルに一度も挿入されていないキー '{key}' が含まれています。"

print("マルチスレッドテストに合格しました！")
print("\n--- すべての LRU キャッシュテストに合格しました！ ---")

if name == 'main':
run_tests()

日本語のみの出力ルール:

• 中国語（簡体字または繁体字）を出力しないでください。自然な日本語のみに翻訳してください。
• テキスト全体を最初から最後まで適切な日本語に翻訳してください。

from future import annotations

import random
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

class RWLock:
"""
シンプルな書き込み優先の readers-writer ロック。

並行実行戦略:
- 複数のリーダーが同時にロックを保持できる。
- ライターは排他的である。
- ひとたびライターが待機していると、新しいリーダーはブロックされ、ライターが飢餓状態にならないようにする。

これは、キーからノードへの対応付けを保持する辞書を保護するために使われる。
存在確認や値の参照だけを必要とする読み取りは並行して進められる。
インデックスまたは LRU リストを変更する操作は書き込みロックを取得する。

注:
Python は、完全にロックフリーな LRU を構築するための真の lock-free プリミティブを標準ライブラリでは提供していない。この実装ではその代わりに、すべてのアクセスに対する単一のグローバル mutex を避けるため、次を用いている:
- key->node インデックス用の readers-writer ロック
- 二重リンク LRU リスト用の別個のロック
- ノード局所状態用のノードごとのロック

これにより、競合下での正しさを保ちながら、単一のグローバルロックよりも大幅に高い並行性が得られる。
"""

def __init__(self) -> None:
    self._cond = threading.Condition(threading.Lock())
    self._readers = 0
    self._writer = False
    self._writers_waiting = 0

def acquire_read(self) -> None:
    with self._cond:
        while self._writer or self._writers_waiting > 0:
            self._cond.wait()
        self._readers += 1

def release_read(self) -> None:
    with self._cond:
        self._readers -= 1
        if self._readers == 0:
            self._cond.notify_all()

def acquire_write(self) -> None:
    with self._cond:
        self._writers_waiting += 1
        try:
            while self._writer or self._readers > 0:
                self._cond.wait()
            self._writer = True
        finally:
            self._writers_waiting -= 1

def release_write(self) -> None:
    with self._cond:
        self._writer = False
        self._cond.notify_all()

@dataclass
class _Node:
key: str
value: Any
prev: Optional["_Node"] = None
next: Optional["_Node"] = None
deleted: bool = False
lock: threading.RLock = field(default_factory=threading.RLock)

class ConcurrentLRUCache:
"""
きめ細かな同期を備えたスレッドセーフな LRU キャッシュ。

設計概要
---------------
データ構造:
- self._map: dict[str, _Node]
  キーをノードに対応付ける。self._index_lock (RWLock) によって保護される。
- 番兵 self._head（MRU 側）および self._tail（LRU 側）を持つ二重リンクリスト
  self._list_lock によって保護される。

ロック規則
-------------
1. 辞書/インデックスの読み取りは self._index_lock.acquire_read() を使う。
2. 辞書/インデックスの変更は self._index_lock.acquire_write() を使う。
3. リンクリスト内のポインタ変更はすべて self._list_lock を使う。
4. ノードごとのロックは、ノードの value/deleted 状態へのアクセスを直列化する。

デッドロックを避けるための順序規則
--------------------------------
複数のロックが必要な場合、操作は安定した順序に従う:
  index lock -> list lock -> node lock(s)

並行性におけるトレードオフ
----------------------
- get() は最近使用時刻を更新する必要があるため、純粋な読み取りにはできない。まず読み取りロック下で並行なインデックス検索を行い、その後、見つかったノードを MRU 位置に移動するために短時間 list lock を取得する。
- put()、delete()、および退避（eviction）はインデックスとリストの両方を変更するため、index write lock と list lock を取得する。
- これは厳密には lock-free ではない。Python の標準ライブラリでは、実用的な lock-free リンクリスト LRU は実現しにくい。しかし、各操作ですべてに対して単一のグローバル mutex を使う設計よりは並行性が高い。
"""

def __init__(self, capacity: int):
    """正の最大容量でキャッシュを初期化する。"""
    if not isinstance(capacity, int) or capacity <= 0:
        raise ValueError("capacity must be a positive integer")

    self._capacity = capacity
    self._map: Dict[str, _Node] = {}
    self._index_lock = RWLock()
    self._list_lock = threading.RLock()

    self._head = _Node("__HEAD__", None)
    self._tail = _Node("__TAIL__", None)
    self._head.next = self._tail
    self._tail.prev = self._head

def _insert_after_head_unlocked(self, node: _Node) -> None:
    """head の直後に node を挿入する。呼び出し側は self._list_lock を保持していなければならない。"""
    first = self._head.next
    node.prev = self._head
    node.next = first
    self._head.next = node
    if first is not None:
        first.prev = node

def _detach_unlocked(self, node: _Node) -> None:
    """リストから node を切り離す。呼び出し側は self._list_lock を保持していなければならない。"""
    prev_node = node.prev
    next_node = node.next
    if prev_node is not None:
        prev_node.next = next_node
    if next_node is not None:
        next_node.prev = prev_node
    node.prev = None
    node.next = None

def _move_to_front_if_present(self, node: _Node) -> None:
    """
    既存ノードがまだ有効なら MRU 位置へ移動する。

    list lock と node lock を使う。呼び出し側が保持する map read lock により、検索中にその node オブジェクトがインデックスから到達可能なままであることが保証される。
    """
    with self._list_lock:
        with node.lock:
            if node.deleted:
                return
            if node.prev is self._head:
                return
            if node.prev is not None and node.next is not None:
                self._detach_unlocked(node)
                self._insert_after_head_unlocked(node)

def get(self, key: str) -> Optional[Any]:
    """
    key が存在すればその値を返し、最も最近使われたものとしてマークする。
    存在しなければ None を返す。
    """
    self._index_lock.acquire_read()
    try:
        node = self._map.get(key)
        if node is None:
            return None
        with node.lock:
            if node.deleted:
                return None
            value = node.value
        self._move_to_front_if_present(node)
        return value
    finally:
        self._index_lock.release_read()

def put(self, key: str, value: Any) -> None:
    """
    キーと値のペアを挿入または更新する。
    容量を超えた場合、最も長く使われていない項目を退避させる。
    """
    self._index_lock.acquire_write()
    try:
        with self._list_lock:
            existing = self._map.get(key)
            if existing is not None:
                with existing.lock:
                    if not existing.deleted:
                        existing.value = value
                        if existing.prev is not self._head:
                            if existing.prev is not None and existing.next is not None:
                                self._detach_unlocked(existing)
                                self._insert_after_head_unlocked(existing)
                        return
                    else:
                        self._map.pop(key, None)

            node = _Node(key, value)
            self._map[key] = node
            self._insert_after_head_unlocked(node)

            while len(self._map) > self._capacity:
                victim = self._tail.prev
                if victim is None or victim is self._head:
                    break
                with victim.lock:
                    if victim.deleted:
                        if victim.prev is not None and victim.next is not None:
                            self._detach_unlocked(victim)
                        self._map.pop(victim.key, None)
                        continue
                    victim.deleted = True
                    self._detach_unlocked(victim)
                    self._map.pop(victim.key, None)
    finally:
        self._index_lock.release_write()

def delete(self, key: str) -> bool:
    """
    キャッシュから key を削除する。
    key が存在していた場合は True、そうでなければ False を返す。
    """
    self._index_lock.acquire_write()
    try:
        with self._list_lock:
            node = self._map.pop(key, None)
            if node is None:
                return False
            with node.lock:
                if node.deleted:
                    return False
                node.deleted = True
                if node.prev is not None and node.next is not None:
                    self._detach_unlocked(node)
                return True
    finally:
        self._index_lock.release_write()

def keys(self) -> List[str]:
    """
    最も最近使われた順から最も長く使われていない順にキーを返す。

    一貫したスナップショットを得るために、index read lock と list lock を取得する。
    """
    result: List[str] = []
    self._index_lock.acquire_read()
    try:
        with self._list_lock:
            cur = self._head.next
            while cur is not None and cur is not self._tail:
                with cur.lock:
                    if not cur.deleted:
                        result.append(cur.key)
                cur = cur.next
        return result
    finally:
        self._index_lock.release_read()

def _debug_size(self) -> int:
    """テスト用の内部ヘルパー。"""
    self._index_lock.acquire_read()
    try:
        return len(self._map)
    finally:
        self._index_lock.release_read()

def _debug_validate(self) -> None:
    """
    テスト用の内部整合性チェッカー。

    検証内容:
    - リストのリンクが正しく構成されていること
    - 生存中の各リストノードが map に存在すること
    - リスト内に重複キーがないこと
    - サイズが容量を超えないこと
    """
    self._index_lock.acquire_read()
    try:
        with self._list_lock:
            assert len(self._map) <= self._capacity, "cache exceeded capacity"

            seen = set()
            cur = self._head
            assert self._head.prev is None
            assert self._tail.next is None

            while cur is not None:
                nxt = cur.next
                if nxt is not None:
                    assert nxt.prev is cur, "broken backward pointer"
                if cur not in (self._head, self._tail):
                    with cur.lock:
                        assert not cur.deleted, "deleted node remained in list"
                        assert cur.key in self._map, "live list node missing from map"
                        assert self._map[cur.key] is cur, "map points to wrong node"
                        assert cur.key not in seen, "duplicate key in list"
                        seen.add(cur.key)
                if cur is self._tail:
                    break
                cur = nxt

            assert seen == set(self._map.keys()), "map/list key mismatch"
    finally:
        self._index_lock.release_read()

def run_tests() -> None:
"""
単一スレッドおよびマルチスレッドのテストを実行する。

マルチスレッドテストで満たす要件:
- 少なくとも 8 スレッド
- 重複するキーに対する get/put/delete の混在
- キャッシュが容量を決して超えないことをアサート
- 一度も挿入されていないキーに対して get が値を返さないことをアサート
"""

# 単一スレッドテスト
c = ConcurrentLRUCache(2)
assert c.get("missing") is None

c.put("a", 1)
c.put("b", 2)
assert c.keys() == ["b", "a"]
assert c.get("a") == 1
assert c.keys() == ["a", "b"]

c.put("c", 3)
assert c.get("b") is None
assert c.keys() == ["c", "a"]

c.put("a", 10)
assert c.get("a") == 10
assert c.keys()[0] == "a"

assert c.delete("a") is True
assert c.delete("a") is False
assert c.get("a") is None
assert c.keys() == ["c"]

c1 = ConcurrentLRUCache(1)
c1.put("x", 1)
assert c1.keys() == ["x"]
c1.put("y", 2)
assert c1.get("x") is None
assert c1.get("y") == 2
assert c1.keys() == ["y"]

c1.put("y", 3)
assert c1.get("y") == 3
assert c1.delete("z") is False
c1._debug_validate()

# 並行テスト
capacity = 16
cache = ConcurrentLRUCache(capacity)
inserted_ever = set()
inserted_lock = threading.Lock()
errors: List[str] = []
errors_lock = threading.Lock()
stop_flag = threading.Event()

key_space = [f"k{i}" for i in range(24)]

def record_error(msg: str) -> None:
    with errors_lock:
        errors.append(msg)

def worker(tid: int) -> None:
    rnd = random.Random(1000 + tid)
    try:
        for i in range(2000):
            key = key_space[rnd.randrange(len(key_space))]
            op = rnd.random()

            if op < 0.45:
                value = (tid, i)
                cache.put(key, value)
                with inserted_lock:
                    inserted_ever.add(key)
            elif op < 0.80:
                value = cache.get(key)
                if value is not None:
                    with inserted_lock:
                        if key not in inserted_ever:
                            record_error(f"get returned value for never-inserted key: {key}")
            else:
                cache.delete(key)

            if i % 100 == 0:
                size = cache._debug_size()
                if size > capacity:
                    record_error(f"capacity exceeded: {size} > {capacity}")
                cache._debug_validate()
    except Exception as exc:
        record_error(f"worker {tid} failed: {exc!r}")
    finally:
        if stop_flag.is_set():
            return

def validator() -> None:
    try:
        for _ in range(400):
            cache._debug_validate()
            size = cache._debug_size()
            if size > capacity:
                record_error(f"capacity exceeded in validator: {size} > {capacity}")
            time.sleep(0.002)
    except Exception as exc:
        record_error(f"validator failed: {exc!r}")
    finally:
        stop_flag.set()

threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
threads += [threading.Thread(target=validator)]

for t in threads:
    t.start()
for t in threads:
    t.join()

cache._debug_validate()
assert cache._debug_size() <= capacity
assert not errors, "\n".join(errors)

# 同一キー競合に焦点を当てたテスト
hot = ConcurrentLRUCache(4)
hot.put("shared", 0)

def same_key_worker(tid: int) -> None:
    rnd = random.Random(2000 + tid)
    for i in range(1000):
        if rnd.random() < 0.5:
            hot.put("shared", (tid, i))
        else:
            _ = hot.get("shared")
        if rnd.random() < 0.1:
            hot.delete("shared")
            hot.put("shared", (tid, i, "reinsert"))

ts = [threading.Thread(target=same_key_worker, args=(i,)) for i in range(8)]
for t in ts:
    t.start()
for t in ts:
    t.join()

hot._debug_validate()
assert hot._debug_size() <= 4
assert hot.get("shared") is not None

print("All tests passed.")

if name == "main":
run_tests()

日本語のみの出力ルール:

• 中国語（簡体字または繁体字）を出力しないこと。自然な日本語にのみ翻訳すること。
• テキスト全体を最初から最後まで適切に日本語へ翻訳すること。