Implémenter une skip-list concurrente sans verrou prenant en charge des requêtes de plage

Réponse : ```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> // Déclaration anticipée struct Node; // Gestionnaire de pointeurs de danger pour la récupération mémoire // Il s'agit d'une implémentation simplifiée à des fins de démonstration. // Une implémentation de pointeurs de danger de niveau production serait plus robuste, // en gérant l'enregistrement/désenregistrement des threads, un balayage plus efficace, // et en utilisant potentiellement une liste globale de nœuds retirés avec récupération basée sur les époques. class HazardPointerManager { public: // Nombre maximal de pointeurs de danger qu'un seul thread peut détenir simultanément. // Typiquement, 2-3 suffisent pour la plupart des structures de données sans verrou (par ex., pred, curr, next). static const int MAX_HAZARD_POINTERS_PER_THREAD = 3; // Nombre maximal de threads censés utiliser la skip list simultanément. // Cela limite la taille de notre tableau statique hazard_records. static const int MAX_THREADS = 64; // Structure pour stocker les pointeurs de danger d'un seul thread. 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); } } }; // Tableau global des enregistrements de danger, indexé par identifiant de thread. static std::vector<HazardRecord> hazard_records; // Compteur pour attribuer des identifiants uniques aux threads. static std::atomic<int> thread_id_counter; // Variable locale au thread pour stocker l'identifiant du thread courant. static thread_local int current_thread_id; // Liste locale au thread des nœuds retirés par ce thread. static thread_local std::vector<Node*> retired_nodes_local; // Initialise le gestionnaire de pointeurs de danger. Doit être appelé une fois au démarrage. static void init() { hazard_records.resize(MAX_THREADS); } // Obtient l'identifiant unique du thread courant, en en attribuant un s'il n'est pas encore assigné. 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: Trop de threads enregistrés."); } } return current_thread_id; } // Définit un pointeur de danger pour le thread courant. 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); } // Efface un pointeur de danger pour le thread courant. 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); } // Retire un nœud en l'ajoutant à la liste locale des nœuds retirés du thread courant. // Déclenche un balayage et une récupération si la liste locale devient trop grande. static void retire_node(Node* node) { if (node) { retired_nodes_local.push_back(node); // Heuristique : balayer lorsque la taille de la liste locale de nœuds retirés dépasse un seuil. // Ce seuil doit être ajusté pour les performances. if (retired_nodes_local.size() >= MAX_HAZARD_POINTERS_PER_THREAD * thread_id_counter.load(std::memory_order_acquire) * 2) { scan_and_reclaim(); } } } // Balaye tous les pointeurs de danger et récupère les nœuds qui ne sont plus protégés. static void scan_and_reclaim() { std::vector<Node*> to_reclaim; std::vector<Node*> new_retired_nodes_local; // Collecte tous les pointeurs actuellement protégés provenant de tous les threads actifs. 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); } } } // Parcourt les nœuds retirés locaux. for (Node* node : retired_nodes_local) { if (protected_pointers.count(node) == 0) { // Le nœud n'est protégé par aucun pointeur de danger, suppression sûre. to_reclaim.push_back(node); } else { // Toujours protégé, on le conserve pour le prochain balayage. new_retired_nodes_local.push_back(node); } } retired_nodes_local = new_retired_nodes_local; // Supprime les nœuds sûrs. for (Node* node : to_reclaim) { delete node; } } // Récupère tous les nœuds retirés restants lorsqu'un thread se termine ou à l'arrêt. static void cleanup_retired_nodes() { for (Node* node : retired_nodes_local) { delete node; } retired_nodes_local.clear(); } }; // Définitions des membres statiques de 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; // Structure de nœud pour la Skip List struct Node { int64_t key; std::string value; std::vector<std::atomic<Node*>> next; // Tableau de pointeurs pour différents niveaux std::atomic<bool> marked_for_deletion; // Drapeau de suppression logique std::atomic<bool> fully_linked; // Indique si le nœud est entièrement inséré int level; // Niveau réel de ce nœud // Constructeur d'un nouveau nœud 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); // Initialise les pointeurs next à nullptr } // Le destructeur est appelé implicitement lorsque la mémoire est récupérée par HazardPointerManager. // Aucun destructeur explicite n'est nécessaire ici pour la gestion mémoire, car elle est gérée à l'extérieur. }; // Structure de données Skip List concurrente class ConcurrentSkipList { private: Node* head; // Nœud sentinelle de tête (key = min_int) Node* tail; // Nœud sentinelle de fin (key = max_int) int max_level; // Niveau maximal possible pour n'importe quel nœud std::atomic<size_t> current_size; // Nombre approximatif d'éléments actifs std::atomic<long> global_version; // Compteur de version pour les requêtes de plage optimistes // Générateur de nombres aléatoires local au thread pour la génération des niveaux static thread_local std::mt19937 generator; static thread_local std::uniform_real_distribution<double> distribution; // Génère un niveau aléatoire pour un nouveau nœud en utilisant une distribution géométrique (p=0.5) int random_level() { int level = 0; while (distribution(generator) < 0.5 && level < max_level) { level++; } return level; } // Fonction auxiliaire pour parcourir la skip list et trouver les prédécesseurs et successeurs. // Remplit les tableaux `preds` et `succs` avec des pointeurs à chaque niveau. // Renvoie true si un nœud avec `key` est trouvé (indépendamment de son état marqué). // Cette fonction utilise des pointeurs de danger pour protéger `pred` et `curr` pendant le parcours. 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); // Protège pred HazardPointerManager::set_hazard_pointer(1, curr); // Protège curr // Boucle pour avancer 'curr' jusqu'à ce qu'il dépasse 'key' ou soit 'tail' while (curr != tail && curr->key < key) { pred = curr; HazardPointerManager::set_hazard_pointer(0, pred); // Protège le nouveau pred curr = pred->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(1, curr); // Protège le nouveau 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: // Constructeur : initialise les nœuds sentinelles head et tail. ConcurrentSkipList(int max_lvl = 32) : max_level(max_lvl), current_size(0), global_version(0) { // Initialise le générateur de nombres aléatoires local au thread generator.seed(std::chrono::high_resolution_clock::now().time_since_epoch().count() + std::hash<std::thread::id>{}(std::this_thread::get_id())); // Crée les nœuds sentinelles avec les clés min/max head = new Node(std::numeric_limits<int64_t>::min(), "", max_level); tail = new Node(std::numeric_limits<int64_t>::max(), "", max_level); // Relie head à tail à tous les niveaux 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); } // Destructeur : nettoie tous les nœuds 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; // S'assure que tous les nœuds retirés par ce thread sont nettoyés. HazardPointerManager::cleanup_retired_nodes(); } // Insère une paire clé-valeur. Met à jour la valeur si la clé existe. Renvoie true si une nouvelle clé est insérée. 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 trouvé et non marqué pour suppression, met à jour la valeur. if (!node_to_update->marked_for_deletion.load(std::memory_order_acquire)) { // Met à jour atomiquement la valeur. Pour std::string, ce n'est pas vraiment sans verrou // sans mécanismes plus complexes (par ex. échange de pointeur vers une nouvelle chaîne immuable). // Pour ce benchmark, une affectation directe est utilisée avec une note sur sa limitation. node_to_update->value = value; return false; // Clé mise à jour } else { // Nœud avec la clé trouvé mais marqué pour suppression. Le traiter comme non trouvé et essayer d'insérer. // L'opération remove devrait finir par le nettoyer. } } // Valide le chemin : s'assure que les prédécesseurs sont toujours valides et pointent vers leurs successeurs. // Cette vérification est cruciale pour éviter une insertion dans un chemin obsolète. bool valid_path = true; for (int level = 0; level <= new_level; ++level) { Node* pred = preds[level]; Node* succ = succs[level]; // Si pred est marqué pour suppression ou si son pointeur next a changé, le chemin est invalide. 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; // Réessaie tout le processus d'insertion } // Crée un nouveau nœud Node* new_node = new Node(key, value, new_level); // Relie d'abord new_node à ses successeurs (de bas en haut pour la cohérence) for (int level = 0; level <= new_level; ++level) { new_node->next[level].store(succs[level], std::memory_order_release); } // Maintenant relie les prédécesseurs à new_node en utilisant CAS (de bas en haut) // Si un CAS échoue, un autre thread a modifié la liste, donc on réessaie. 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 a échoué. Le new_node n'est pas entièrement relié et n'est pas atteignable. // Il est sûr de le supprimer directement. delete new_node; goto retry_insert; // Saute pour réessayer toute l'opération } } 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); // Incrémente la version à l'écriture return true; // Nouvelle clé insérée retry_insert:; // Étiquette pour goto } } // Supprime logiquement une paire clé-valeur. Renvoie true si elle est trouvée et supprimée. 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; // Clé non trouvée } node_to_remove = succs[0]; // Nœud candidat à la suppression // Si le nœud n'est pas entièrement relié ou déjà marqué pour suppression, réessaie. if (!node_to_remove->fully_linked.load(std::memory_order_acquire) || node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { // Si déjà marqué, un autre thread s'en occupe ou l'a déjà fait. // Si pas entièrement relié, c'est une insertion incomplète, on la laisse se terminer ou être nettoyée. // Par simplicité, on renvoie false si déjà marqué. if (node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { return false; } continue; // Réessaie si pas encore entièrement relié } // Marque atomiquement le nœud pour suppression. if (!node_to_remove->marked_for_deletion.compare_exchange_strong(false, true, std::memory_order_release, std::memory_order_acquire)) { // Un autre thread l'a marqué concurremment, réessaie. continue; } // Le nœud est maintenant logiquement supprimé. Procède à son déchaînage physique. // Valide de nouveau le chemin avant de le déchaîner. bool valid_path = true; for (int level = 0; level <= node_to_remove->level; ++level) { Node* pred = preds[level]; // Si pred est marqué ou si son pointeur next ne pointe plus vers node_to_remove, le chemin est invalide. 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; // Réessaie tout le processus de suppression } // Déchaînage de haut en bas (ou de bas en haut, les deux fonctionnent pour remove). // Les opérations CAS sont utilisées pour mettre à jour les pointeurs next des prédécesseurs. 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); // Tente de mettre à jour le pointeur next de pred. En cas d'échec, un autre thread a peut-être // déjà déchaîné le nœud ou inséré quelque chose. Il n'est pas nécessaire de réessayer toute // l'opération remove, car le nœud est déjà marqué pour suppression. 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); // Incrémente la version à l'écriture HazardPointerManager::retire_node(node_to_remove); // Retire le nœud pour récupération return true; } } // Renvoie la valeur associée à la clé, ou indique son absence. 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); } } // Après le parcours, curr->next[0] devrait être le nœud cible potentiel. Node* target = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(2, target); // Protège 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; } // Renvoie toutes les paires clé-valeur telles que low <= key <= high, sous forme de liste triée par clé. // Fournit une isolation par instantané en utilisant un mécanisme de contrôle de concurrence optimiste. 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); // Commence au niveau 0 HazardPointerManager::set_hazard_pointer(0, curr); while (curr != tail && curr->key <= high) { if (curr->key >= low) { // Vérifie si le nœud est valide à ce stade 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); // Met à jour le HP pour le nœud suivant } HazardPointerManager::clear_hazard_pointer(0); long end_version = global_version.load(std::memory_order_acquire); if (start_version == end_version) { // Le parcours du niveau 0 produit naturellement des clés triées. return result; } // La version a changé, réessaie toute la requête pour obtenir un instantané cohérent. } } // Renvoie le nombre approximatif d'éléments actifs (non supprimés). // Ce compte est approximatif car il n'est pas synchronisé avec un verrou 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; } }; // Initialisation des membres statiques de ConcurrentSkipList thread_local std::mt19937 ConcurrentSkipList::generator; thread_local std::uniform_real_distribution<double> ConcurrentSkipList::distribution(0.0, 1.0); // --- Code de test / démonstration --- #include <thread> #include <vector> #include <numeric> #include <map> #include <set> #include <algorithm> #include <mutex> // Mutex global pour l'affichage afin d'éviter des sorties entrelacées provenant de plusieurs threads 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) { // Insert 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) { // Remove 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) { // Find 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 { // Range Query 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; // Pour la validation, on pourrait vouloir afficher ou stocker les résultats. // Par souci de brièveté, on n'affiche pas tous les éléments ici. } } } } // Un simple modèle de référence séquentiel pour valider la skip list concurrente. // Il n'est pas concurrent, seulement destiné aux vérifications de correction. 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); // Valide l'opération find std::optional<std::string> concurrent_val = concurrent_list.find(key); std::optional<std::string> reference_val = reference_list.find(key); // Cette validation est délicate parce que la liste de référence est séquentielle et ne reflète pas // l'état concurrent à un instant précis. Une meilleure validation impliquerait un modèle // capable de raisonner sur les points de linéarisation ou sur l'isolation par instantané. // Pour une vérification simple, on peut s'assurer que si la liste concurrente trouve une valeur, elle est cohérente // avec ce qui *pourrait* être dans la liste de référence, ou que si rien n'est trouvé, cela reste cohérent. // Il s'agit d'une vérification faible pour une structure de données concurrente. // Une vérification plus forte nécessiterait un modèle de référence concurrent ou un vérificateur basé sur l'historique. // Pour l'instant, on vérifie simplement l'absence de crashs et une cohérence de base. // Si concurrent_val est présent, reference_val devrait idéalement correspondre, ou cela signifie // que le modèle de référence est désynchronisé ou que la liste concurrente est dans un état transitoire. // Il s'agit davantage de s'assurer que la liste concurrente ne plante pas et renvoie *un* état valide. // Valide range_query (vérification de base : pas de crashs, résultats triés) 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); // Vérifie si range_result est trié 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: Le résultat de la requête de plage n'est pas trié ! Thread " << thread_id << std::endl; // exit(EXIT_FAILURE); // Pour les erreurs critiques } // Vérifie les lectures fantômes dans les requêtes de plage (très difficile à valider sans référence concurrente) // La requête de plage optimiste devrait empêcher les lectures fantômes en réessayant si une écriture survient. // Ceci est implicitement testé par le mécanisme de compteur de version. } } 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); } // Une simple liste de référence pour des vérifications de bon sens élémentaires. Pas un validateur concurrent complet. ReferenceSkipList reference_list; // Remplissez la liste de référence avec quelques données initiales, ou laissez-la vide. // Pour une vraie validation, il faudrait un modèle concurrent ou une approche transactionnelle. // Ajoutez un ou deux threads de validation pour vérifier la cohérence en continu // Il s'agit d'une validation faible, principalement pour détecter les crashs et vérifier des propriétés de base. 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(); } // Vérification finale de la taille et de la requête de plage après toutes les opérations 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--- Résultats des tests ---" << std::endl; std::cout << "Taille approximative finale : " << final_size << std::endl; std::cout << "La requête de plage finale (toutes les clés) a renvoyé " << final_range.size() << " éléments." << std::endl; // Effectue un balayage et une récupération finaux pour s'assurer que tous les nœuds retirés sont supprimés. HazardPointerManager::scan_and_reclaim(); std::cout << "Nettoyage du Hazard Pointer Manager terminé." << std::endl; // Validation de base : si la liste est vide, la requête de plage devrait être vide. // Ce n'est pas une validation forte de la correction concurrente, mais de la fonctionnalité de base. if (final_size == 0 && !final_range.empty()) { std::cerr << "ERROR: La liste est vide mais la requête de plage a renvoyé des éléments !" << std::endl; } else if (final_size > 0 && final_range.empty()) { std::cerr << "ERROR: La liste contient des éléments mais la requête de plage a renvoyé vide !" << std::endl; } std::cout << "Test terminé. Vérifiez la console pour d'éventuels messages ERROR." << std::endl; return 0; } /* Section d'analyse : 1. Garanties de linéarisabilité (ou d'isolation par instantané) : - insert(key, value), remove(key), find(key) : Ces opérations sont conçues pour être linéarisables. Une opération `insert` ou `remove` réussie semble prendre effet instantanément à un certain point entre son invocation et sa réponse. Cela est obtenu grâce à l'utilisation d'opérations Compare-And-Swap (CAS) sur les pointeurs `next` et de drapeaux atomiques (`marked_for_deletion`, `fully_linked`), combinés à des boucles de réessai. Si un CAS échoue à cause d'une modification concurrente, l'opération réessaie, trouvant ainsi effectivement un nouveau point de linéarisation. Les opérations `find` renvoient une valeur qui était présente à un certain moment entre leur invocation et leur réponse, reflétant une vue linéarisable de la clé individuelle. - range_query(low, high) : Cette opération fournit une **isolation par instantané**. Elle utilise un mécanisme de contrôle de concurrence optimiste impliquant un compteur `global_version`. Lorsqu'une `range_query` démarre, elle enregistre `start_version`. Elle parcourt ensuite la skip list en collectant tous les éléments actifs (non marqués pour suppression, entièrement reliés) dans l'intervalle spécifié. Après la collecte, elle enregistre `end_version`. Si `start_version == end_version`, cela signifie qu'aucune opération d'écriture (insert/remove) ne s'est produite pendant le parcours, et ainsi le résultat collecté représente un instantané cohérent de la skip list à un moment donné. Si les versions diffèrent, cela implique qu'une écriture s'est produite, et la `range_query` réessaie pour obtenir un nouvel instantané. Cela garantit que la liste renvoyée n'inclut pas de clés qui n'ont jamais été présentes simultanément pendant l'exécution de l'opération. - size() : L'opération `size()` fournit un compte **approximatif**. Elle parcourt le niveau le plus bas et compte les nœuds actifs. Comme elle n'utilise pas de synchronisation globale, le compte peut changer pendant le parcours, ce qui signifie que la valeur renvoyée peut ne pas refléter la taille exacte à un instant précis. Cela est cohérent avec l'exigence de l'invite pour 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()); } }