Implementar una caché LRU concurrente sin bloqueo

Respuesta: ```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("La capacidad debe ser un entero positivo.") self.capacity = capacity self.cache = {} self.head = Node(None, None) # Cabeza ficticia self.tail = Node(None, None) # Cola ficticia self.head.next = self.tail self.tail.prev = self.head self.lock = threading.RLock() # Usando RLock por simplicidad en este ejemplo, pero es posible un enfoque más detallado. 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: # Expulsar el elemento 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) # --- Explicación de la estrategia de concurrencia --- # Esta implementación usa un único threading.RLock para todas las operaciones (get, put, delete). # Aunque esto garantiza la seguridad de los hilos, no minimiza la contención como solicita el enunciado. # Un enfoque más avanzado y detallado implicaría segmentar la caché o usar # estructuras de datos sin bloqueo. Por ejemplo, se podrían usar varios bloqueos, cada uno protegiendo un subconjunto # de las entradas de la caché (bloqueo por franjas), o usar bloqueos de lectura-escritura para permitir que varios lectores # accedan a diferentes partes de la caché de forma concurrente. # El RLock actual garantiza que solo un hilo pueda modificar el estado interno de la caché (el diccionario y la lista doblemente enlazada) a la vez. # Esto evita condiciones de carrera en las que varios hilos podrían intentar actualizar el mismo nodo, agregar/quitar nodos o expulsar entradas simultáneamente, lo que llevaría a corrupción de datos o a un orden LRU incorrecto. # El uso de RLock (bloqueo reentrante) permite que un hilo adquiera el bloqueo varias veces, lo que puede ser útil en escenarios complejos, pero no cambia fundamentalmente las características de contención en comparación con un Lock estándar en este uso específico. # --- Demostración --- 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) # Simular algo de trabajo y permitir que otros hilos se ejecuten time.sleep(0.001) def run_demonstration(): print("Ejecutando demostración de caché 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) # Claves que podrían o no estar en la caché 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"Demostración terminada. Tamaño final de la caché: {len(cache)}") # Verificar que la caché nunca exceda su capacidad assert len(cache) <= capacity, f"¡Se excedió la capacidad de la caché! Tamaño actual: {len(cache)}, Capacidad: {capacity}" print("Verificación superada: se mantiene la capacidad de la caché.") # Comprobación básica de integridad de datos (no exhaustiva) print("Realizando comprobación básica de integridad...") # Esta es una comprobación muy básica. Una prueba más robusta implicaría rastrear los estados esperados. # Por ahora, solo nos aseguramos de que no haya corrupción evidente como tamaño negativo o excepciones durante el acceso. try: # Intentar acceder a algunas claves que probablemente fueron insertadas for i in range(min(5, capacity)): key_to_check = (i * ops_per_thread) % (capacity * 2) cache.get(key_to_check) print("Comprobación básica de integridad superada.") except Exception as e: print(f"La comprobación de integridad falló con la excepción: {e}") assert False, "La comprobación de integridad de la caché falló." if __name__ == "__main__": run_demonstration() ```

Caché LRU concurrente (LRU estricto) con contención minimizada. Estrategia de concurrencia (breve): - La caché utiliza un diccionario para clave->Nodo y una lista doblemente enlazada para el orden LRU global. - Para reducir la contención en las operaciones de clave-valor, el espacio de claves se divide en múltiples bloqueos. Las actualizaciones de una clave particular (inserción/actualización/eliminación/lectura de la asignación y los campos del nodo) utilizan solo el bloqueo de división de esa clave. - Para preservar la semántica *LRU global estricta*, todas las mutaciones de la lista LRU se serializan con un bloqueo `lru_lock` dedicado. Esto evita condiciones de carrera que podrían corromper la lista y asegura que haya un único orden total de operaciones de "toque". - La expulsión requiere coordinación: bajo `lru_lock` seleccionamos la cola actual (menos usada recientemente) y luego adquirimos el bloqueo de división de esa clave para verificar que sigue siendo el mismo nodo y eliminarlo del mapa. Esto evita expulsar incorrectamente una clave actualizada concurrentemente. Por qué esto es correcto: - La lista LRU solo se muta bajo `lru_lock`, por lo que no puede corromperse y cada toque (get/put para claves existentes) tiene un orden total, lo que produce LRU estricto. - La entrada del mapa y el valor del nodo están protegidos por el bloqueo de división para esa clave, lo que evita que las actualizaciones/eliminaciones concurrentes para la misma clave corrompan el estado. - Las expulsiones se basan en la cola real en el momento en que se decide la expulsión bajo `lru_lock`; si el nodo candidato cambia debido a un toque/eliminación concurrente, la expulsión se reintenta. Nota: Python no proporciona primitivas verdaderamente sin bloqueo para estructuras compartidas complejas; esta implementación evita un único bloqueo global alrededor de *todo* separando las responsabilidades: - Las operaciones a nivel de clave se dividen. - Solo el mantenimiento del orden de la lista utiliza un único bloqueo. Se incluye una demostración en __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 # se establece en False cuando se elimina/expulsa class ConcurrentLRUCache: def __init__(self, capacity: int, stripes: int = 64): if capacity <= 0: raise ValueError("capacity must be positive") if stripes <= 0: raise ValueError("stripes must be positive") self.capacity = int(capacity) self._map: Dict[Any, _Node] = {} # Cabeza/cola centinela para la lista doblemente enlazada (MRU en head.next, LRU en tail.prev) self._head = _Node("<HEAD>", None) self._tail = _Node("<TAIL>", None) self._head.next = self._tail self._tail.prev = self._head # Serializa las operaciones de la lista y las decisiones de tamaño/expulsión self._lru_lock = threading.Lock() # Bloqueos divididos para operaciones de mapa/nodo a nivel de clave self._stripes_n = int(stripes) self._stripes: List[threading.Lock] = [threading.Lock() for _ in range(self._stripes_n)] # Tamaño rastreado bajo lru_lock self._size = 0 def _stripe_lock(self, key: Any) -> threading.Lock: return self._stripes[hash(key) % self._stripes_n] # ----- Ayudantes de lista doblemente enlazada (requieren mantener 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: # el nodo debe estar en la lista self._unlink(node) self._link_front(node) def _current_lru_node(self) -> Optional[_Node]: # devuelve el nodo LRU real (tail.prev) o None si está vacío n = self._tail.prev if n is None or n is self._head: return None return n # ----- API pública ----- 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 # Mover al frente requiere el bloqueo de la lista; hacerlo fuera del bloqueo de división para reducir la contención. with self._lru_lock: # Recomprobar la membresía bajo el bloqueo de división para evitar mover un nodo eliminado/reemplazado. with lock: cur = self._map.get(key) if cur is not node or cur is None or not cur.alive: return value # el valor era válido en el momento de la lectura; el toque se perdió debido a una condición de carrera. # Bajo lru_lock, podemos mover este nodo de forma segura. self._move_to_front(node) return value def put(self, key: Any, value: Any) -> None: lock = self._stripe_lock(key) # Primero, comprueba si la clave existe y actualiza bajo el bloqueo de división. with lock: node = self._map.get(key) if node is not None and node.alive: node.value = value # Tocar (mover al frente) bajo lru_lock. # ¿Liberar el bloqueo de división rápidamente? Podemos mantenerlo corto: la actualización está hecha. pass else: node = None if node is not None: with self._lru_lock: # Asegurarse de que todavía está presente y es el mismo nodo 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 # Insertar nuevo nodo new_node = _Node(key, value) with self._lru_lock: # Adquirir el bloqueo de división mientras se mantiene lru_lock para garantizar la inserción+enlace consistentes. with lock: existing = self._map.get(key) if existing is not None and existing.alive: # Alguien insertó concurrentemente; actualiza ese en su lugar. existing.value = value # Tocarlo self._move_to_front(existing) return # Insertar asignación y enlace self._map[key] = new_node self._link_front(new_node) self._size += 1 # Expulsar mientras se supera la capacidad while self._size > self.capacity: if not self._evict_one_locked(): # Si la expulsión falló debido a condiciones de carrera, reintentar continue def _evict_one_locked(self) -> bool: """Expulsa un único nodo LRU. Requiere mantener self._lru_lock. Devuelve True si se produjo una expulsión, False si necesita reintentar. """ 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: # la víctima cambió o ya se eliminó; reintentar return False # Eliminar del mapa primero del self._map[vkey] victim.alive = False # Ahora eliminar de la lista (mantenemos lru_lock) self._unlink(victim) self._size -= 1 return True def delete(self, key: Any) -> bool: lock = self._stripe_lock(key) # Adquirir lru_lock primero para evitar interbloqueos con la ruta de expulsión/inserción. 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 # Desenlazar de la lista 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: """Rutina de depuración/verificación: verifica la consistencia interna.""" with self._lru_lock: # Comprobar la estructura de la lista y contar los nodos 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 # El mapa y la lista coinciden (mejor esfuerzo; se necesitan bloqueos de división para ser perfectos). # Bajo lru_lock, el mapa puede estar cambiando bajo bloqueos de división; por lo que solo hacemos una verificación conservadora: # cada nodo de la lista debe existir en el mapa. 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) # Afirmar ocasionalmente que el tamaño no excede la capacidad 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() # Ejecutar un poco time.sleep(1.5) # Detener y unir stop.set() for t in threads: t.join(timeout=2.0) # Sin errores de trabajador if errors: raise errors[0] # Verificaciones finales de invariantes assert len(cache) <= cap cache._assert_invariants() # Sanidad básica: put/get repetidos en la misma clave desde múltiples hilos no deben corromper 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("Demo completada; los invariantes de la caché se mantienen.")