Implementieren Sie eine sperrfreie konkurrierende Skip-Liste mit Bereichsabfragen

Antwort: ```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> // Vorwärtsdeklaration struct Node; // Hazard-Pointer-Manager für Speicherbereinigung // Dies ist eine vereinfachte Implementierung zu Demonstrationszwecken. // Eine produktionsreife Hazard-Pointer-Implementierung wäre robuster, // würde Thread-Registrierung/-Abmeldung, effizienteres Scannen // und möglicherweise eine globale Retired-Liste mit epochenbasierter Bereinigung behandeln. class HazardPointerManager { public: // Maximale Anzahl von Hazard Pointern, die ein einzelner Thread gleichzeitig halten kann. // Typischerweise reichen 2–3 für die meisten lock-freien Datenstrukturen aus (z. B. pred, curr, next). static const int MAX_HAZARD_POINTERS_PER_THREAD = 3; // Maximale Anzahl von Threads, von denen erwartet wird, dass sie die Skip-Liste gleichzeitig verwenden. // Dies begrenzt die Größe unseres statischen hazard_records-Arrays. static const int MAX_THREADS = 64; // Struktur zum Halten von Hazard Pointern für einen einzelnen 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); } } }; // Globales Array von Hazard-Records, indiziert nach Thread-ID. static std::vector<HazardRecord> hazard_records; // Zähler zur Vergabe eindeutiger Thread-IDs. static std::atomic<int> thread_id_counter; // Thread-lokale Variable zum Speichern der ID des aktuellen Threads. static thread_local int current_thread_id; // Thread-lokale Liste von Knoten, die von diesem Thread in den Ruhestand versetzt wurden. static thread_local std::vector<Node*> retired_nodes_local; // Initialisiert den Hazard-Pointer-Manager. Muss beim Start einmal aufgerufen werden. static void init() { hazard_records.resize(MAX_THREADS); } // Holt die eindeutige ID des aktuellen Threads und vergibt eine, falls noch keine vergeben wurde. 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: Zu viele Threads registriert."); } } return current_thread_id; } // Setzt einen Hazard Pointer für den aktuellen Thread. 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); } // Löscht einen Hazard Pointer für den aktuellen Thread. 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); } // Versetzt einen Knoten in den Ruhestand und fügt ihn der lokalen Retired-Liste des aktuellen Threads hinzu. // Löst ein Scannen und Bereinigen aus, wenn die lokale Liste zu groß wird. static void retire_node(Node* node) { if (node) { retired_nodes_local.push_back(node); // Heuristik: scannen, wenn die Größe der lokalen Retired-Liste einen Schwellenwert überschreitet. // Dieser Schwellenwert sollte für die Leistung abgestimmt werden. if (retired_nodes_local.size() >= MAX_HAZARD_POINTERS_PER_THREAD * thread_id_counter.load(std::memory_order_acquire) * 2) { scan_and_reclaim(); } } } // Scannt alle Hazard Pointer und gibt Knoten frei, die nicht mehr geschützt sind. static void scan_and_reclaim() { std::vector<Node*> to_reclaim; std::vector<Node*> new_retired_nodes_local; // Sammle alle derzeit geschützten Pointer aus allen aktiven Threads. 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); } } } // Iteriere durch lokale Retired-Knoten. for (Node* node : retired_nodes_local) { if (protected_pointers.count(node) == 0) { // Knoten wird von keinem Hazard Pointer geschützt, sicher zu löschen. to_reclaim.push_back(node); } else { // Noch geschützt, für den nächsten Scan behalten. new_retired_nodes_local.push_back(node); } } retired_nodes_local = new_retired_nodes_local; // Lösche die sicheren Knoten. for (Node* node : to_reclaim) { delete node; } } // Gibt verbleibende Retired-Knoten frei, wenn ein Thread endet oder beim Herunterfahren. static void cleanup_retired_nodes() { for (Node* node : retired_nodes_local) { delete node; } retired_nodes_local.clear(); } }; // Definitionen statischer Member für 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; // Knotenstruktur für die Skip-Liste struct Node { int64_t key; std::string value; std::vector<std::atomic<Node*>> next; // Array von Zeigern für verschiedene Ebenen std::atomic<bool> marked_for_deletion; // Flag für logische Löschung std::atomic<bool> fully_linked; // Gibt an, ob der Knoten vollständig eingefügt ist int level; // Tatsächliche Ebene dieses Knotens // Konstruktor für einen neuen Knoten 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); // Initialisiere next-Zeiger mit nullptr } // Der Destruktor wird implizit aufgerufen, wenn der Speicher durch HazardPointerManager freigegeben wird. // Kein expliziter Destruktor ist hier für die Speicherverwaltung nötig, da diese extern gehandhabt wird. }; // Nebenläufige Skip-List-Datenstruktur class ConcurrentSkipList { private: Node* head; // Sentinel-Head-Knoten (key = min_int) Node* tail; // Sentinel-Tail-Knoten (key = max_int) int max_level; // Maximal mögliche Ebene für jeden Knoten std::atomic<size_t> current_size; // Ungefähre Anzahl aktiver Elemente std::atomic<long> global_version; // Versionszähler für optimistische Bereichsabfragen // Thread-lokaler Zufallszahlengenerator für Ebenenerzeugung static thread_local std::mt19937 generator; static thread_local std::uniform_real_distribution<double> distribution; // Erzeugt eine zufällige Ebene für einen neuen Knoten mithilfe einer geometrischen Verteilung (p=0.5) int random_level() { int level = 0; while (distribution(generator) < 0.5 && level < max_level) { level++; } return level; } // Hilfsfunktion zum Durchlaufen der Skip-Liste und Finden von Vorgängern und Nachfolgern. // Füllt die Arrays `preds` und `succs` mit Zeigern auf jeder Ebene. // Gibt true zurück, wenn ein Knoten mit `key` gefunden wird (unabhängig von seinem Markierungsstatus). // Diese Funktion verwendet Hazard Pointer, um `pred` und `curr` während der Traversierung zu schützen. 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); // pred schützen HazardPointerManager::set_hazard_pointer(1, curr); // curr schützen // Schleife, um 'curr' voranzubewegen, bis es hinter 'key' liegt oder 'tail' ist while (curr != tail && curr->key < key) { pred = curr; HazardPointerManager::set_hazard_pointer(0, pred); // neuen pred schützen curr = pred->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(1, curr); // neuen curr schützen } 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: // Konstruktor: Initialisiert Head- und Tail-Sentinel-Knoten. ConcurrentSkipList(int max_lvl = 32) : max_level(max_lvl), current_size(0), global_version(0) { // Thread-lokalen Zufallszahlengenerator initialisieren generator.seed(std::chrono::high_resolution_clock::now().time_since_epoch().count() + std::hash<std::thread::id>{}(std::this_thread::get_id())); // Sentinel-Knoten mit min/max-Schlüsseln erstellen head = new Node(std::numeric_limits<int64_t>::min(), "", max_level); tail = new Node(std::numeric_limits<int64_t>::max(), "", max_level); // head auf allen Ebenen mit tail verknüpfen 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); } // Destruktor: Bereinigt alle Knoten in der Skip-Liste. ~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; // Sicherstellen, dass alle von diesem Thread in den Ruhestand versetzten Knoten bereinigt werden. HazardPointerManager::cleanup_retired_nodes(); } // Fügt ein Schlüssel-Wert-Paar ein. Aktualisiert den Wert, wenn der Schlüssel existiert. Gibt true zurück, wenn ein neuer Schlüssel eingefügt wurde. 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]; // Wenn gefunden und nicht zur Löschung markiert, Wert aktualisieren. if (!node_to_update->marked_for_deletion.load(std::memory_order_acquire)) { // Wert atomar aktualisieren. Für std::string ist dies nicht wirklich lock-frei // ohne komplexere Mechanismen (z. B. Pointer-Swap auf einen neuen unveränderlichen String). // Für diesen Benchmark wird direkte Zuweisung verwendet, mit Hinweis auf diese Einschränkung. node_to_update->value = value; return false; // Schlüssel aktualisiert } else { // Knoten mit Schlüssel gefunden, aber zur Löschung markiert. Als nicht gefunden behandeln und Einfügen versuchen. // Die remove-Operation sollte ihn letztlich bereinigen. } } // Pfad validieren: sicherstellen, dass Vorgänger noch gültig sind und auf ihre Nachfolger zeigen. // Diese Prüfung ist entscheidend, um das Einfügen in einen veralteten Pfad zu verhindern. bool valid_path = true; for (int level = 0; level <= new_level; ++level) { Node* pred = preds[level]; Node* succ = succs[level]; // Wenn pred zur Löschung markiert ist oder sich sein next-Zeiger geändert hat, ist der Pfad ungültig. 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; // Gesamten Einfügevorgang erneut versuchen } // Neuen Knoten erstellen Node* new_node = new Node(key, value, new_level); // new_node zuerst mit seinen Nachfolgern verknüpfen (von unten nach oben zur Konsistenz) for (int level = 0; level <= new_level; ++level) { new_node->next[level].store(succs[level], std::memory_order_release); } // Nun Vorgänger per CAS mit new_node verknüpfen (von unten nach oben) // Falls ein CAS fehlschlägt, hat ein anderer Thread die Liste verändert, also erneut versuchen. 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 fehlgeschlagen. Der new_node ist nicht vollständig verknüpft und nicht erreichbar. // Es ist sicher, ihn direkt zu löschen. delete new_node; goto retry_insert; // Zur gesamten Operation zurückspringen } } 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); // Version bei Schreibzugriff erhöhen return true; // Neuer Schlüssel eingefügt retry_insert:; // Label für goto } } // Löscht ein Schlüssel-Wert-Paar logisch. Gibt true zurück, wenn gefunden und entfernt. 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; // Schlüssel nicht gefunden } node_to_remove = succs[0]; // Kandidatenknoten zum Entfernen // Wenn der Knoten nicht vollständig verknüpft oder bereits zur Löschung markiert ist, erneut versuchen. if (!node_to_remove->fully_linked.load(std::memory_order_acquire) || node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { // Wenn bereits markiert, bearbeitet ihn ein anderer Thread oder hat es bereits getan. // Wenn nicht vollständig verknüpft, ist es ein unvollständiges Einfügen; dieses soll abgeschlossen oder bereinigt werden. // Der Einfachheit halber geben wir false zurück, wenn bereits markiert. if (node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { return false; } continue; // Erneut versuchen, falls noch nicht vollständig verknüpft } // Knoten atomar zur Löschung markieren. if (!node_to_remove->marked_for_deletion.compare_exchange_strong(false, true, std::memory_order_release, std::memory_order_acquire)) { // Ein anderer Thread hat ihn gleichzeitig markiert, erneut versuchen. continue; } // Knoten ist nun logisch gelöscht. Mit dem physischen Entketten fortfahren. // Pfad vor dem Entketten erneut validieren. bool valid_path = true; for (int level = 0; level <= node_to_remove->level; ++level) { Node* pred = preds[level]; // Wenn pred markiert ist oder sein next-Zeiger nicht mehr auf node_to_remove zeigt, ist der Pfad ungültig. 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; // Gesamten Löschvorgang erneut versuchen } // Von oben nach unten entketten (oder von unten nach oben, beides funktioniert für remove). // CAS-Operationen werden verwendet, um die next-Zeiger der Vorgänger zu aktualisieren. 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); // Versuche, pred's next-Zeiger zu aktualisieren. Falls das fehlschlägt, könnte ein anderer Thread // ihn bereits entkettet oder etwas eingefügt haben. Wir müssen den gesamten remove-Vorgang nicht // erneut versuchen, da der Knoten bereits zur Löschung markiert ist. 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); // Version bei Schreibzugriff erhöhen HazardPointerManager::retire_node(node_to_remove); // Knoten zur späteren Speicherfreigabe in den Ruhestand versetzen return true; } } // Gibt den dem Schlüssel zugeordneten Wert zurück oder signalisiert Abwesenheit. 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); } } // Nach der Traversierung sollte curr->next[0] der potenzielle Zielknoten sein. Node* target = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(2, target); // Ziel schützen 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; } // Gibt alle Schlüssel-Wert-Paare zurück, für die low <= key <= high gilt, als nach Schlüssel sortierte Liste. // Bietet Snapshot-Isolation mithilfe eines optimistischen Nebenläufigkeitskontrollmechanismus. 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); // Auf Ebene 0 beginnen HazardPointerManager::set_hazard_pointer(0, curr); while (curr != tail && curr->key <= high) { if (curr->key >= low) { // Prüfen, ob der Knoten an dieser Stelle gültig ist 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); // HP für den nächsten Knoten aktualisieren } HazardPointerManager::clear_hazard_pointer(0); long end_version = global_version.load(std::memory_order_acquire); if (start_version == end_version) { // Die Traversierung von Ebene 0 liefert von Natur aus sortierte Schlüssel. return result; } // Version hat sich geändert, gesamte Abfrage wiederholen, um einen konsistenten Snapshot zu erhalten. } } // Gibt die ungefähre Anzahl aktiver (nicht gelöschter) Elemente zurück. // Diese Anzahl ist ungefähr, da sie nicht mit einer globalen Sperre synchronisiert ist. 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; } }; // Initialisierung statischer Member für ConcurrentSkipList thread_local std::mt19937 ConcurrentSkipList::generator; thread_local std::uniform_real_distribution<double> ConcurrentSkipList::distribution(0.0, 1.0); // --- Test- / Demonstrationscode --- #include <thread> #include <vector> #include <numeric> #include <map> #include <set> #include <algorithm> #include <mutex> // Globaler Mutex für die Ausgabe, um verschachtelte Ausgaben aus mehreren Threads zu vermeiden 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) { // Einfügen 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) { // Entfernen 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) { // Suchen 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 { // Bereichsabfrage 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; // Für die Validierung könnten wir die Ergebnisse ausgeben oder speichern. // Der Kürze halber werden hier nicht alle Elemente ausgegeben. } } } } // Ein einfaches sequenzielles Referenzmodell zur Validierung der nebenläufigen Skip-Liste. // Dies ist nicht nebenläufig, sondern dient nur Korrektheitsprüfungen. 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); // Find-Operation validieren std::optional<std::string> concurrent_val = concurrent_list.find(key); std::optional<std::string> reference_val = reference_list.find(key); // Diese Validierung ist schwierig, weil die Referenzliste sequenziell ist und den // nebenläufigen Zustand zu keinem einzelnen Zeitpunkt widerspiegelt. Eine bessere Validierung // würde ein Modell einbeziehen, das über Linearizierungspunkte oder Snapshot-Isolation // argumentieren kann. // Für eine einfache Prüfung können wir sicherstellen, dass, falls die nebenläufige Liste einen Wert findet, // dieser konsistent damit ist, was *in* der Referenzliste sein *könnte*, oder falls nichts gefunden wird, // dass das ebenfalls konsistent ist. // Dies ist eine schwache Prüfung für eine nebenläufige Datenstruktur. // Eine stärkere Prüfung würde ein nebenläufiges Referenzmodell oder einen verlaufsbasierten Checker erfordern. // Im Moment prüfen wir nur auf Abstürze und grundlegende Konsistenz. // Wenn concurrent_val vorhanden ist, sollte reference_val idealerweise übereinstimmen, oder es bedeutet, // dass das Referenzmodell nicht synchron ist oder sich die nebenläufige Liste in einem transienten Zustand befindet. // Es geht hier mehr darum sicherzustellen, dass die nebenläufige Liste nicht abstürzt und *einen* gültigen Zustand zurückgibt. // Bereichsabfrage validieren (grundlegende Prüfung: keine Abstürze, Ergebnisse sind sortiert) 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); // Prüfen, ob range_result sortiert ist 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: Ergebnis der Bereichsabfrage nicht sortiert! Thread " << thread_id << std::endl; // exit(EXIT_FAILURE); // Für kritische Fehler } // Auf Phantom-Reads in Bereichsabfragen prüfen (ohne nebenläufige Referenz sehr schwer zu validieren) // Die optimistische Bereichsabfrage sollte Phantom-Reads verhindern, indem sie bei einem Schreibzugriff erneut versucht. // Dies wird implizit durch den Versionszählermechanismus getestet. } } 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); } // Eine einfache Referenzliste für grundlegende Plausibilitätsprüfungen. Kein vollständiger nebenläufiger Validator. ReferenceSkipList reference_list; // Referenzliste mit einigen Anfangsdaten füllen oder leer lassen. // Für eine echte Validierung bräuchte man ein nebenläufiges Modell oder einen transaktionalen Ansatz. // Einen oder zwei Validierungsthreads hinzufügen, um die Konsistenz kontinuierlich zu prüfen // Dies ist eine schwache Validierung, primär zur Erkennung von Abstürzen und grundlegenden Eigenschaften. 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(); } // Endprüfung von size und range_query nach allen Operationen 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--- Testergebnisse ---" << std::endl; std::cout << "Endgültige ungefähre Größe: " << final_size << std::endl; std::cout << "Endgültige Bereichsabfrage (alle Schlüssel) lieferte " << final_range.size() << " Elemente." << std::endl; // Einen finalen Scan und Bereinigung durchführen, um sicherzustellen, dass alle Retired-Knoten gelöscht werden. HazardPointerManager::scan_and_reclaim(); std::cout << "Hazard Pointer Manager-Bereinigung abgeschlossen." << std::endl; // Grundlegende Validierung: Wenn die Liste leer ist, sollte die Bereichsabfrage leer sein. // Dies ist keine starke Validierung der nebenläufigen Korrektheit, sondern der Grundfunktionalität. if (final_size == 0 && !final_range.empty()) { std::cerr << "ERROR: Liste ist leer, aber Bereichsabfrage lieferte Elemente!" << std::endl; } else if (final_size > 0 && final_range.empty()) { std::cerr << "ERROR: Liste hat Elemente, aber Bereichsabfrage lieferte leer!" << std::endl; } std::cout << "Test abgeschlossen. Konsole auf etwaige ERROR-Meldungen prüfen." << std::endl; return 0; } /* Analyseabschnitt: 1. Garantien bezüglich Linearizierbarkeit (oder Snapshot-Isolation): - insert(key, value), remove(key), find(key): Diese Operationen sind so konzipiert, dass sie linearizierbar sind. Eine erfolgreiche `insert`- oder `remove`-Operation scheint zu irgendeinem Zeitpunkt zwischen ihrem Aufruf und ihrer Antwort augenblicklich wirksam zu werden. Dies wird durch die Verwendung von Compare-And-Swap-(CAS)-Operationen auf `next`-Zeigern und atomaren Flags (`marked_for_deletion`, `fully_linked`) in Kombination mit Retry-Schleifen erreicht. Wenn ein CAS aufgrund einer nebenläufigen Modifikation fehlschlägt, versucht die Operation es erneut und findet dadurch effektiv einen neuen Linearizierungspunkt. `find`-Operationen geben einen Wert zurück, der irgendwann zwischen ihrem Aufruf und ihrer Antwort vorhanden war, und spiegeln eine linearizierbare Sicht auf den einzelnen Schlüssel wider. - range_query(low, high): Diese Operation bietet **Snapshot-Isolation**. Sie verwendet einen optimistischen Nebenläufigkeitskontrollmechanismus mit einem `global_version`-Zähler. Wenn eine `range_query` beginnt, speichert sie die `start_version`. Anschließend durchläuft sie die Skip-Liste und sammelt alle aktiven (nicht zur Löschung markierten, vollständig verknüpften) Elemente innerhalb des angegebenen Bereichs. Nach dem Sammeln speichert sie eine `end_version`. Wenn `start_version == end_version`, bedeutet das, dass während der Traversierung keine Schreiboperationen (insert/remove) stattgefunden haben und das gesammelte Ergebnis daher einen konsistenten Snapshot der Skip-Liste zu einem einzigen Zeitpunkt darstellt. Wenn sich die Versionen unterscheiden, bedeutet das, dass ein Schreibzugriff stattgefunden hat, und `range_query` wird erneut versucht, um einen frischen Snapshot zu erhalten. Dies garantiert, dass die zurückgegebene Liste keine Schlüssel enthält, die während der Ausführung der Operation nie gleichzeitig vorhanden waren. - size(): Die Operation `size()` liefert eine **ungefähre** Anzahl. Sie durchläuft die unterste Ebene und zählt aktive Knoten. Da sie keine globale Synchronisation verwendet, kann sich die Anzahl während der Traversierung ändern, sodass der zurückgegebene Wert möglicherweise nicht die exakte Größe zu einem einzelnen Zeitpunkt widerspiegelt. Dies stimmt mit der Anforderung des Prompts an eine

// Concurrent Skip List mit Bereichsabfragen (Snapshot) in Java // ------------------------------------------------------ // Sprachwahl: Java (GC vereinfacht die Speicherbereinigung im Vergleich zu C++/Rust). // Nebenläufigkeitsstrategie: // - Die Skip-List-Indexstruktur verwendet Lock-Free-Verknüpfung mit CAS auf nächsten Zeigern // und logische Löschung über Markierungsbits (AtomicMarkableReference). // - Aktualisierungen des Knotenswerts sind atomar über AtomicReference<String>. // - Physische Entfernung wird durch Traversierungen unterstützt (klassischer Harris-Michael-Stil). // - Bereichsabfragen liefern einen konsistenten Snapshot durch Kombination von: // (1) Einem globalen, Lock-Free "Sequence Lock" (Seqlock)-Zähler (AtomicLong). // Schreiber erhöhen ihn vor und nach strukturellen Änderungen (ungerade während des Schreibens, // gerade im Ruhezustand). // (2) Leser versuchen es erneut, wenn sie feststellen, dass sich der Zähler geändert hat oder ungerade ist. // Dies ergibt einen Snapshot, der einem bestimmten Zeitpunkt während des erfolgreichen Lesens entspricht. // Es vermeidet Phantome (Schlüssel, die während der Operation nie gleichzeitig vorhanden waren). // - Es wird keine einzelne globale Mux-Sperre verwendet; Seqlock ist nicht blockierend (nur bei Konflikten wird gesponnen). // // Operationen: // insert(key, value): Einfügen, wenn nicht vorhanden; andernfalls Wert atomar aktualisieren. // remove(key): Logisches Löschen, dann physisches Entkoppeln. // find(key): Wert zurückgeben, wenn vorhanden und nicht markiert. // range_query(low, high): Snapshot-konsistentes Ergebnis durch Seqlock-Wiederholung. // size(): Ungefähr (LongAdder wird bei erfolgreichem Einfügen/Entfernen gepflegt). // // Hinweise: // - Der Seqlock-Zähler ist eine globale atomare Variable, aber keine Sperre; Schreiber blockieren // sich nicht gegenseitig über eine Mux-Sperre; sie koordinieren sich nur durch Erhöhung eines Atomwerts. // Dies serialisiert Schreiber auf der Granularität der Sequenzerhöhungen, aber // strukturelle Änderungen sind immer noch CAS-basiert. // - In Java wird ABA gemildert, da Referenzen nicht als rohe Adressen // wiederverwendet werden, was CAS in der Praxis bricht; außerdem recyceln wir keine Knoten. // // Erstellen und Ausführen: // javac ConcurrentSkipListSnapshot.java && java ConcurrentSkipListSnapshot // // ------------------------------------------------------ // Analyse (kurz): // Linearität / Snapshot: // - insert/remove/find sind linearisierbar in Bezug auf die abstrakte Menge der Skip-Liste. // insert linearisiert bei erfolgreichem CAS-Linking auf Ebene 0 für einen neuen Knoten oder beim // AtomicReference-Set für die Aktualisierung. // remove linearisiert bei der logischen Markierung von next[0]. // - range_query liefert einen konsistenten Snapshot: Es gibt eine Menge von (Schlüssel, Wert)-Paaren zurück, // die einem bestimmten Zeitpunkt zwischen Anfang und Ende entsprechen, sichergestellt durch die Seqlock- // Prüfung (gerade Sequenz unverändert). Wenn gleichzeitige Schreiber auftreten, wird es wiederholt. // Komplexität: // - Erwartet O(log n) für find/insert/remove; range_query O(log n + k), wobei k Ergebnisse. // - size() O(1) ungefähr. // Einschränkungen: // - Bei starker Schreibkonflikten kann range_query aufgrund von Wiederholungen verhungern. // - Schreiber erhöhen eine globale Sequenz zweimal pro Operation; dies ist ein Skalierbarkeitsengpass // im Vergleich zu vollständig Lock-Free-Snapshot-Techniken. // - Physische Entfernung ist Best-Effort; lange Ketten gelöschter Knoten werden schließlich von Helfern bereinigt. // ------------------------------------------------------ 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 ist nur für Sentinel erlaubt @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: gerade = stabil; ungerade = Schreibvorgang in Bearbeitung. private final AtomicLong seq = new AtomicLong(0); // Ungefähre Größe 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; // geometrische Verteilung mit p=0.5 while (lvl < MAX_LEVEL - 1 && rnd.get().nextDouble() < P) { lvl++; } return lvl; } // Schreiberbereich Beginn/Ende. Keine Sperre; sorgt für Snapshot-Konsistenz für range_query. private void writerBegin() { seq.getAndIncrement(); // ungerade machen } private void writerEnd() { seq.getAndIncrement(); // gerade machen } // Vorgänger und Nachfolger finden; hilft beim Entkoppeln markierter Knoten. // Gibt true zurück, wenn der Schlüssel gefunden wurde (succs[0].key == target und auf Ebene 0 nicht markiert). private boolean findSplice(long key, Node[] preds, Node[] succs) { boolean[] marked = {false}; boolean snip; Node pred, curr, succ; retry: // Label für Wiederholungsversuch 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 ist markiert; Versuch des Entkoppelns 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]; // Wenn der Knoten logisch gelöscht ist, beim Entfernen helfen und erneut versuchen if (nodeFound.isMarked(0)) { continue; } // Wert atomar aktualisieren 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); } // Strukturelle Änderung: Bereichs-Snapshot mit Seqlock schützen writerBegin(); try { Node pred = preds[0]; Node succ = succs[0]; if (!pred.next[0].compareAndSet(succ, newNode, false, false)) { continue; // Erneut versuchen } // Höhere Ebenen verknüpfen 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]; // Höhere Ebenen zuerst markieren 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); } } // Jetzt Ebene 0 markieren (Linearisierungspunkt) Node succ0 = nodeToRemove.next[0].get(marked); while (true) { if (marked[0]) { // bereits entfernt return false; } writerBegin(); boolean iMarkedIt = false; try { iMarkedIt = nodeToRemove.next[0].compareAndSet(succ0, succ0, false, true); } finally { writerEnd(); } if (iMarkedIt) { // physische Entfernung unterstützen 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]) { // Entkopplung unterstützen 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) { // Schreiber in Bearbeitung Thread.onSpinWait(); continue; } ArrayList<Map.Entry<Long, String>> out = new ArrayList<>(); // Ersten Knoten >= low finden 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(); // Aktuellen Wert einschließen out.add(new AbstractMap.SimpleImmutableEntry<>(curr.key, v)); } curr = succ; } long s2 = seq.get(); if (s1 == s2 && (s2 & 1L) == 0L) { // konsistenter Snapshot return out; } // andernfalls erneut versuchen } } public long size() { return size.sum(); } // Zum Testen: Aktuelle Schlüssel sammeln (nicht-Snapshot) unter Seqlock zum Vergleichen. 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); // Referenz-Map zur Validierung mit einer gestreiften Sperre (nur Test-Harness) 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) { // einfügen/aktualisieren String v = "v" + id + "_" + i; boolean inserted = sl.insert(k, v); // Referenz aktualisieren ReentrantLock lk = locks[stripe(k)]; lk.lock(); try { boolean existed = ref.containsKey(k); ref.put(k, v); if (inserted == existed) { // inserted sollte true sein, wenn es nicht existierte throw new AssertionError("insert Rückgabefehler für Schlüssel " + k); } } finally { lk.unlock(); } } else if (op < 60) { // entfernen 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 Rückgabefehler für Schlüssel " + k); } } finally { lk.unlock(); } } else if (op < 85) { // finden String got = sl.find(k); ReentrantLock lk = locks[stripe(k)]; lk.lock(); try { String exp = ref.get(k); // find ist linearisierbar; direkt vergleichen if (!Objects.equals(got, exp)) { throw new AssertionError("find Fehler Schlüssel=" + k + " got=" + got + " exp=" + exp); } } finally { lk.unlock(); } } else { // Bereichsabfrage-Validierung: Vergleich mit einem konsistenten Snapshot der Referenz 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); // Referenz-Snapshot: Alle Streifen in Reihenfolge sperren 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("Bereichsgrößenfehler 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("Bereichsfehler bei " + 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); // Finale Validierung: Vollständigen Snapshot vergleichen List<Map.Entry<Long, String>> gotAll = sl.rangeQuery(Long.MIN_VALUE, Long.MAX_VALUE); // Erwartete Liste erstellen 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("Finale Größenfehler 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("Finaler Fehler bei " + i + " got=" + g + " exp=" + e); } } System.out.println("OK. Threads=" + THREADS + " ops/thread=" + OPS_PER_THREAD + " finalSize=" + sl.size()); } }