Implementar una caché LRU concurrente sin bloqueo global

import threading import random import time from typing import Any, Optional, List class _Node: """ Representa un nodo en la lista doblemente enlazada utilizada para el ordenamiento LRU. Usa __slots__ para eficiencia de memoria. """ __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: """ Una implementación personalizada de bloqueo de lectura-escritura. Este bloqueo permite que múltiples lectores mantengan el bloqueo de forma concurrente, pero solo un escritor a la vez. Un escritor debe esperar a que todos los lectores terminen, y los lectores deben esperar a que un escritor termine. Esta es una implementación con preferencia por los lectores, lo que significa que nuevos lectores pueden adquirir el bloqueo de lectura incluso si un escritor está esperando. Esta clase también proporciona administradores de contexto para una adquisición y liberación de bloqueos más limpias. """ def __init__(self): self._read_lock = threading.Lock() # Protege el conteo de _readers y la variable de condición self._write_lock = threading.Lock() # El bloqueo de escritura real self._readers = 0 # Número de lectores activos self._no_readers = threading.Condition(self._read_lock) # Condición para que los escritores esperen a los lectores def acquire_read(self): """Adquiere un bloqueo de lectura. Múltiples lectores pueden mantener este bloqueo concurrentemente.""" with self._read_lock: # Espera si un escritor mantiene actualmente el bloqueo de escritura. # Los nuevos lectores se bloquearán si un escritor está activo. while self._write_lock.locked(): self._no_readers.wait() self._readers += 1 def release_read(self): """Libera un bloqueo de lectura.""" with self._read_lock: self._readers -= 1 # Si no quedan lectores, notifica a cualquier escritor en espera if self._readers == 0: self._no_readers.notify_all() def acquire_write(self): """Adquiere un bloqueo de escritura. Solo un escritor puede mantener este bloqueo a la vez.""" # Adquiere primero el bloqueo principal de escritura para señalar la intención de escribir self._write_lock.acquire() with self._read_lock: # Luego adquiere read_lock para esperar a los lectores activos # Espera hasta que todos los lectores activos hayan terminado while self._readers > 0: self._no_readers.wait() def release_write(self): """Libera un bloqueo de escritura.""" with self._read_lock: # Adquiere read_lock para notificar a posibles lectores/escritores self._no_readers.notify_all() # Notifica a cualquier lector o escritor en espera self._write_lock.release() # Libera el bloqueo principal de escritura 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): """Devuelve un administrador de contexto para adquirir un bloqueo de lectura.""" return self._ReadContextManager(self) def write_locked(self): """Devuelve un administrador de contexto para adquirir un bloqueo de escritura.""" return self._WriteContextManager(self) class LRUCache: """ Una implementación de caché LRU (Least Recently Used) segura para hilos. Esta caché usa un ReadWriteLock para gestionar el acceso concurrente. El diseño busca permitir lecturas concurrentes para fallos de caché y la operación `keys()`. Sin embargo, las operaciones que modifican el estado de la caché o el orden LRU (`put`, `delete`, y aciertos de caché en `get`) adquieren un bloqueo de escritura, garantizando atomicidad y consistencia para estas secciones críticas. Este enfoque proporciona mejor concurrencia que un único mutex global para todas las operaciones, especialmente para cargas de trabajo intensivas en lectura con fallos de caché frecuentes o llamadas a `keys()`. La operación `get` emplea un patrón de "actualización": primero intenta adquirir un bloqueo de lectura para comprobar la existencia de la clave. Si la encuentra, libera el bloqueo de lectura y adquiere un bloqueo de escritura para actualizar el orden LRU y recuperar el valor más reciente. Este patrón se implementa cuidadosamente para manejar las condiciones de carrera que pueden ocurrir durante la actualización del bloqueo (p. ej., que otra hebra elimine o actualice la clave). """ def __init__(self, capacity: int): if capacity <= 0: raise ValueError("La capacidad debe ser un entero positivo.") self._capacity = capacity self._cache = {} # Almacena key -> _Node para búsqueda O(1) self._size = 0 # Nodos ficticios de cabeza y cola para la lista doblemente enlazada. # Esto simplifica los casos límite para agregar/eliminar nodos, ya que los nodos de datos reales # siempre están entre la cabeza y la cola. self._head = _Node("dummy_head", None) self._tail = _Node("dummy_tail", None) self._head.next = self._tail self._tail.prev = self._head # Control de concurrencia: un único ReadWriteLock protege todo el estado de la caché. self._rw_lock = ReadWriteLock() def _add_node(self, node: _Node): """Agrega un nodo justo después de la cabeza (posición de uso más reciente). Asume que se mantiene el bloqueo de escritura.""" node.prev = self._head node.next = self._head.next self._head.next.prev = node self._head.next = node def _remove_node(self, node: _Node): """Elimina un nodo de la lista doblemente enlazada. Asume que se mantiene el bloqueo de escritura.""" 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): """Mueve un nodo existente al frente de la lista (más recientemente usado). Asume que se mantiene el bloqueo de escritura.""" self._remove_node(node) self._add_node(node) def _pop_tail(self) -> _Node: """Elimina y devuelve el nodo menos recientemente usado (el que está justo antes de la cola). Asume que se mantiene el bloqueo de escritura.""" node = self._tail.prev self._remove_node(node) return node def get(self, key: str) -> Optional[Any]: """ Devuelve el valor asociado con la clave si existe, y la marca como usada recientemente. Devuelve None si la clave no está en la caché. Esta operación primero adquiere un bloqueo de lectura para comprobar la existencia de la clave. Si la clave se encuentra, libera el bloqueo de lectura y luego adquiere un bloqueo de escritura para actualizar el orden LRU y recuperar el valor más reciente. Este patrón de "actualización" permite operaciones concurrentes de fallo en `get` y operaciones `keys()`, pero serializa los aciertos de `get` con otras escrituras. """ 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 # Si se encontró la clave, adquiere un bloqueo de escritura para actualizar el orden LRU y obtener el valor más reciente. # Este es un patrón de "actualización". Aquí puede ocurrir una condición de carrera: otra hebra # podría eliminar o actualizar la clave entre la liberación del bloqueo de lectura y la adquisición del bloqueo de escritura. with self._rw_lock.write_locked(): # Vuelve a comprobar si la clave aún existe y se refiere exactamente al mismo objeto nodo. # Esto es crucial para la corrección bajo contención. Si otra hebra realizó # un `put` para la misma clave, `_cache.get(key)` devolvería un nodo *nuevo*. # Si otra hebra realizó un `delete`, devolvería `None`. current_node = self._cache.get(key) if current_node is node_found_under_read_lock: # Asegura que sea exactamente el mismo objeto nodo self._move_to_front(node_found_under_read_lock) value = current_node.value # Obtiene el valor más reciente después de moverlo return value else: # El nodo fue eliminado o reemplazado por otra operación 'put' # entre nuestra adquisición del bloqueo de lectura y el de escritura. Trátalo como un fallo. return None def put(self, key: str, value: Any) -> None: """ Inserta o actualiza el par clave-valor. Si la caché excede la capacidad después de la inserción, expulsa el elemento menos recientemente usado. Esta es una operación de escritura y adquiere un bloqueo de escritura. """ with self._rw_lock.write_locked(): node = self._cache.get(key) if node: # La clave existe: actualiza el valor y la mueve al frente node.value = value self._move_to_front(node) else: # La clave no existe: agrega un nodo nuevo new_node = _Node(key, value) self._cache[key] = new_node self._add_node(new_node) self._size += 1 # Expulsa si se excedió la capacidad 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: """ Elimina la clave de la caché. Devuelve True si la clave estaba presente, False en caso contrario. Esta es una operación de escritura y adquiere un bloqueo de escritura. """ 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]: """ Devuelve una lista de todas las claves actualmente en la caché, ordenadas de la más recientemente usada a la menos recientemente usada. Esta es una operación de lectura y adquiere un bloqueo de lectura. """ 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(): """ Ejecuta una serie de pruebas de un solo hilo y de múltiples hilos para verificar la implementación de LRUCache. """ print("--- Iniciando pruebas de la caché LRU ---") # --- Pruebas de un solo hilo --- print("\n--- Pruebas de un solo hilo ---") cache = LRUCache(capacity=3) assert cache.get("a") is None, "ST: Obtener una clave inexistente debe devolver None" assert cache.keys() == [], "ST: Una caché vacía debe tener una lista de claves vacía" cache.put("a", 1) assert cache.get("a") == 1, "ST: Obtener una clave existente después de put" assert cache.keys() == ["a"], "ST: Lista de claves después de un put" cache.put("b", 2) cache.put("c", 3) assert cache.keys() == ["c", "b", "a"], "ST: Lista de claves después de múltiples puts" assert cache._size == 3, "ST: El tamaño de la caché debe ser 3" # Probar la expulsión por capacidad cache.put("d", 4) # 'a' debe ser expulsada por ser la LRU assert cache.get("a") is None, "ST: La clave expulsada 'a' debe ser None" assert cache.get("d") == 4, "ST: La nueva clave 'd' debe estar presente" assert cache.keys() == ["d", "c", "b"], "ST: Lista de claves después de la expulsión" assert cache._size == 3, "ST: El tamaño de la caché debe seguir siendo 3 después de la expulsión" # Probar put con una clave existente (actualizar valor y mover al frente) cache.put("b", 22) # 'b' debe moverse al frente, valor actualizado assert cache.get("b") == 22, "ST: Valor actualizado para 'b'" assert cache.keys() == ["b", "d", "c"], "ST: Lista de claves después de actualizar y mover 'b'" # Probar delete assert cache.delete("c") is True, "ST: Eliminar la clave existente 'c'" assert cache.get("c") is None, "ST: La clave eliminada 'c' debe ser None" assert cache.keys() == ["b", "d"], "ST: Lista de claves después de eliminar 'c'" assert cache._size == 2, "ST: Tamaño de la caché después de delete" assert cache.delete("z") is False, "ST: Eliminar la clave inexistente 'z'" assert cache.keys() == ["b", "d"], "ST: Lista de claves sin cambios después de eliminar una clave inexistente" # Probar capacidad 1 cache_cap1 = LRUCache(capacity=1) cache_cap1.put("x", 10) assert cache_cap1.get("x") == 10, "ST: Capacidad 1, obtener 'x'" cache_cap1.put("y", 20) # 'x' debe ser expulsada assert cache_cap1.get("x") is None, "ST: Capacidad 1, 'x' expulsada" assert cache_cap1.get("y") == 20, "ST: Capacidad 1, obtener 'y'" assert cache_cap1.keys() == ["y"], "ST: Capacidad 1, lista de claves" print("¡Las pruebas de un solo hilo pasaron!") # --- Pruebas de múltiples hilos --- print("\n--- Pruebas de múltiples hilos ---") MT_CAPACITY = 10 MT_NUM_THREADS = 8 MT_OPS_PER_THREAD = 2000 MT_KEY_RANGE = 20 # Claves de 0 a 19 (p. ej., "0", "1", ..., "19") mt_cache = LRUCache(capacity=MT_CAPACITY) # Barrera para asegurar que todos los hilos trabajadores inicien sus operaciones simultáneamente barrier = threading.Barrier(MT_NUM_THREADS + 1) # +1 para el hilo principal error_flag = threading.Event() # Se activa si algún hilo encuentra un error # Rastrea claves que *alguna vez* fueron insertadas con éxito en la caché por cualquier hilo. # Se usa para verificar que 'get' nunca devuelva un valor para una clave que realmente nunca fue insertada. 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): """Función trabajadora para pruebas de múltiples hilos.""" barrier_instance.wait() # Espera a que todos los hilos estén listos para iniciar operaciones for i in range(MT_OPS_PER_THREAD): if error_event.is_set(): break # Detenerse si otro hilo ya encontró un error # Elegir aleatoriamente una operación con sesgo hacia 'get' y '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: # Verifica que si se devuelve un valor, la clave debe haberse insertado en algún momento. with ever_inserted_keys_lock: if key not in ever_inserted_keys_set: print(f"MT Error: El hilo {thread_id} obtuvo el valor '{val}' para la clave '{key}' " f"que nunca fue insertada globalmente. Esto no debería ocurrir.") error_event.set() break # Verifica que el valor no esté corrupto (comprobación simple del formato) if not isinstance(val, str) or not val.startswith("value_T"): print(f"MT Error: El hilo {thread_id} obtuvo un valor corrupto '{val}' para la clave '{key}'.") error_event.set() break elif op == 'delete': cache_instance.delete(key) # Opcional: Introducir una pequeña pausa aleatoria para aumentar el cambio de contexto y la contención # time.sleep(random.uniform(0.00001, 0.0001)) # print(f"El hilo {thread_id} terminó.") # Descomentar para salida detallada 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() # Esperar a que todos los hilos trabajadores lleguen a la barrera e inicien operaciones barrier.wait() print(f"Todos los {MT_NUM_THREADS} hilos iniciaron operaciones.") # Esperar a que todos los hilos trabajadores completen sus operaciones for thread in threads: thread.join() print("Todos los hilos completaron.") # Aserciones finales para pruebas de múltiples hilos assert not error_flag.is_set(), "MT: Se detectaron errores durante las operaciones multihilo. Revise la consola para más detalles." assert mt_cache._size <= MT_CAPACITY, f"MT: El tamaño de la caché ({mt_cache._size}) excedió la capacidad ({MT_CAPACITY})." # Verificar que todas las claves que permanecen en la caché fueron insertadas en algún momento 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 Error: La caché final contiene la clave '{key}' que nunca fue insertada globalmente." print("¡Las pruebas multihilo pasaron!") print("\n--- ¡Todas las pruebas de la caché LRU pasaron! ---") 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: """ Una cerradura simple de lectores-escritores con preferencia para el escritor. Estrategia de concurrencia: - Múltiples lectores pueden mantener la cerradura de forma concurrente. - Los escritores son exclusivos. - Una vez que un escritor está esperando, se bloquean los nuevos lectores para que los escritores no mueran de hambre. Esto se utiliza para proteger el diccionario que mapea claves a nodos. Las lecturas que solo necesitan búsqueda de pertenencia/valor pueden proceder de forma concurrente. Las operaciones que mutan el índice o la lista LRU toman la cerradura de escritura. Nota: Python no proporciona primitivas verdaderamente libres de cerraduras para construir una lista LRU completamente libre de cerraduras en la biblioteca estándar. Esta implementación, en cambio, evita un único mutex global para todo el acceso utilizando: - una cerradura de lectores-escritores para el índice de clave->nodo - una cerradura separada para la lista LRU doblemente enlazada - cerraduras por nodo para el estado local del nodo Eso proporciona una concurrencia sustancialmente mejor que una única cerradura global mientras se preserva la corrección bajo contención. """ 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: """ Caché LRU segura para hilos con sincronización de grano fino. Descripción general del diseño ----------------------------- Estructuras de datos: - self._map: dict[str, _Node] Mapea claves a nodos. Protegido por self._index_lock (RWLock). - Lista doblemente enlazada con centinelas self._head (lado MRU) y self._tail (lado LRU) Protegida por self._list_lock. Reglas de bloqueo ----------------- 1. Las lecturas del diccionario/índice usan self._index_lock.acquire_read(). 2. Las mutaciones del diccionario/índice usan self._index_lock.acquire_write(). 3. Cualquier cambio de puntero en la lista enlazada usa self._list_lock. 4. Las cerraduras por nodo serializan el acceso al valor/estado de borrado de un nodo. Regla de orden para evitar el interbloqueo ----------------------------------------- Cuando se necesitan múltiples cerraduras, las operaciones siguen un orden estable: cerradura de índice -> cerradura de lista -> cerradura(s) de nodo Compromisos de concurrencia --------------------------- - get() necesita actualizar la recencia, por lo que no puede ser una lectura pura. Primero realiza una búsqueda de índice concurrente bajo una cerradura de lectura, luego toma brevemente la cerradura de lista para mover el nodo localizado a la posición MRU. - put(), delete() y eviction modifican tanto el índice como la lista, por lo que toman la cerradura de escritura del índice y la cerradura de la lista. - Esto no está estrictamente libre de cerraduras; la biblioteca estándar de Python no hace que una lista LRU factible y libre de cerraduras sea factible. Sin embargo, es más concurrente que un diseño que utiliza un único mutex global para cada operación. """ def __init__(self, capacity: int): """Inicializa la caché con una capacidad máxima positiva.""" if not isinstance(capacity, int) or capacity <= 0: raise ValueError("la capacidad debe ser un entero positivo") 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: """Inserta el nodo inmediatamente después de la cabeza. El llamador debe mantener 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: """Separa el nodo de la lista. El llamador debe mantener 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: """ Mueve un nodo existente a la posición MRU si todavía está activo. Utiliza la cerradura de lista más la cerradura del nodo. La cerradura de lectura del mapa mantenida por el llamador asegura que el objeto nodo permanezca accesible desde el índice durante la búsqueda. """ 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]: """ Devuelve el valor para la clave si está presente y lo marca como el más usado recientemente. Devuelve None si está ausente. """ 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: """ Inserta o actualiza un par clave-valor. Si se excede la capacidad, expulsa el elemento menos usado recientemente. """ 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: """ Elimina la clave de la caché. Devuelve True si la clave existía, de lo contrario 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]: """ Devuelve las claves de la más usada recientemente a la menos usada recientemente. Tomamos la cerradura de lectura del índice y la cerradura de lista para obtener una instantánea consistente. """ 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: """Ayudante interno para pruebas.""" self._index_lock.acquire_read() try: return len(self._map) finally: self._index_lock.release_read() def _debug_validate(self) -> None: """ Verificador de consistencia interna para pruebas. Verifica: - los enlaces de lista están bien formados - cada nodo de lista activo está presente en el mapa - no hay claves duplicadas en la lista - el tamaño nunca excede la capacidad """ self._index_lock.acquire_read() try: with self._list_lock: assert len(self._map) <= self._capacity, "la caché excedió la capacidad" 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, "puntero hacia atrás roto" if cur not in (self._head, self._tail): with cur.lock: assert not cur.deleted, "nodo borrado permaneció en la lista" assert cur.key in self._map, "nodo de lista activo faltante en el mapa" assert self._map[cur.key] is cur, "el mapa apunta al nodo incorrecto" assert cur.key not in seen, "clave duplicada en la lista" seen.add(cur.key) if cur is self._tail: break cur = nxt assert seen == set(self._map.keys()), "desajuste de clave mapa/lista" finally: self._index_lock.release_read() def run_tests() -> None: """ Ejecuta pruebas de un solo hilo y multihilo. Requisitos de la prueba multihilo cubiertos: - al menos 8 hilos - mezcla de get/put/delete en claves superpuestas - afirmar que la caché nunca excede la capacidad - afirmar que get nunca devuelve un valor para una clave que nunca se insertó """ # Pruebas de un solo hilo 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() # Pruebas concurrentes 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 devolvió valor para clave nunca insertada: {key}") else: cache.delete(key) if i % 100 == 0: size = cache._debug_size() if size > capacity: record_error(f"capacidad excedida: {size} > {capacity}") cache._debug_validate() except Exception as exc: record_error(f"el trabajador {tid} falló: {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"capacidad excedida en el validador: {size} > {capacity}") time.sleep(0.002) except Exception as exc: record_error(f"el validador falló: {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) # Prueba de contención enfocada en la misma clave 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("Todas las pruebas pasaron.") if __name__ == "__main__": run_tests()