Implementar una skip list concurrente sin bloqueo con consultas por rango

Respuesta: ```cpp #include <atomic> #include <vector> #include <string> #include <optional> #include <random> #include <chrono> #include <thread> #include <unordered_set> #include <iostream> #include <algorithm> #include <stdexcept> #include <memory> // Declaración anticipada struct Node; // Gestor de Hazard Pointers para la recuperación de memoria // Esta es una implementación simplificada con fines de demostración. // Una implementación de Hazard Pointers de grado de producción sería más robusta, // manejando el registro/anulación de registro de hilos, un escaneo más eficiente, // y posiblemente usando una lista global de retirados con recuperación basada en épocas. class HazardPointerManager { public: // Número máximo de hazard pointers que un solo hilo puede mantener simultáneamente. // Normalmente, 2-3 son suficientes para la mayoría de las estructuras de datos lock-free (p. ej., pred, curr, next). static const int MAX_HAZARD_POINTERS_PER_THREAD = 3; // Número máximo de hilos que se espera que usen la skip list de forma concurrente. // Esto limita el tamaño de nuestro arreglo estático hazard_records. static const int MAX_THREADS = 64; // Estructura para almacenar hazard pointers de un solo hilo. struct HazardRecord { std::atomic<Node*> hp[MAX_HAZARD_POINTERS_PER_THREAD]; HazardRecord() { for (int i = 0; i < MAX_HAZARD_POINTERS_PER_THREAD; ++i) { hp[i].store(nullptr, std::memory_order_release); } } }; // Arreglo global de registros de hazard, indexado por ID de hilo. static std::vector<HazardRecord> hazard_records; // Contador para asignar IDs de hilo únicos. static std::atomic<int> thread_id_counter; // Variable local al hilo para almacenar el ID del hilo actual. static thread_local int current_thread_id; // Lista local al hilo de nodos retirados por este hilo. static thread_local std::vector<Node*> retired_nodes_local; // Inicializa el gestor de hazard pointers. Debe llamarse una vez al inicio. static void init() { hazard_records.resize(MAX_THREADS); } // Obtiene el ID único del hilo actual, asignando uno si aún no se ha asignado. static int get_thread_id() { if (current_thread_id == -1) { current_thread_id = thread_id_counter.fetch_add(1, std::memory_order_relaxed); if (current_thread_id >= MAX_THREADS) { throw std::runtime_error("HazardPointerManager: Too many threads registered."); } } return current_thread_id; } // Establece un hazard pointer para el hilo actual. static void set_hazard_pointer(int hp_idx, Node* ptr) { int tid = get_thread_id(); hazard_records[tid].hp[hp_idx].store(ptr, std::memory_order_release); } // Limpia un hazard pointer para el hilo actual. static void clear_hazard_pointer(int hp_idx) { int tid = get_thread_id(); hazard_records[tid].hp[hp_idx].store(nullptr, std::memory_order_release); } // Retira un nodo, agregándolo a la lista local de retirados del hilo actual. // Dispara un escaneo y recuperación si la lista local crece demasiado. static void retire_node(Node* node) { if (node) { retired_nodes_local.push_back(node); // Heurística: escanear cuando el tamaño de la lista local de retirados supere un umbral. // Este umbral debe ajustarse para el rendimiento. if (retired_nodes_local.size() >= MAX_HAZARD_POINTERS_PER_THREAD * thread_id_counter.load(std::memory_order_acquire) * 2) { scan_and_reclaim(); } } } // Escanea todos los hazard pointers y recupera nodos que ya no están protegidos. static void scan_and_reclaim() { std::vector<Node*> to_reclaim; std::vector<Node*> new_retired_nodes_local; // Recopilar todos los punteros protegidos actualmente de todos los hilos activos. std::unordered_set<Node*> protected_pointers; int active_threads = thread_id_counter.load(std::memory_order_acquire); for (int i = 0; i < active_threads; ++i) { for (int j = 0; j < MAX_HAZARD_POINTERS_PER_THREAD; ++j) { Node* p = hazard_records[i].hp[j].load(std::memory_order_acquire); if (p) { protected_pointers.insert(p); } } } // Iterar a través de los nodos retirados locales. for (Node* node : retired_nodes_local) { if (protected_pointers.count(node) == 0) { // El nodo no está protegido por ningún hazard pointer, es seguro eliminarlo. to_reclaim.push_back(node); } else { // Sigue protegido, conservarlo para el siguiente escaneo. new_retired_nodes_local.push_back(node); } } retired_nodes_local = new_retired_nodes_local; // Eliminar los nodos seguros. for (Node* node : to_reclaim) { delete node; } } // Recupera cualquier nodo retirado restante cuando un hilo sale o en el apagado. static void cleanup_retired_nodes() { for (Node* node : retired_nodes_local) { delete node; } retired_nodes_local.clear(); } }; // Definiciones de miembros estáticos para HazardPointerManager std::vector<HazardPointerManager::HazardRecord> HazardPointerManager::hazard_records; std::atomic<int> HazardPointerManager::thread_id_counter(0); thread_local int HazardPointerManager::current_thread_id = -1; thread_local std::vector<Node*> HazardPointerManager::retired_nodes_local; // Estructura de nodo para la Skip List struct Node { int64_t key; std::string value; std::vector<std::atomic<Node*>> next; // Arreglo de punteros para distintos niveles std::atomic<bool> marked_for_deletion; // Marca de eliminación lógica std::atomic<bool> fully_linked; // Indica si el nodo está completamente insertado int level; // Nivel real de este nodo // Constructor para un nodo nuevo Node(int64_t k, const std::string& v, int lvl) : key(k), value(v), level(lvl), marked_for_deletion(false), fully_linked(false) { next.resize(lvl + 1, nullptr); // Inicializar punteros next a nullptr } // El destructor se llama implícitamente cuando la memoria es recuperada por HazardPointerManager. // No se necesita un destructor explícito aquí para la gestión de memoria, ya que se maneja externamente. }; // Estructura de datos Concurrent Skip List class ConcurrentSkipList { private: Node* head; // Nodo centinela de cabeza (key = min_int) Node* tail; // Nodo centinela de cola (key = max_int) int max_level; // Nivel máximo posible para cualquier nodo std::atomic<size_t> current_size; // Número aproximado de elementos activos std::atomic<long> global_version; // Contador de versión para consultas de rango optimistas // Generador de números aleatorios local al hilo para generar niveles static thread_local std::mt19937 generator; static thread_local std::uniform_real_distribution<double> distribution; // Genera un nivel aleatorio para un nodo nuevo usando una distribución geométrica (p=0.5) int random_level() { int level = 0; while (distribution(generator) < 0.5 && level < max_level) { level++; } return level; } // Función auxiliar para recorrer la skip list y encontrar predecesores y sucesores. // Rellena los arreglos `preds` y `succs` con punteros en cada nivel. // Devuelve true si se encuentra un nodo con `key` (independientemente de su estado marcado). // Esta función usa hazard pointers para proteger `pred` y `curr` durante el recorrido. bool find_path(int64_t key, std::vector<Node*>& preds, std::vector<Node*>& succs) { Node* pred = head; Node* curr = nullptr; bool found = false; for (int level = max_level; level >= 0; --level) { curr = pred->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(0, pred); // Proteger pred HazardPointerManager::set_hazard_pointer(1, curr); // Proteger curr // Bucle para avanzar 'curr' hasta que supere 'key' o sea 'tail' while (curr != tail && curr->key < key) { pred = curr; HazardPointerManager::set_hazard_pointer(0, pred); // Proteger nuevo pred curr = pred->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(1, curr); // Proteger nuevo curr } preds[level] = pred; succs[level] = curr; if (curr != tail && curr->key == key) { found = true; } } HazardPointerManager::clear_hazard_pointer(0); HazardPointerManager::clear_hazard_pointer(1); return found; } public: // Constructor: inicializa los nodos centinela head y tail. ConcurrentSkipList(int max_lvl = 32) : max_level(max_lvl), current_size(0), global_version(0) { // Inicializar el generador de números aleatorios local al hilo generator.seed(std::chrono::high_resolution_clock::now().time_since_epoch().count() + std::hash<std::thread::id>{}(std::this_thread::get_id())); // Crear nodos centinela con claves min/max head = new Node(std::numeric_limits<int64_t>::min(), "", max_level); tail = new Node(std::numeric_limits<int64_t>::max(), "", max_level); // Enlazar head con tail en todos los niveles for (int i = 0; i <= max_level; ++i) { head->next[i].store(tail, std::memory_order_release); } head->fully_linked.store(true, std::memory_order_release); tail->fully_linked.store(true, std::memory_order_release); } // Destructor: limpia todos los nodos de la skip list. ~ConcurrentSkipList() { Node* curr = head->next[0].load(std::memory_order_acquire); Node* next_node; while (curr != tail) { next_node = curr->next[0].load(std::memory_order_acquire); delete curr; curr = next_node; } delete head; delete tail; // Asegurarse de que cualquier nodo retirado por este hilo sea limpiado. HazardPointerManager::cleanup_retired_nodes(); } // Inserta un par clave-valor. Actualiza el valor si la clave existe. Devuelve true si se insertó una nueva clave. bool insert(int64_t key, const std::string& value) { int new_level = random_level(); std::vector<Node*> preds(max_level + 1); std::vector<Node*> succs(max_level + 1); while (true) { bool found = find_path(key, preds, succs); if (found) { Node* node_to_update = succs[0]; // Si se encontró y no está marcado para eliminación, actualizar valor. if (!node_to_update->marked_for_deletion.load(std::memory_order_acquire)) { // Actualizar el valor atómicamente. Para std::string, esto no es realmente lock-free // sin mecanismos más complejos (p. ej., intercambio de puntero a una nueva cadena inmutable). // Para este benchmark, se usa asignación directa con una nota sobre su limitación. node_to_update->value = value; return false; // Clave actualizada } else { // Se encontró un nodo con la clave pero marcado para eliminación. Tratar como no encontrado e intentar insertar. // La operación remove debería eventualmente limpiarlo. } } // Validar ruta: asegurar que los predecesores siguen siendo válidos y apuntan a sus sucesores. // Esta comprobación es crucial para evitar insertar en una ruta obsoleta. bool valid_path = true; for (int level = 0; level <= new_level; ++level) { Node* pred = preds[level]; Node* succ = succs[level]; // Si pred está marcado para eliminación o su puntero next cambió, la ruta es inválida. if (pred->marked_for_deletion.load(std::memory_order_acquire) || pred->next[level].load(std::memory_order_acquire) != succ) { valid_path = false; break; } } if (!valid_path) { continue; // Reintentar todo el proceso de inserción } // Crear nuevo nodo Node* new_node = new Node(key, value, new_level); // Enlazar new_node con sus sucesores primero (de abajo hacia arriba para consistencia) for (int level = 0; level <= new_level; ++level) { new_node->next[level].store(succs[level], std::memory_order_release); } // Ahora enlazar predecesores con new_node usando CAS (de abajo hacia arriba) // Si algún CAS falla, otro hilo modificó la lista, así que reintentar. for (int level = 0; level <= new_level; ++level) { Node* pred = preds[level]; Node* expected_succ = succs[level]; if (!pred->next[level].compare_exchange_strong(expected_succ, new_node, std::memory_order_release, std::memory_order_acquire)) { // CAS falló. El new_node no está completamente enlazado y no es alcanzable. // Es seguro eliminarlo directamente. delete new_node; goto retry_insert; // Saltar para reintentar toda la operación } } new_node->fully_linked.store(true, std::memory_order_release); current_size.fetch_add(1, std::memory_order_relaxed); global_version.fetch_add(1, std::memory_order_release); // Incrementar versión en escritura return true; // Nueva clave insertada retry_insert:; // Etiqueta para goto } } // Elimina lógicamente un par clave-valor. Devuelve true si se encontró y eliminó. bool remove(int64_t key) { std::vector<Node*> preds(max_level + 1); std::vector<Node*> succs(max_level + 1); Node* node_to_remove = nullptr; while (true) { bool found = find_path(key, preds, succs); if (!found) { return false; // Clave no encontrada } node_to_remove = succs[0]; // Nodo candidato para eliminación // Si el nodo no está completamente enlazado o ya está marcado para eliminación, reintentar. if (!node_to_remove->fully_linked.load(std::memory_order_acquire) || node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { // Si ya está marcado, otro hilo lo está manejando o ya lo hizo. // Si no está completamente enlazado, es una inserción incompleta, dejar que termine o se limpie. // Por simplicidad, devolvemos false si ya está marcado. if (node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { return false; } continue; // Reintentar si aún no está completamente enlazado } // Marcar atómicamente el nodo para eliminación. if (!node_to_remove->marked_for_deletion.compare_exchange_strong(false, true, std::memory_order_release, std::memory_order_acquire)) { // Otro hilo lo marcó concurrentemente, reintentar. continue; } // El nodo ahora está eliminado lógicamente. Proceder a desvincularlo físicamente. // Validar la ruta de nuevo antes de desvincular. bool valid_path = true; for (int level = 0; level <= node_to_remove->level; ++level) { Node* pred = preds[level]; // Si pred está marcado o su puntero next ya no apunta a node_to_remove, la ruta es inválida. if (pred->marked_for_deletion.load(std::memory_order_acquire) || pred->next[level].load(std::memory_order_acquire) != node_to_remove) { valid_path = false; break; } } if (!valid_path) { continue; // Reintentar todo el proceso de eliminación } // Desvincular de arriba hacia abajo (o de abajo hacia arriba, ambos funcionan para eliminación). // Se usan operaciones CAS para actualizar los punteros next de los predecesores. for (int level = node_to_remove->level; level >= 0; --level) { Node* pred = preds[level]; Node* expected_succ = node_to_remove; Node* new_succ = node_to_remove->next[level].load(std::memory_order_acquire); // Intentar actualizar el puntero next de pred. Si falla, otro hilo podría haberlo // ya desvinculado o insertado algo. No necesitamos reintentar toda la // eliminación, ya que el nodo ya está marcado para eliminación. pred->next[level].compare_exchange_strong(expected_succ, new_succ, std::memory_order_release, std::memory_order_acquire); } current_size.fetch_sub(1, std::memory_order_relaxed); global_version.fetch_add(1, std::memory_order_release); // Incrementar versión en escritura HazardPointerManager::retire_node(node_to_remove); // Retirar el nodo para recuperación return true; } } // Devuelve el valor asociado con la clave, o indica ausencia. std::optional<std::string> find(int64_t key) { Node* curr = head; for (int level = max_level; level >= 0; --level) { Node* next_node = curr->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(0, curr); HazardPointerManager::set_hazard_pointer(1, next_node); while (next_node != tail && next_node->key < key) { curr = next_node; HazardPointerManager::set_hazard_pointer(0, curr); next_node = curr->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(1, next_node); } } // Después del recorrido, curr->next[0] debería ser el nodo objetivo potencial. Node* target = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(2, target); // Proteger target std::optional<std::string> result; if (target != tail && target->key == key && target->fully_linked.load(std::memory_order_acquire) && !target->marked_for_deletion.load(std::memory_order_acquire)) { result = target->value; } HazardPointerManager::clear_hazard_pointer(0); HazardPointerManager::clear_hazard_pointer(1); HazardPointerManager::clear_hazard_pointer(2); return result; } // Devuelve todos los pares clave-valor donde low <= key <= high, como una lista ordenada por clave. // Proporciona aislamiento de instantánea usando un mecanismo de control de concurrencia optimista. std::vector<std::pair<int64_t, std::string>> range_query(int64_t low, int64_t high) { std::vector<std::pair<int64_t, std::string>> result; while (true) { long start_version = global_version.load(std::memory_order_acquire); result.clear(); Node* curr = head->next[0].load(std::memory_order_acquire); // Comenzar en nivel 0 HazardPointerManager::set_hazard_pointer(0, curr); while (curr != tail && curr->key <= high) { if (curr->key >= low) { // Comprobar si el nodo es válido en este punto bool marked = curr->marked_for_deletion.load(std::memory_order_acquire); bool linked = curr->fully_linked.load(std::memory_order_acquire); if (!marked && linked) { result.push_back({curr->key, curr->value}); } } curr = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(0, curr); // Actualizar HP para el siguiente nodo } HazardPointerManager::clear_hazard_pointer(0); long end_version = global_version.load(std::memory_order_acquire); if (start_version == end_version) { // El recorrido del nivel 0 produce naturalmente claves ordenadas. return result; } // La versión cambió, reintentar toda la consulta para obtener una instantánea consistente. } } // Devuelve el número aproximado de elementos activos (no eliminados). // Este conteo es aproximado ya que no está sincronizado con un bloqueo global. size_t size() { size_t count = 0; Node* curr = head->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(0, curr); while (curr != tail) { if (!curr->marked_for_deletion.load(std::memory_order_acquire) && curr->fully_linked.load(std::memory_order_acquire)) { count++; } curr = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(0, curr); } HazardPointerManager::clear_hazard_pointer(0); return count; } }; // Inicialización de miembros estáticos para ConcurrentSkipList thread_local std::mt19937 ConcurrentSkipList::generator; thread_local std::uniform_real_distribution<double> ConcurrentSkipList::distribution(0.0, 1.0); // --- Código de prueba / demostración --- #include <thread> #include <vector> #include <numeric> #include <map> #include <set> #include <algorithm> #include <mutex> // Mutex global para imprimir y evitar salida entrelazada de múltiples hilos std::mutex cout_mutex; void worker_thread(ConcurrentSkipList& skip_list, int thread_id, int num_operations, int key_range_start, int key_range_end) { std::mt19937 rng(std::chrono::high_resolution_clock::now().time_since_epoch().count() + thread_id); std::uniform_int_distribution<int64_t> key_dist(key_range_start, key_range_end); std::uniform_int_distribution<int> op_dist(0, 100); // 0-40: insert, 41-70: remove, 71-90: find, 91-100: range_query for (int i = 0; i < num_operations; ++i) { int op = op_dist(rng); int64_t key = key_dist(rng); std::string value = "value_" + std::to_string(key) + "_thread_" + std::to_string(thread_id); if (op <= 40) { // Insertar bool inserted = skip_list.insert(key, value); { std::lock_guard<std::mutex> lock(cout_mutex); // std::cout << "Thread " << thread_id << ": " << (inserted ? "Inserted" : "Updated") << " key " << key << std::endl; } } else if (op <= 70) { // Eliminar bool removed = skip_list.remove(key); { std::lock_guard<std::mutex> lock(cout_mutex); // std::cout << "Thread " << thread_id << ": " << (removed ? "Removed" : "Not found for removal") << " key " << key << std::endl; } } else if (op <= 90) { // Buscar std::optional<std::string> val = skip_list.find(key); { std::lock_guard<std::mutex> lock(cout_mutex); // if (val) { // std::cout << "Thread " << thread_id << ": Found key " << key << ", value: " << *val << std::endl; // } else { // std::cout << "Thread " << thread_id << ": Key " << key << " not found." << std::endl; // } } } else { // Consulta de rango int64_t low = key_dist(rng); int64_t high = key_dist(rng); if (low > high) std::swap(low, high); std::vector<std::pair<int64_t, std::string>> range_result = skip_list.range_query(low, high); { std::lock_guard<std::mutex> lock(cout_mutex); // std::cout << "Thread " << thread_id << ": Range query [" << low << ", " << high << "] returned " << range_result.size() << " elements." << std::endl; // Para validación, podríamos querer imprimir o almacenar los resultados. // Por brevedad, no imprimimos aquí todos los elementos. } } } } // Un modelo de referencia secuencial simple para validar la skip list concurrente. // Esto no es concurrente, solo es para verificaciones de corrección. class ReferenceSkipList { private: std::map<int64_t, std::string> data; public: bool insert(int64_t key, const std::string& value) { auto it = data.find(key); if (it != data.end()) { it->second = value; return false; } data[key] = value; return true; } bool remove(int64_t key) { return data.erase(key) > 0; } std::optional<std::string> find(int64_t key) { auto it = data.find(key); if (it != data.end()) { return it->second; } return std::nullopt; } std::vector<std::pair<int64_t, std::string>> range_query(int64_t low, int64_t high) { std::vector<std::pair<int64_t, std::string>> result; for (const auto& pair : data) { if (pair.first >= low && pair.first <= high) { result.push_back(pair); } } return result; } size_t size() { return data.size(); } }; void validation_thread(ConcurrentSkipList& concurrent_list, ReferenceSkipList& reference_list, int thread_id, int num_checks, int key_range_start, int key_range_end) { std::mt19937 rng(std::chrono::high_resolution_clock::now().time_since_epoch().count() + thread_id + 1000); std::uniform_int_distribution<int64_t> key_dist(key_range_start, key_range_end); for (int i = 0; i < num_checks; ++i) { int64_t key = key_dist(rng); // Validar la operación find std::optional<std::string> concurrent_val = concurrent_list.find(key); std::optional<std::string> reference_val = reference_list.find(key); // Esta validación es complicada porque la lista de referencia es secuencial y no refleja // el estado concurrente en ningún punto único. Una mejor validación implicaría un modelo // que pueda razonar sobre puntos de linealización o aislamiento de instantánea. // Para una comprobación simple, podemos asegurar que si la lista concurrente encuentra un valor, sea consistente // con lo que *podría* estar en la lista de referencia, o si no se encuentra, que también sea consistente. // Esta es una comprobación débil para una estructura de datos concurrente. // Una comprobación más fuerte requeriría un modelo de referencia concurrente o un verificador basado en historial. // Por ahora, simplemente verificaremos que no haya fallos y una consistencia básica. // Si concurrent_val está presente, idealmente reference_val debería coincidir, o significa // que el modelo de referencia está desincronizado o que la lista concurrente está en un estado transitorio. // Esto tiene más que ver con asegurar que la lista concurrente no falle y devuelva *algún* estado válido. // Validar consulta de rango (comprobación básica: sin fallos, resultados ordenados) int64_t low = key_dist(rng); int64_t high = key_dist(rng); if (low > high) std::swap(low, high); std::vector<std::pair<int64_t, std::string>> range_result = concurrent_list.range_query(low, high); // Comprobar si range_result está ordenado bool is_sorted = true; for (size_t j = 0; j + 1 < range_result.size(); ++j) { if (range_result[j].first > range_result[j+1].first) { is_sorted = false; break; } } if (!is_sorted) { std::lock_guard<std::mutex> lock(cout_mutex); std::cerr << "ERROR: Range query result not sorted! Thread " << thread_id << std::endl; // exit(EXIT_FAILURE); // Para errores críticos } // Comprobar lecturas fantasma en consultas de rango (muy difícil de validar sin una referencia concurrente) // La consulta de rango optimista debería impedir lecturas fantasma reintentando si ocurre una escritura. // Esto se prueba implícitamente mediante el mecanismo del contador de versión. } } int main() { HazardPointerManager::init(); const int MAX_LEVEL = 32; ConcurrentSkipList skip_list(MAX_LEVEL); const int NUM_THREADS = 8; const int OPERATIONS_PER_THREAD = 10000; const int KEY_RANGE_START = 1; const int KEY_RANGE_END = 10000; std::vector<std::thread> workers; for (int i = 0; i < NUM_THREADS; ++i) { workers.emplace_back(worker_thread, std::ref(skip_list), i, OPERATIONS_PER_THREAD, KEY_RANGE_START, KEY_RANGE_END); } // Una lista de referencia simple para comprobaciones básicas de sanidad. No es un validador concurrente completo. ReferenceSkipList reference_list; // Poblar la lista de referencia con algunos datos iniciales, o dejarla vacía. // Para una validación real, necesitarías un modelo concurrente o un enfoque transaccional. // Agregar uno o dos hilos de validación para comprobar continuamente la consistencia // Esta es una validación débil, principalmente para detectar fallos y comprobar propiedades básicas. std::vector<std::thread> validators; const int NUM_VALIDATORS = 2; const int CHECKS_PER_VALIDATOR = 5000; for (int i = 0; i < NUM_VALIDATORS; ++i) { validators.emplace_back(validation_thread, std::ref(skip_list), std::ref(reference_list), NUM_THREADS + i, CHECKS_PER_VALIDATOR, KEY_RANGE_START, KEY_RANGE_END); } for (auto& t : workers) { t.join(); } for (auto& t : validators) { t.join(); } // Comprobación final de size y range_query después de todas las operaciones size_t final_size = skip_list.size(); std::vector<std::pair<int64_t, std::string>> final_range = skip_list.range_query(KEY_RANGE_START, KEY_RANGE_END); std::lock_guard<std::mutex> lock(cout_mutex); std::cout << "\n--- Resultados de la prueba ---" << std::endl; std::cout << "Tamaño aproximado final: " << final_size << std::endl; std::cout << "La consulta de rango final (todas las claves) devolvió " << final_range.size() << " elementos." << std::endl; // Realizar un escaneo y recuperación final para asegurar que todos los nodos retirados sean eliminados. HazardPointerManager::scan_and_reclaim(); std::cout << "Limpieza del Hazard Pointer Manager completada." << std::endl; // Validación básica: si la lista está vacía, la consulta de rango debería estar vacía. // Esta no es una validación fuerte de corrección concurrente, sino de funcionalidad básica. if (final_size == 0 && !final_range.empty()) { std::cerr << "ERROR: List is empty but range query returned elements!" << std::endl; } else if (final_size > 0 && final_range.empty()) { std::cerr << "ERROR: List has elements but range query returned empty!" << std::endl; } std::cout << "Prueba finalizada. Revise la consola por si hay mensajes de ERROR." << std::endl; return 0; } /* Sección de análisis: 1. Garantías de linealizabilidad (o aislamiento de instantánea): - insert(key, value), remove(key), find(key): Estas operaciones están diseñadas para ser linealizables. Una operación `insert` o `remove` exitosa parece surtir efecto instantáneamente en algún punto entre su invocación y su respuesta. Esto se logra mediante el uso de operaciones Compare-And-Swap (CAS) sobre punteros `next` y banderas atómicas (`marked_for_deletion`, `fully_linked`), combinadas con bucles de reintento. Si un CAS falla debido a una modificación concurrente, la operación reintenta, encontrando efectivamente un nuevo punto de linealización. Las operaciones `find` devuelven un valor que estuvo presente en algún punto entre su invocación y su respuesta, reflejando una vista linealizable de la clave individual. - range_query(low, high): Esta operación proporciona **aislamiento de instantánea**. Usa un mecanismo de control de concurrencia optimista que implica un contador `global_version`. Cuando comienza una `range_query`, registra `start_version`. Luego recorre la skip list, recopilando todos los elementos activos (no marcados para eliminación, completamente enlazados) dentro del rango especificado. Después de la recopilación, registra `end_version`. Si `start_version == end_version`, significa que no ocurrieron operaciones de escritura (insert/remove) durante el recorrido y, por lo tanto, el resultado recopilado representa una instantánea consistente de la skip list en un único punto en el tiempo. Si las versiones difieren, implica que ocurrió una escritura y la `range_query` reintenta para obtener una instantánea nueva. Esto garantiza que la lista devuelta no incluya claves que nunca estuvieron presentes simultáneamente durante la ejecución de la operación. - size(): La operación `size()` proporciona un conteo **aproximado**. Recorre el nivel más bajo y cuenta los nodos activos. Como no usa sincronización global, el conteo puede cambiar durante el recorrido, lo que significa que el valor devuelto podría no reflejar el tamaño exacto en ningún punto único del tiempo. Esto es consistente con el requisito del prompt de un

// Concurrent Skip List with Range Queries (Snapshot) in Java // ------------------------------------------------------ // Language choice: Java (GC simplifies memory reclamation vs C++/Rust). // Concurrency strategy: // - The skip list index structure uses lock-free linking with CAS on next pointers // and logical deletion via mark bits (AtomicMarkableReference). // - Updates to a node's value are atomic via AtomicReference<String>. // - Physical removal is helped by traversals (classic Harris-Michael style). // - Range queries provide a consistent snapshot by combining: // (1) A global, lock-free "sequence lock" (seqlock) counter (AtomicLong). // Writers increment it before and after structural changes (odd during write, // even when quiescent). // (2) Readers retry if they observe the counter changed or odd. // This yields a snapshot that corresponds to some instant during the successful read. // It avoids phantoms (keys that were never simultaneously present during the operation). // - No single global mutex is used; seqlock is non-blocking (spinning only on contention). // // Operations: // insert(key, value): insert if absent; else update value atomically. // remove(key): logical delete then physical unlink. // find(key): return value if present and not marked. // range_query(low, high): snapshot-consistent result by seqlock retry. // size(): approximate (LongAdder maintained on successful insert/remove). // // Notes: // - The seqlock counter is a global atomic variable, but not a lock; writers do // not block each other via mutex; they only coordinate by incrementing an atomic. // This does serialize writers at the granularity of the sequence increments, but // structural changes are still CAS-based. // - In Java, ABA is mitigated because references are not reused as raw addresses // in a way that breaks CAS in practice; additionally, we do not recycle nodes. // // Build & run: // javac ConcurrentSkipListSnapshot.java && java ConcurrentSkipListSnapshot // // ------------------------------------------------------ // Analysis (brief): // Linearizability / snapshot: // - insert/remove/find are linearizable with respect to the skip list's abstract set. // insert linearizes at successful CAS linking at level 0 for a new node, or at the // AtomicReference set for update. // remove linearizes at logical mark of next[0]. // - range_query provides a consistent snapshot: it returns a set of (key,value) pairs // corresponding to some instant between its start and end, ensured by the seqlock // check (even sequence unchanged). If concurrent writers occur, it retries. // Complexity: // - Expected O(log n) for find/insert/remove; range_query O(log n + k) where k results. // - size() O(1) approximate. // Limitations: // - Under heavy write contention, range_query may starve due to retries. // - Writers increment a global sequence twice per op; this is a scalability bottleneck // compared to fully lock-free snapshot techniques. // - Physical removal is best-effort; long chains of deleted nodes are eventually cleaned // by helpers. // ------------------------------------------------------ import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.*; public class ConcurrentSkipListSnapshot { static final int MAX_LEVEL = 32; static final double P = 0.5; static final class Node { final long key; final int topLevel; final AtomicReference<String> value; // null is allowed only for sentinels @SuppressWarnings("unchecked") final AtomicMarkableReference<Node>[] next = (AtomicMarkableReference<Node>[]) new AtomicMarkableReference[MAX_LEVEL]; Node(long key, String value, int topLevel) { this.key = key; this.topLevel = topLevel; this.value = new AtomicReference<>(value); for (int i = 0; i < MAX_LEVEL; i++) { next[i] = new AtomicMarkableReference<>(null, false); } } boolean isMarked(int level) { boolean[] mark = new boolean[1]; next[level].get(mark); return mark[0]; } } static final class SkipList { private final Node head; private final Node tail; private final ThreadLocal<Random> rnd = ThreadLocal.withInitial(Random::new); // Snapshot seqlock: even = stable; odd = write in progress. private final AtomicLong seq = new AtomicLong(0); // Approximate size private final LongAdder size = new LongAdder(); SkipList() { tail = new Node(Long.MAX_VALUE, null, MAX_LEVEL - 1); head = new Node(Long.MIN_VALUE, null, MAX_LEVEL - 1); for (int i = 0; i < MAX_LEVEL; i++) { head.next[i].set(tail, false); } } private int randomLevel() { int lvl = 0; // geometric distribution with p=0.5 while (lvl < MAX_LEVEL - 1 && rnd.get().nextDouble() < P) { lvl++; } return lvl; } // Writer section begin/end. Not a lock; provides snapshot consistency for range_query. private void writerBegin() { seq.getAndIncrement(); // make odd } private void writerEnd() { seq.getAndIncrement(); // make even } // Find predecessors and successors; helps unlink marked nodes. // Returns true if key found (succs[0] key==target and not marked at level0). private boolean findSplice(long key, Node[] preds, Node[] succs) { boolean[] marked = {false}; boolean snip; Node pred, curr, succ; retry: while (true) { pred = head; for (int level = MAX_LEVEL - 1; level >= 0; level--) { curr = pred.next[level].getReference(); while (true) { succ = curr.next[level].get(marked); while (marked[0]) { // curr is marked; try to unlink snip = pred.next[level].compareAndSet(curr, succ, false, false); if (!snip) { continue retry; } curr = pred.next[level].getReference(); succ = curr.next[level].get(marked); } if (curr.key < key) { pred = curr; curr = succ; } else { break; } } preds[level] = pred; succs[level] = curr; } return (succs[0].key == key); } } public boolean insert(long key, String val) { Objects.requireNonNull(val, "value"); Node[] preds = new Node[MAX_LEVEL]; Node[] succs = new Node[MAX_LEVEL]; int topLevel = randomLevel(); while (true) { boolean found = findSplice(key, preds, succs); if (found) { Node nodeFound = succs[0]; // If node is logically deleted, help remove and retry if (nodeFound.isMarked(0)) { continue; } // Update value atomically nodeFound.value.set(val); return false; } Node newNode = new Node(key, val, topLevel); for (int level = 0; level <= topLevel; level++) { Node succ = succs[level]; newNode.next[level].set(succ, false); } // structural change: protect range snapshot with seqlock writerBegin(); try { Node pred = preds[0]; Node succ = succs[0]; if (!pred.next[0].compareAndSet(succ, newNode, false, false)) { continue; // retry } // link higher levels for (int level = 1; level <= topLevel; level++) { while (true) { pred = preds[level]; succ = succs[level]; if (pred.next[level].compareAndSet(succ, newNode, false, false)) { break; } findSplice(key, preds, succs); } } size.increment(); return true; } finally { writerEnd(); } } } public boolean remove(long key) { Node[] preds = new Node[MAX_LEVEL]; Node[] succs = new Node[MAX_LEVEL]; Node nodeToRemove; boolean[] marked = {false}; while (true) { boolean found = findSplice(key, preds, succs); if (!found) return false; nodeToRemove = succs[0]; // mark higher levels first for (int level = nodeToRemove.topLevel; level >= 1; level--) { Node succ = nodeToRemove.next[level].get(marked); while (!marked[0]) { nodeToRemove.next[level].attemptMark(succ, true); succ = nodeToRemove.next[level].get(marked); } } // now mark level 0 (linearization point) Node succ0 = nodeToRemove.next[0].get(marked); while (true) { if (marked[0]) { // already removed return false; } writerBegin(); boolean iMarkedIt = false; try { iMarkedIt = nodeToRemove.next[0].compareAndSet(succ0, succ0, false, true); } finally { writerEnd(); } if (iMarkedIt) { // help physical removal findSplice(key, preds, succs); size.decrement(); return true; } else { succ0 = nodeToRemove.next[0].get(marked); } } } } public String find(long key) { Node pred = head; boolean[] marked = {false}; for (int level = MAX_LEVEL - 1; level >= 0; level--) { Node curr = pred.next[level].getReference(); while (true) { Node succ = curr.next[level].get(marked); while (marked[0]) { // help unlink pred.next[level].compareAndSet(curr, succ, false, false); curr = pred.next[level].getReference(); succ = curr.next[level].get(marked); } if (curr.key < key) { pred = curr; curr = succ; } else { break; } } } Node curr = pred.next[0].getReference(); if (curr.key != key) return null; if (curr.isMarked(0)) return null; return curr.value.get(); } public List<Map.Entry<Long, String>> rangeQuery(long low, long high) { if (low > high) return Collections.emptyList(); while (true) { long s1 = seq.get(); if ((s1 & 1L) != 0L) { // writer in progress Thread.onSpinWait(); continue; } ArrayList<Map.Entry<Long, String>> out = new ArrayList<>(); // find first node >= low Node pred = head; boolean[] marked = {false}; for (int level = MAX_LEVEL - 1; level >= 0; level--) { Node curr = pred.next[level].getReference(); while (true) { Node succ = curr.next[level].get(marked); while (marked[0]) { pred.next[level].compareAndSet(curr, succ, false, false); curr = pred.next[level].getReference(); succ = curr.next[level].get(marked); } if (curr.key < low) { pred = curr; curr = succ; } else { break; } } } Node curr = pred.next[0].getReference(); while (curr.key <= high) { boolean[] m0 = {false}; Node succ = curr.next[0].get(m0); if (!m0[0] && curr.key >= low && curr.key != Long.MAX_VALUE) { String v = curr.value.get(); // include current value out.add(new AbstractMap.SimpleImmutableEntry<>(curr.key, v)); } curr = succ; } long s2 = seq.get(); if (s1 == s2 && (s2 & 1L) == 0L) { // consistent snapshot return out; } // else retry } } public long size() { return size.sum(); } // For testing: collect current keys (non-snapshot) under seqlock to compare. public List<Long> snapshotKeys() { List<Map.Entry<Long, String>> entries = rangeQuery(Long.MIN_VALUE, Long.MAX_VALUE); ArrayList<Long> keys = new ArrayList<>(entries.size()); for (Map.Entry<Long, String> e : entries) keys.add(e.getKey()); return keys; } } // ------------------------- Demo / Test ------------------------- public static void main(String[] args) throws Exception { final SkipList sl = new SkipList(); final int THREADS = Math.max(4, Runtime.getRuntime().availableProcessors()); final int OPS_PER_THREAD = 50_000; final long KEY_SPACE = 5_000; ExecutorService pool = Executors.newFixedThreadPool(THREADS); CyclicBarrier barrier = new CyclicBarrier(THREADS); // Reference map for validation using a striped lock (test harness only) final int STRIPES = 64; final ReentrantLock[] locks = new ReentrantLock[STRIPES]; for (int i = 0; i < STRIPES; i++) locks[i] = new ReentrantLock(); final HashMap<Long, String> ref = new HashMap<>(); class Worker implements Callable<Void> { final int id; Worker(int id) { this.id = id; } int stripe(long k) { return (int)(Long.remainderUnsigned(k * 0x9E3779B97F4A7C15L, STRIPES)); } @Override public Void call() throws Exception { Random r = new Random(12345L + id * 99991L); barrier.await(); for (int i = 0; i < OPS_PER_THREAD; i++) { long k = r.nextInt((int)KEY_SPACE); int op = r.nextInt(100); if (op < 40) { // insert/update String v = "v" + id + "_" + i; boolean inserted = sl.insert(k, v); // update ref ReentrantLock lk = locks[stripe(k)]; lk.lock(); try { boolean existed = ref.containsKey(k); ref.put(k, v); if (inserted == existed) { // inserted should be true iff it did not exist throw new AssertionError("insert return mismatch for key " + k); } } finally { lk.unlock(); } } else if (op < 60) { // remove boolean removed = sl.remove(k); ReentrantLock lk = locks[stripe(k)]; lk.lock(); try { boolean existed = ref.containsKey(k); if (existed) ref.remove(k); if (removed != existed) { throw new AssertionError("remove return mismatch for key " + k); } } finally { lk.unlock(); } } else if (op < 85) { // find String got = sl.find(k); ReentrantLock lk = locks[stripe(k)]; lk.lock(); try { String exp = ref.get(k); // find is linearizable; compare directly if (!Objects.equals(got, exp)) { throw new AssertionError("find mismatch key=" + k + " got=" + got + " exp=" + exp); } } finally { lk.unlock(); } } else { // range query validation: compare to a consistent snapshot of ref long a = r.nextInt((int)KEY_SPACE); long b = r.nextInt((int)KEY_SPACE); long low = Math.min(a, b); long high = Math.max(a, b); List<Map.Entry<Long, String>> got = sl.rangeQuery(low, high); // Snapshot ref: lock all stripes in order for (int s = 0; s < STRIPES; s++) locks[s].lock(); List<Map.Entry<Long, String>> exp = new ArrayList<>(); try { for (Map.Entry<Long, String> e : ref.entrySet()) { long kk = e.getKey(); if (kk >= low && kk <= high) { exp.add(new AbstractMap.SimpleImmutableEntry<>(kk, e.getValue())); } } } finally { for (int s = STRIPES - 1; s >= 0; s--) locks[s].unlock(); } exp.sort(Comparator.comparingLong(Map.Entry::getKey)); if (got.size() != exp.size()) { throw new AssertionError("range size mismatch low=" + low + " high=" + high + " got=" + got.size() + " exp=" + exp.size()); } for (int idx = 0; idx < got.size(); idx++) { Map.Entry<Long, String> ge = got.get(idx); Map.Entry<Long, String> ee = exp.get(idx); if (!ge.getKey().equals(ee.getKey()) || !Objects.equals(ge.getValue(), ee.getValue())) { throw new AssertionError("range mismatch at " + idx + " got=" + ge + " exp=" + ee); } } } } return null; } } List<Future<Void>> futs = new ArrayList<>(); for (int i = 0; i < THREADS; i++) futs.add(pool.submit(new Worker(i))); for (Future<Void> f : futs) f.get(); pool.shutdown(); pool.awaitTermination(30, TimeUnit.SECONDS); // Final validation: compare full snapshot List<Map.Entry<Long, String>> gotAll = sl.rangeQuery(Long.MIN_VALUE, Long.MAX_VALUE); // build expected ArrayList<Map.Entry<Long, String>> expAll = new ArrayList<>(); for (int s = 0; s < STRIPES; s++) locks[s].lock(); try { for (Map.Entry<Long, String> e : ref.entrySet()) { expAll.add(new AbstractMap.SimpleImmutableEntry<>(e.getKey(), e.getValue())); } } finally { for (int s = STRIPES - 1; s >= 0; s--) locks[s].unlock(); } expAll.sort(Comparator.comparingLong(Map.Entry::getKey)); if (gotAll.size() != expAll.size()) { throw new AssertionError("final size mismatch got=" + gotAll.size() + " exp=" + expAll.size()); } for (int i = 0; i < gotAll.size(); i++) { Map.Entry<Long, String> g = gotAll.get(i); Map.Entry<Long, String> e = expAll.get(i); if (!g.getKey().equals(e.getKey()) || !Objects.equals(g.getValue(), e.getValue())) { throw new AssertionError("final mismatch at " + i + " got=" + g + " exp=" + e); } } System.out.println("OK. Threads=" + THREADS + " ops/thread=" + OPS_PER_THREAD + " finalSize=" + sl.size()); } }