範囲クエリを備えたロックフリー並行スキップリストを実装する

# Hazard Pointer Manager これはメモリ再利用のためのハザードポインタマネージャです。 デモンストレーション目的の簡略化された実装です。 本番グレードのハザードポインタ実装は、より堅牢で、スレッドの登録/登録解除、より効率的なスキャン、およびエポックベースの再利用を使用したグローバルリタイアリストの利用などを扱います。 ```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 */

// Javaでの同時実行スキップリスト（範囲クエリ付き、スナップショット） // ------------------------------------------------------ // 言語選択：Java（GCによりメモリ解放がC++/Rustより容易）。 // 並行処理戦略： // - スキップリストのインデックス構造は、CAS（Compare-And-Swap）による次のポインタのロックフリーリンクと、 // マークビット（AtomicMarkableReference）による論理削除を使用。 // - ノードの値への更新はAtomicReference<String>によりアトミックに行われる。 // - 物理的な削除は、トラバーサル（古典的なHarris-Michaelスタイル）によって支援される。 // - 範囲クエリは、以下の組み合わせにより一貫したスナップショットを提供する： // (1) グローバルなロックフリー「シーケンスロック」（seqlock）カウンタ（AtomicLong）。 // ライターは、構造的な変更の前後にこれをインクリメントする（書き込み中は奇数、 // アイドル時は偶数）。 // (2) リーダーは、カウンタが変更されたり奇数になったりしたことを検出した場合、リトライする。 // これにより、操作中に同時に存在したことのないキー（ファントム）を回避し、 // ある瞬間に対応するスナップショットが得られる。 // - 単一のグローバルミューテックスは使用しない。seqlockはノンブロッキング（競合時のみスピニング）。 // // 操作： // insert(key, value): キーが存在しない場合は挿入、存在する場合は値をアトミックに更新。 // remove(key): 論理削除後、物理的にアンリンク。 // find(key): キーが存在し、マークされていない場合は値を返す。 // range_query(low, high): seqlockリトライによりスナップショットの一貫性を保証した結果を返す。 // size(): 近似値（成功した挿入/削除時にLongAdderで維持）。 // // 注意点： // - seqlockカウンタはグローバルなアトミック変数であり、ロックではない。ライターはミューテックス // によって互いにブロックしない。アトミックなインクリメントによってのみ調整される。 // これにより、シーケンスインクリメントの粒度でライターは直列化されるが、構造的な変更は // 依然としてCASベースである。 // - Javaでは、参照がCASを実質的に壊すような方法で生のメモリアドレスとして再利用されないため、 // ABA問題は軽減される。さらに、ノードを再利用しない。 // // ビルドと実行： // javac ConcurrentSkipListSnapshot.java && java ConcurrentSkipListSnapshot // // ------------------------------------------------------ // 解析（簡潔）： // 線形化可能性 / スナップショット： // - insert/remove/findは、スキップリストの抽象セットに関して線形化可能である。 // insertは、新しいノードに対するレベル0でのCASリンクの成功時、または値の更新時の // AtomicReferenceの設定時に線形化される。 // removeは、レベル0での論理マーク時に線形化される。 // - range_queryは一貫したスナップショットを提供する。これは、seqlockチェック（偶数のシーケンスが変更されていないこと） // によって保証される、開始から終了までのある瞬間に対応する(key,value)ペアのセットを返す。 // 並行ライターが発生した場合、リトライする。 // 複雑性： // - find/insert/removeの期待値はO(log n)。range_queryはO(log n + k)（kは結果の数）。 // - size()はO(1)の近似値。 // 制限事項： // - 重い書き込み競合下では、リトライによりrange_queryがスターベーションする可能性がある。 // - ライターは操作ごとにグローバルシーケンスを2回インクリメントする。これは、完全にロックフリーな // スナップショット技術と比較した場合のスケーラビリティのボトルネックとなる。 // - 物理削除はベストエフォート。削除されたノードの長い連鎖は、ヘルパーによって最終的にクリーンアップされる。 // ------------------------------------------------------ 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はセンチネルにのみ許可される @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); // スナップショット用シーケンスロック：偶数=安定、奇数=書き込み中。 private final AtomicLong seq = new AtomicLong(0); // 近似サイズ 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; // p=0.5の幾何分布 while (lvl < MAX_LEVEL - 1 && rnd.get().nextDouble() < P) { lvl++; } return lvl; } // 書き込みセクションの開始/終了。ロックではなく、range_queryのスナップショットの一貫性を提供する。 private void writerBegin() { seq.getAndIncrement(); // 奇数にする } private void writerEnd() { seq.getAndIncrement(); // 偶数にする } // 前者と後者を見つける。マークされたノードのアンリンクを支援する。 // キーが見つかった場合はtrueを返す（succs[0]のキーがターゲットであり、レベル0でマークされていない場合）。 private boolean findSplice(long key, Node[] preds, Node[] succs) { boolean[] marked = {false}; boolean snip; Node pred, curr, succ; retry: // ラベル付きcontinueのため 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はマークされている。アンリンクを試みる。 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 (nodeFound.isMarked(0)) { continue; } // 値をアトミックに更新する 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); } // 構造的な変更：範囲スナップショットをseqlockで保護する writerBegin(); try { Node pred = preds[0]; Node succ = succs[0]; if (!pred.next[0].compareAndSet(succ, newNode, false, false)) { continue; // リトライ } // 高位レベルをリンクする 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); // 競合が発生した場合、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]; // まず高位レベルをマークする 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); } } // 次にレベル0をマークする（線形化ポイント） Node succ0 = nodeToRemove.next[0].get(marked); while (true) { if (marked[0]) { // すでに削除されている return false; } writerBegin(); boolean iMarkedIt = false; try { iMarkedIt = nodeToRemove.next[0].compareAndSet(succ0, succ0, false, true); } finally { writerEnd(); } if (iMarkedIt) { // 物理削除を支援する findSplice(key, preds, succs); size.decrement(); return true; } else { succ0 = nodeToRemove.next[0].get(marked); // 再試行のためにsucc0を更新 } } } } 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]) { // アンリンクを支援する 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(); // CPUをビジーにしないように待機 continue; } ArrayList<Map.Entry<Long, String>> out = new ArrayList<>(); // 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) { // tailノードは除外 String v = curr.value.get(); // 現在の値を含める out.add(new AbstractMap.SimpleImmutableEntry<>(curr.key, v)); } curr = succ; } long s2 = seq.get(); if (s1 == s2 && (s2 & 1L) == 0L) { // シーケンスが変更されておらず、偶数（安定）の場合 // 一貫したスナップショット return out; } // それ以外の場合はリトライ } } public long size() { return size.sum(); } // テスト用：seqlock下で現在のキーを収集する（スナップショットではない） 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; } } // ------------------------- デモ / テスト ------------------------- 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); // 検証用の参照マップ（ストライプ化されたロックを使用、テストハーネスのみ） 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) { // 40%の確率で挿入/更新 String v = "v" + id + "_" + i; boolean inserted = sl.insert(k, v); // 参照マップを更新 ReentrantLock lk = locks[stripe(k)]; lk.lock(); try { boolean existed = ref.containsKey(k); ref.put(k, v); if (inserted == existed) { // 挿入はキーが存在しなかった場合にのみtrueになるはず throw new AssertionError("insert return mismatch for key " + k); } } finally { lk.unlock(); } } else if (op < 60) { // 20%の確率で削除 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) { // 削除はキーが存在した場合にのみtrueになるはず throw new AssertionError("remove return mismatch for key " + k); } } finally { lk.unlock(); } } else if (op < 85) { // 25%の確率で検索 String got = sl.find(k); ReentrantLock lk = locks[stripe(k)]; lk.lock(); try { String exp = ref.get(k); // findは線形化可能なので直接比較 if (!Objects.equals(got, exp)) { throw new AssertionError("find mismatch key=" + k + " got=" + got + " exp=" + exp); } } finally { lk.unlock(); } } else { // 15%の確率で範囲クエリ検証 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); // 参照マップのスナップショット：全ストライプを順序通りにロック 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); // 最終検証：完全なスナップショットを比較 List<Map.Entry<Long, String>> gotAll = sl.rangeQuery(Long.MIN_VALUE, Long.MAX_VALUE); // 期待値を構築 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()); } }