Implement a Lock-Free Concurrent Skip List with Range Queries

```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> // Forward declaration struct Node; // Hazard Pointer Manager for memory reclamation // This is a simplified implementation for demonstration purposes. // A production-grade Hazard Pointer implementation would be more robust, // handling thread registration/deregistration, more efficient scanning, // and potentially using a global retired list with epoch-based reclamation. class HazardPointerManager { public: // Maximum number of hazard pointers a single thread can hold simultaneously. // Typically, 2-3 are sufficient for most lock-free data structures (e.g., pred, curr, next). static const int MAX_HAZARD_POINTERS_PER_THREAD = 3; // Maximum number of threads expected to use the skip list concurrently. // This limits the size of our static hazard_records array. static const int MAX_THREADS = 64; // Structure to hold hazard pointers for a single 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); } } }; // Global array of hazard records, indexed by thread ID. static std::vector<HazardRecord> hazard_records; // Counter for assigning unique thread IDs. static std::atomic<int> thread_id_counter; // Thread-local variable to store the current thread's ID. static thread_local int current_thread_id; // Thread-local list of nodes retired by this thread. static thread_local std::vector<Node*> retired_nodes_local; // Initialize the hazard pointer manager. Must be called once at startup. static void init() { hazard_records.resize(MAX_THREADS); } // Get the current thread's unique ID, assigning one if not already assigned. 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; } // Set a hazard pointer for the current 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); } // Clear a hazard pointer for the current 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); } // Retire a node, adding it to the current thread's local retired list. // Triggers a scan and reclaim if the local list grows too large. static void retire_node(Node* node) { if (node) { retired_nodes_local.push_back(node); // Heuristic: scan when local retired list size exceeds a threshold. // This threshold should be tuned for performance. if (retired_nodes_local.size() >= MAX_HAZARD_POINTERS_PER_THREAD * thread_id_counter.load(std::memory_order_acquire) * 2) { scan_and_reclaim(); } } } // Scan all hazard pointers and reclaim nodes that are no longer protected. static void scan_and_reclaim() { std::vector<Node*> to_reclaim; std::vector<Node*> new_retired_nodes_local; // Collect all currently protected pointers from all active 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); } } } // Iterate through local retired nodes. for (Node* node : retired_nodes_local) { if (protected_pointers.count(node) == 0) { // Node is not protected by any hazard pointer, safe to delete. to_reclaim.push_back(node); } else { // Still protected, keep it for the next scan. new_retired_nodes_local.push_back(node); } } retired_nodes_local = new_retired_nodes_local; // Delete the safe nodes. for (Node* node : to_reclaim) { delete node; } } // Reclaim any remaining retired nodes when a thread exits or at shutdown. static void cleanup_retired_nodes() { for (Node* node : retired_nodes_local) { delete node; } retired_nodes_local.clear(); } }; // Static member definitions for 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; // Node structure for the Skip List struct Node { int64_t key; std::string value; std::vector<std::atomic<Node*>> next; // Array of pointers for different levels std::atomic<bool> marked_for_deletion; // Logical deletion flag std::atomic<bool> fully_linked; // Indicates if node is fully inserted int level; // Actual level of this node // Constructor for a new node 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); // Initialize next pointers to nullptr } // Destructor is implicitly called when memory is reclaimed by HazardPointerManager. // No explicit destructor needed here for memory management, as it's handled externally. }; // Concurrent Skip List Data Structure class ConcurrentSkipList { private: Node* head; // Sentinel head node (key = min_int) Node* tail; // Sentinel tail node (key = max_int) int max_level; // Maximum possible level for any node std::atomic<size_t> current_size; // Approximate number of active elements std::atomic<long> global_version; // Version counter for optimistic range queries // Thread-local random number generator for level generation static thread_local std::mt19937 generator; static thread_local std::uniform_real_distribution<double> distribution; // Generates a random level for a new node using a geometric distribution (p=0.5) int random_level() { int level = 0; while (distribution(generator) < 0.5 && level < max_level) { level++; } return level; } // Helper function to traverse the skip list and find predecessors and successors. // Fills `preds` and `succs` arrays with pointers at each level. // Returns true if a node with `key` is found (regardless of its marked status). // This function uses hazard pointers to protect `pred` and `curr` during traversal. 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); // Protect pred HazardPointerManager::set_hazard_pointer(1, curr); // Protect curr // Loop to advance 'curr' until it's past 'key' or is 'tail' while (curr != tail && curr->key < key) { pred = curr; HazardPointerManager::set_hazard_pointer(0, pred); // Protect new pred curr = pred->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(1, curr); // Protect new 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: Initializes head and tail sentinel nodes. ConcurrentSkipList(int max_lvl = 32) : max_level(max_lvl), current_size(0), global_version(0) { // Initialize thread-local random number generator generator.seed(std::chrono::high_resolution_clock::now().time_since_epoch().count() + std::hash<std::thread::id>{}(std::this_thread::get_id())); // Create sentinel nodes with min/max keys head = new Node(std::numeric_limits<int64_t>::min(), "", max_level); tail = new Node(std::numeric_limits<int64_t>::max(), "", max_level); // Link head to tail at all levels 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: Cleans up all nodes in the 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; // Ensure any retired nodes by this thread are cleaned up. HazardPointerManager::cleanup_retired_nodes(); } // Inserts a key-value pair. Updates value if key exists. Returns true if new key inserted. 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]; // If found and not marked for deletion, update value. if (!node_to_update->marked_for_deletion.load(std::memory_order_acquire)) { // Atomically update the value. For std::string, this is not truly lock-free // without more complex mechanisms (e.g., pointer swap to new immutable string). // For this benchmark, direct assignment is used with a note on its limitation. node_to_update->value = value; return false; // Key updated } else { // Node with key found but marked for deletion. Treat as not found and try to insert. // The remove operation should eventually clean it up. } } // Validate path: ensure predecessors are still valid and point to their successors. // This check is crucial to prevent inserting into a stale path. bool valid_path = true; for (int level = 0; level <= new_level; ++level) { Node* pred = preds[level]; Node* succ = succs[level]; // If pred is marked for deletion or its next pointer changed, the path is invalid. 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; // Retry the entire insertion process } // Create new node Node* new_node = new Node(key, value, new_level); // Link new_node to its successors first (bottom-up for consistency) for (int level = 0; level <= new_level; ++level) { new_node->next[level].store(succs[level], std::memory_order_release); } // Now link predecessors to new_node using CAS (bottom-up) // If any CAS fails, another thread modified the list, so retry. 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 failed. The new_node is not fully linked and not reachable. // It's safe to delete it directly. delete new_node; goto retry_insert; // Jump to retry the entire operation } } 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); // Increment version on write return true; // New key inserted retry_insert:; // Label for goto } } // Logically deletes a key-value pair. Returns true if found and removed. 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; // Key not found } node_to_remove = succs[0]; // Candidate node for removal // If the node is not fully linked or already marked for deletion, retry. if (!node_to_remove->fully_linked.load(std::memory_order_acquire) || node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { // If already marked, another thread is handling it or already did. // If not fully linked, it's an incomplete insertion, let it complete or be cleaned up. // For simplicity, we return false if already marked. if (node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { return false; } continue; // Retry if not fully linked yet } // Atomically mark the node for deletion. if (!node_to_remove->marked_for_deletion.compare_exchange_strong(false, true, std::memory_order_release, std::memory_order_acquire)) { // Another thread marked it concurrently, retry. continue; } // Node is now logically deleted. Proceed to physically unlink it. // Validate path again before unlinking. bool valid_path = true; for (int level = 0; level <= node_to_remove->level; ++level) { Node* pred = preds[level]; // If pred is marked or its next pointer no longer points to node_to_remove, path is invalid. 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; // Retry the entire removal process } // Unlink from top-down (or bottom-up, both work for removal). // CAS operations are used to update predecessors' next pointers. 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); // Attempt to update pred's next pointer. If it fails, another thread might have // already unlinked it or inserted something. We don't need to retry the whole // remove, as the node is already marked for deletion. 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); // Increment version on write HazardPointerManager::retire_node(node_to_remove); // Retire the node for reclamation return true; } } // Returns the value associated with the key, or indicates 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); } } // After traversal, curr->next[0] should be the potential target node. Node* target = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(2, target); // Protect 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; } // Returns all key-value pairs where low <= key <= high, as a list sorted by key. // Provides snapshot isolation using an optimistic concurrency control mechanism. 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); // Start at level 0 HazardPointerManager::set_hazard_pointer(0, curr); while (curr != tail && curr->key <= high) { if (curr->key >= low) { // Check if node is valid at this point 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); // Update HP for next node } HazardPointerManager::clear_hazard_pointer(0); long end_version = global_version.load(std::memory_order_acquire); if (start_version == end_version) { // The level 0 traversal naturally yields sorted keys. return result; } // Version changed, retry the entire query to get a consistent snapshot. } } // Returns the approximate number of active (non-deleted) elements. // This count is approximate as it's not synchronized with a global lock. 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; } }; // Static member initialization for ConcurrentSkipList thread_local std::mt19937 ConcurrentSkipList::generator; thread_local std::uniform_real_distribution<double> ConcurrentSkipList::distribution(0.0, 1.0); // --- Test / Demonstration Code --- #include <thread> #include <vector> #include <numeric> #include <map> #include <set> #include <algorithm> #include <mutex> // Global mutex for printing to avoid interleaved output from multiple 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; // For validation, we might want to print or store the results. // For brevity, not printing all elements here. } } } } // A simple sequential reference model to validate the concurrent skip list. // This is not concurrent, just for correctness checks. 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); // Validate find operation std::optional<std::string> concurrent_val = concurrent_list.find(key); std::optional<std::string> reference_val = reference_list.find(key); // This validation is tricky because the reference list is sequential and doesn't reflect // the concurrent state at any single point. A better validation would involve a model // that can reason about linearizability points or snapshot isolation. // For a simple check, we can ensure that if the concurrent list finds a value, it's consistent // with what *could* be in the reference list, or if not found, it's consistent. // This is a weak check for a concurrent data structure. // A stronger check would require a concurrent reference model or a history-based checker. // For now, we'll just check for crashes and basic consistency. // If concurrent_val is present, reference_val should ideally match, or it means // the reference model is out of sync or the concurrent list is in a transient state. // This is more about ensuring the concurrent list doesn't crash and returns *some* valid state. // Validate range query (basic check: no crashes, results are sorted) 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); // Check if range_result is sorted 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); // For critical errors } // Check for phantom reads in range queries (very hard to validate without a concurrent reference) // The optimistic range query should prevent phantom reads by retrying if a write occurs. // This is implicitly tested by the version counter mechanism. } } 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); } // A simple reference list for basic sanity checks. Not a full concurrent validator. ReferenceSkipList reference_list; // Populate reference list with some initial data, or let it be empty. // For a real validation, you'd need a concurrent model or a transactional approach. // Add a validation thread or two to continuously check consistency // This is a weak validation, primarily for crash detection and basic property checks. 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(); } // Final check of size and range query after all operations 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--- Test Results ---" << std::endl; std::cout << "Final approximate size: " << final_size << std::endl; std::cout << "Final range query (all keys) returned " << final_range.size() << " elements." << std::endl; // Perform a final scan and reclaim to ensure all retired nodes are deleted. HazardPointerManager::scan_and_reclaim(); std::cout << "Hazard Pointer Manager cleanup complete." << std::endl; // Basic validation: if the list is empty, range query should be empty. // This is not a strong validation for concurrent correctness, but for basic functionality. 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 << "Test finished. Check console for any ERROR messages." << std::endl; return 0; } /* Analysis Section: 1. Linearizability (or Snapshot Isolation) Guarantees: - insert(key, value), remove(key), find(key): These operations are designed to be linearizable. A successful `insert` or `remove` operation appears to take effect instantaneously at some point between its invocation and response. This is achieved through the use of Compare-And-Swap (CAS) operations on `next` pointers and atomic flags (`marked_for_deletion`, `fully_linked`), combined with retry loops. If a CAS fails due to concurrent modification, the operation retries, effectively finding a new linearization point. `find` operations return a value that was present at some point between its invocation and response, reflecting a linearizable view of the individual key. - range_query(low, high): This operation provides **snapshot isolation**. It uses an optimistic concurrency control mechanism involving a `global_version` counter. When a `range_query` starts, it records the `start_version`. It then traverses the skip list, collecting all active (not marked for deletion, fully linked) elements within the specified range. After collection, it records an `end_version`. If `start_version == end_version`, it means no write operations (insert/remove) occurred during the traversal, and thus the collected result represents a consistent snapshot of the skip list at a single point in time. If the versions differ, it implies a write occurred, and the `range_query` retries to obtain a fresh snapshot. This guarantees that the returned list does not include keys that were never simultaneously present during the operation's execution. - size(): The `size()` operation provides an **approximate** count. It traverses the lowest level and counts active nodes. Since it does not use global synchronization, the count can change during traversal, meaning the returned value might not reflect the exact size at any single point in time. This is consistent with the prompt's requirement for an

// 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()); } }