Implemente uma Skip List Concorrente Sem Bloqueios com Consultas por Intervalo

Resposta: ```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> // Declaração antecipada struct Node; // Gerenciador de Hazard Pointers para recuperação de memória // Esta é uma implementação simplificada para fins de demonstração. // Uma implementação de Hazard Pointer de nível de produção seria mais robusta, // lidando com registro/cancelamento de registro de threads, varredura mais eficiente, // e potencialmente usando uma lista global de nós aposentados com recuperação baseada em época. class HazardPointerManager { public: // Número máximo de hazard pointers que uma única thread pode manter simultaneamente. // Normalmente, 2-3 são suficientes para a maioria das estruturas de dados lock-free (ex.: pred, curr, next). static const int MAX_HAZARD_POINTERS_PER_THREAD = 3; // Número máximo de threads esperado para usar a skip list concorrentemente. // Isso limita o tamanho do nosso array estático hazard_records. static const int MAX_THREADS = 64; // Estrutura para armazenar hazard pointers de uma única 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); } } }; // Array global de registros de hazard, indexado por ID da thread. static std::vector<HazardRecord> hazard_records; // Contador para atribuição de IDs únicos às threads. static std::atomic<int> thread_id_counter; // Variável thread-local para armazenar o ID da thread atual. static thread_local int current_thread_id; // Lista thread-local de nós aposentados por esta thread. static thread_local std::vector<Node*> retired_nodes_local; // Inicializa o gerenciador de hazard pointers. Deve ser chamado uma vez na inicialização. static void init() { hazard_records.resize(MAX_THREADS); } // Obtém o ID único da thread atual, atribuindo um se ainda não tiver sido atribuído. 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: Número excessivo de threads registradas."); } } return current_thread_id; } // Define um hazard pointer para a thread atual. 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); } // Limpa um hazard pointer da thread atual. 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); } // Aposenta um nó, adicionando-o à lista local de nós aposentados da thread atual. // Dispara uma varredura e recuperação se a lista local crescer demais. static void retire_node(Node* node) { if (node) { retired_nodes_local.push_back(node); // Heurística: varrer quando o tamanho da lista local de nós aposentados exceder um limite. // Esse limite deve ser ajustado para desempenho. if (retired_nodes_local.size() >= MAX_HAZARD_POINTERS_PER_THREAD * thread_id_counter.load(std::memory_order_acquire) * 2) { scan_and_reclaim(); } } } // Varre todos os hazard pointers e recupera nós que não estão mais protegidos. static void scan_and_reclaim() { std::vector<Node*> to_reclaim; std::vector<Node*> new_retired_nodes_local; // Coleta todos os ponteiros atualmente protegidos de todas as threads ativas. 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); } } } // Itera pelos nós aposentados locais. for (Node* node : retired_nodes_local) { if (protected_pointers.count(node) == 0) { // O nó não está protegido por nenhum hazard pointer, é seguro deletá-lo. to_reclaim.push_back(node); } else { // Ainda protegido, mantê-lo para a próxima varredura. new_retired_nodes_local.push_back(node); } } retired_nodes_local = new_retired_nodes_local; // Deleta os nós seguros. for (Node* node : to_reclaim) { delete node; } } // Recupera quaisquer nós aposentados remanescentes quando uma thread termina ou no desligamento. static void cleanup_retired_nodes() { for (Node* node : retired_nodes_local) { delete node; } retired_nodes_local.clear(); } }; // Definições dos membros estáticos de HazardPointerManager std::vector<HazardPointerManager::HazardRecord> HazardPointerManager::hazard_records; std::atomic<int> HazardPointerManager::thread_id_counter(0); thread_local int HazardPointerManager::current_thread_id = -1; thread_local std::vector<Node*> HazardPointerManager::retired_nodes_local; // Estrutura de nó para a Skip List struct Node { int64_t key; std::string value; std::vector<std::atomic<Node*>> next; // Array de ponteiros para diferentes níveis std::atomic<bool> marked_for_deletion; // Flag de exclusão lógica std::atomic<bool> fully_linked; // Indica se o nó foi totalmente inserido int level; // Nível real deste nó // Construtor para um novo nó 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); // Inicializa os ponteiros next como nullptr } // O destrutor é chamado implicitamente quando a memória é recuperada por HazardPointerManager. // Nenhum destrutor explícito é necessário aqui para gerenciamento de memória, pois é tratado externamente. }; // Estrutura de Dados Skip List Concorrente class ConcurrentSkipList { private: Node* head; // Nó sentinela de cabeça (key = min_int) Node* tail; // Nó sentinela de cauda (key = max_int) int max_level; // Nível máximo possível para qualquer nó std::atomic<size_t> current_size; // Número aproximado de elementos ativos std::atomic<long> global_version; // Contador de versão para consultas de intervalo otimistas // Gerador de números aleatórios thread-local para geração de nível static thread_local std::mt19937 generator; static thread_local std::uniform_real_distribution<double> distribution; // Gera um nível aleatório para um novo nó usando uma distribuição geométrica (p=0.5) int random_level() { int level = 0; while (distribution(generator) < 0.5 && level < max_level) { level++; } return level; } // Função auxiliar para percorrer a skip list e encontrar predecessores e sucessores. // Preenche os arrays `preds` e `succs` com ponteiros em cada nível. // Retorna true se um nó com `key` for encontrado (independentemente de estar marcado). // Esta função usa hazard pointers para proteger `pred` e `curr` durante a travessia. 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); // Protege pred HazardPointerManager::set_hazard_pointer(1, curr); // Protege curr // Laço para avançar 'curr' até ultrapassar 'key' ou ser 'tail' while (curr != tail && curr->key < key) { pred = curr; HazardPointerManager::set_hazard_pointer(0, pred); // Protege o novo pred curr = pred->next[level].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(1, curr); // Protege o novo 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: // Construtor: inicializa os nós sentinela head e tail. ConcurrentSkipList(int max_lvl = 32) : max_level(max_lvl), current_size(0), global_version(0) { // Inicializa o gerador de números aleatórios thread-local generator.seed(std::chrono::high_resolution_clock::now().time_since_epoch().count() + std::hash<std::thread::id>{}(std::this_thread::get_id())); // Cria nós sentinela com chaves mín/máx head = new Node(std::numeric_limits<int64_t>::min(), "", max_level); tail = new Node(std::numeric_limits<int64_t>::max(), "", max_level); // Liga head a tail em todos os níveis 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); } // Destrutor: limpa todos os nós da 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; // Garante que quaisquer nós aposentados por esta thread sejam limpos. HazardPointerManager::cleanup_retired_nodes(); } // Insere um par chave-valor. Atualiza o valor se a chave existir. Retorna true se uma nova chave foi inserida. 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]; // Se encontrado e não marcado para exclusão, atualiza o valor. if (!node_to_update->marked_for_deletion.load(std::memory_order_acquire)) { // Atualiza o valor atomicamente. Para std::string, isso não é realmente lock-free // sem mecanismos mais complexos (ex.: troca de ponteiro para nova string imutável). // Para este benchmark, a atribuição direta é usada com uma observação sobre sua limitação. node_to_update->value = value; return false; // Chave atualizada } else { // Nó com a chave encontrado, mas marcado para exclusão. Tratar como não encontrado e tentar inserir. // A operação remove deve eventualmente limpá-lo. } } // Valida o caminho: garante que os predecessores ainda são válidos e apontam para seus sucessores. // Esta verificação é crucial para evitar inserção em um caminho obsoleto. bool valid_path = true; for (int level = 0; level <= new_level; ++level) { Node* pred = preds[level]; Node* succ = succs[level]; // Se pred estiver marcado para exclusão ou seu ponteiro next tiver mudado, o caminho é inválido. 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; // Tenta novamente todo o processo de inserção } // Cria novo nó Node* new_node = new Node(key, value, new_level); // Liga new_node aos seus sucessores primeiro (de baixo para cima por consistência) for (int level = 0; level <= new_level; ++level) { new_node->next[level].store(succs[level], std::memory_order_release); } // Agora liga predecessores a new_node usando CAS (de baixo para cima) // Se qualquer CAS falhar, outra thread modificou a lista, então tenta novamente. 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 falhou. O new_node não está totalmente ligado e não é alcançável. // É seguro deletá-lo diretamente. delete new_node; goto retry_insert; // Salta para tentar novamente toda a operação } } 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); // Incrementa versão em escrita return true; // Nova chave inserida retry_insert:; // Rótulo para goto } } // Exclui logicamente um par chave-valor. Retorna true se encontrado e removido. 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; // Chave não encontrada } node_to_remove = succs[0]; // Nó candidato à remoção // Se o nó não estiver totalmente ligado ou já estiver marcado para exclusão, tenta novamente. if (!node_to_remove->fully_linked.load(std::memory_order_acquire) || node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { // Se já estiver marcado, outra thread está lidando com ele ou já lidou. // Se não estiver totalmente ligado, é uma inserção incompleta, deixe completar ou ser limpa. // Por simplicidade, retornamos false se já estiver marcado. if (node_to_remove->marked_for_deletion.load(std::memory_order_acquire)) { return false; } continue; // Tenta novamente se ainda não estiver totalmente ligado } // Marca atomicamente o nó para exclusão. if (!node_to_remove->marked_for_deletion.compare_exchange_strong(false, true, std::memory_order_release, std::memory_order_acquire)) { // Outra thread o marcou concorrentemente, tenta novamente. continue; } // O nó agora está logicamente excluído. Prossegue para desvinculá-lo fisicamente. // Valida o caminho novamente antes de desvincular. bool valid_path = true; for (int level = 0; level <= node_to_remove->level; ++level) { Node* pred = preds[level]; // Se pred estiver marcado ou seu ponteiro next não apontar mais para node_to_remove, o caminho é inválido. 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; // Tenta novamente todo o processo de remoção } // Desvincula de cima para baixo (ou de baixo para cima, ambos funcionam para remoção). // Operações CAS são usadas para atualizar os ponteiros next dos predecessores. 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); // Tenta atualizar o ponteiro next de pred. Se falhar, outra thread pode // já tê-lo desvinculado ou inserido algo. Não precisamos tentar novamente toda a // remoção, pois o nó já está marcado para exclusão. 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); // Incrementa versão em escrita HazardPointerManager::retire_node(node_to_remove); // Aposenta o nó para recuperação return true; } } // Retorna o valor associado à chave, ou indica ausência. 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); } } // Após a travessia, curr->next[0] deve ser o possível nó-alvo. Node* target = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(2, target); // Protege 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; } // Retorna todos os pares chave-valor onde low <= key <= high, como uma lista ordenada por chave. // Fornece isolamento de snapshot usando um mecanismo de controle de concorrência otimista. 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); // Começa no nível 0 HazardPointerManager::set_hazard_pointer(0, curr); while (curr != tail && curr->key <= high) { if (curr->key >= low) { // Verifica se o nó é válido neste ponto 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); // Atualiza HP para o próximo nó } HazardPointerManager::clear_hazard_pointer(0); long end_version = global_version.load(std::memory_order_acquire); if (start_version == end_version) { // A travessia do nível 0 produz naturalmente chaves ordenadas. return result; } // A versão mudou, tente novamente toda a consulta para obter um snapshot consistente. } } // Retorna o número aproximado de elementos ativos (não excluídos). // Essa contagem é aproximada, pois não é sincronizada com um bloqueio global. size_t size() { size_t count = 0; Node* curr = head->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(0, curr); while (curr != tail) { if (!curr->marked_for_deletion.load(std::memory_order_acquire) && curr->fully_linked.load(std::memory_order_acquire)) { count++; } curr = curr->next[0].load(std::memory_order_acquire); HazardPointerManager::set_hazard_pointer(0, curr); } HazardPointerManager::clear_hazard_pointer(0); return count; } }; // Inicialização dos membros estáticos de ConcurrentSkipList thread_local std::mt19937 ConcurrentSkipList::generator; thread_local std::uniform_real_distribution<double> ConcurrentSkipList::distribution(0.0, 1.0); // --- Código de Teste / Demonstração --- #include <thread> #include <vector> #include <numeric> #include <map> #include <set> #include <algorithm> #include <mutex> // Mutex global para impressão, para evitar saída intercalada de múltiplas 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; // Para validação, talvez queiramos imprimir ou armazenar os resultados. // Por brevidade, não imprimimos todos os elementos aqui. } } } } // Um modelo de referência sequencial simples para validar a skip list concorrente. // Isto não é concorrente, apenas para verificações de corretude. 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); // Valida operação find std::optional<std::string> concurrent_val = concurrent_list.find(key); std::optional<std::string> reference_val = reference_list.find(key); // Esta validação é complicada porque a lista de referência é sequencial e não reflete // o estado concorrente em nenhum ponto único. Uma validação melhor envolveria um modelo // que pudesse raciocinar sobre pontos de linearizabilidade ou isolamento de snapshot. // Para uma verificação simples, podemos garantir que, se a lista concorrente encontrar um valor, // ele seja consistente com o que *poderia* estar na lista de referência, ou se não for encontrado, seja consistente. // Esta é uma verificação fraca para uma estrutura de dados concorrente. // Uma verificação mais forte exigiria um modelo de referência concorrente ou um verificador baseado em histórico. // Por enquanto, apenas verificaremos se não há falhas e a consistência básica. // Se concurrent_val estiver presente, reference_val idealmente deveria corresponder, ou isso significa // que o modelo de referência está fora de sincronia ou que a lista concorrente está em um estado transitório. // Isto é mais para garantir que a lista concorrente não falhe e retorne *algum* estado válido. // Valida range query (verificação básica: sem falhas, resultados ordenados) int64_t low = key_dist(rng); int64_t high = key_dist(rng); if (low > high) std::swap(low, high); std::vector<std::pair<int64_t, std::string>> range_result = concurrent_list.range_query(low, high); // Verifica se range_result está ordenado bool is_sorted = true; for (size_t j = 0; j + 1 < range_result.size(); ++j) { if (range_result[j].first > range_result[j+1].first) { is_sorted = false; break; } } if (!is_sorted) { std::lock_guard<std::mutex> lock(cout_mutex); std::cerr << "ERROR: Resultado de consulta de intervalo não está ordenado! Thread " << thread_id << std::endl; // exit(EXIT_FAILURE); // Para erros críticos } // Verifica leituras fantasmas em consultas de intervalo (muito difícil de validar sem uma referência concorrente) // A consulta de intervalo otimista deve prevenir leituras fantasmas tentando novamente se ocorrer uma escrita. // Isso é testado implicitamente pelo mecanismo do contador de versão. } } 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); } // Uma lista de referência simples para verificações básicas de sanidade. Não é um validador concorrente completo. ReferenceSkipList reference_list; // Preencha a lista de referência com alguns dados iniciais, ou deixe-a vazia. // Para uma validação real, você precisaria de um modelo concorrente ou de uma abordagem transacional. // Adiciona uma ou duas threads de validação para verificar continuamente a consistência // Esta é uma validação fraca, principalmente para detecção de falhas e verificações de propriedades básicas. std::vector<std::thread> validators; const int NUM_VALIDATORS = 2; const int CHECKS_PER_VALIDATOR = 5000; for (int i = 0; i < NUM_VALIDATORS; ++i) { validators.emplace_back(validation_thread, std::ref(skip_list), std::ref(reference_list), NUM_THREADS + i, CHECKS_PER_VALIDATOR, KEY_RANGE_START, KEY_RANGE_END); } for (auto& t : workers) { t.join(); } for (auto& t : validators) { t.join(); } // Verificação final de size e range query após todas as operações size_t final_size = skip_list.size(); std::vector<std::pair<int64_t, std::string>> final_range = skip_list.range_query(KEY_RANGE_START, KEY_RANGE_END); std::lock_guard<std::mutex> lock(cout_mutex); std::cout << "\n--- Resultados do Teste ---" << std::endl; std::cout << "Tamanho final aproximado: " << final_size << std::endl; std::cout << "Consulta final de intervalo (todas as chaves) retornou " << final_range.size() << " elementos." << std::endl; // Realiza uma varredura e recuperação final para garantir que todos os nós aposentados sejam deletados. HazardPointerManager::scan_and_reclaim(); std::cout << "Limpeza do Hazard Pointer Manager concluída." << std::endl; // Validação básica: se a lista estiver vazia, range query deve estar vazia. // Esta não é uma validação forte para corretude concorrente, mas para funcionalidade básica. if (final_size == 0 && !final_range.empty()) { std::cerr << "ERROR: A lista está vazia, mas a consulta de intervalo retornou elementos!" << std::endl; } else if (final_size > 0 && final_range.empty()) { std::cerr << "ERROR: A lista tem elementos, mas a consulta de intervalo retornou vazia!" << std::endl; } std::cout << "Teste concluído. Verifique o console em busca de quaisquer mensagens ERROR." << std::endl; return 0; } /* Seção de Análise: 1. Garantias de Linearizabilidade (ou Isolamento de Snapshot): - insert(key, value), remove(key), find(key): Essas operações são projetadas para serem linearizáveis. Uma operação `insert` ou `remove` bem-sucedida parece ter efeito instantâneo em algum ponto entre sua invocação e sua resposta. Isso é alcançado pelo uso de operações Compare-And-Swap (CAS) nos ponteiros `next` e em flags atômicas (`marked_for_deletion`, `fully_linked`), combinadas com laços de repetição. Se um CAS falha devido a modificação concorrente, a operação tenta novamente, efetivamente encontrando um novo ponto de linearização. Operações `find` retornam um valor que esteve presente em algum ponto entre sua invocação e resposta, refletindo uma visão linearizável da chave individual. - range_query(low, high): Esta operação fornece **isolamento de snapshot**. Ela usa um mecanismo de controle de concorrência otimista envolvendo um contador `global_version`. Quando uma `range_query` começa, ela registra `start_version`. Em seguida, percorre a skip list, coletando todos os elementos ativos (não marcados para exclusão, totalmente ligados) dentro do intervalo especificado. Após a coleta, registra `end_version`. Se `start_version == end_version`, isso significa que nenhuma operação de escrita (insert/remove) ocorreu durante a travessia e, portanto, o resultado coletado representa um snapshot consistente da skip list em um único ponto no tempo. Se as versões diferirem, isso implica que ocorreu uma escrita, e a `range_query` tenta novamente para obter um novo snapshot. Isso garante que a lista retornada não inclua chaves que nunca estiveram simultaneamente presentes durante a execução da operação. - size(): A operação `size()` fornece uma contagem **aproximada**. Ela percorre o nível mais baixo e conta nós ativos. Como não usa sincronização global, a contagem pode mudar durante a travessia, o que significa que o valor retornado pode não refletir o tamanho exato em nenhum ponto único no tempo. Isso é consistente com a exigência do prompt de uma

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