Files
futriix/internal/storage/skiplist.go
T

706 lines
22 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/*
* Copyright 2026 Safronov Grigorii
*
* Licensed under the CDDL, Version 1.0 (the "License");
* you may not use this file except in compliance with the License.
*
* You may obtain a copy of the License at
* https://opensource.org/licenses/CDDL-1.0
*/
// Файл: internal/storage/skiplist.go
// Назначение: Lock-free skip list (список с пропусками) для инвертированного
// индекса коллекции. Заменяет sync.Map в Index.data.
//
// ИСПРАВЛЕНО (2026-10):
// - Баг #6: рассинхронизация size при воскрешении узла (Insert после Delete).
// Теперь проверяется wasDeleted и size увеличивается корректно.
// - Баг #7: дубликат на верхних уровнях при конкурентной вставке одного
// ключа с разными newLevel. Решение: двухфазная вставка — сначала
// уровень 0 (гарантированно единственный узел с таким ключом), затем
// верхние уровни (best-effort, дубликаты безопасны).
//
// КЛЮЧЕВОЙ ИНВАРИАНТ:
// - На уровне 0 не может быть двух узлов с одинаковым ключом.
// - На уровнях > 0 дубликаты допустимы (они лишь замедляют поиск, но не
// ломают корректность, потому что Search всегда спускается до уровня 0
// и там находит единственный узел с нужным ключом).
//
// ДРУГИЕ ИСПРАВЛЕНИЯ (2026):
// - Устранён data race при обновлении value существующего узла:
// value хранится в atomic.Value.
// - Исправлен обход Range: спуск на базовый уровень теперь корректный.
// - Исправлена вставка на уровнях выше текущего sl.level.
// - Идемпотентный Delete.
// - RangeWithKeyPrefix корректно останавливается при выходе за префикс.
package storage
import (
"fmt"
"math/rand"
"strings"
"sync"
"sync/atomic"
)
// =============================================================================
// КОНСТАНТЫ
// =============================================================================
const (
// Максимальное количество уровней в skip list.
maxSkipListLevel = 32
// Вероятность "подъёма" на следующий уровень.
skipListProbability = 0.5
// Порог, после которого Range запускает физическую очистку
// помеченных (tombstone) узлов.
skipListSweepThreshold = 1024
)
// =============================================================================
// ТИПЫ
// =============================================================================
// skipListNode представляет узел skip list.
type skipListNode struct {
// key — значение индексированного поля.
key interface{}
// value — ID документа. Хранится в atomic.Value для защиты от
// data race при обновлении существующего ключа.
// atomic.Value не может хранить nil строку напрямую, поэтому
// используется обёртка *string.
value atomic.Value // *string
// deleted — tombstone-флаг.
deleted atomic.Bool
// forward — массив атомарных указателей на следующий узел
// на каждом уровне.
forward []atomic.Value // *skipListNode
}
// loadValue атомарно загружает value.
func (n *skipListNode) loadValue() string {
val := n.value.Load()
if val == nil {
return ""
}
ptr, ok := val.(*string)
if !ok || ptr == nil {
return ""
}
return *ptr
}
// storeValue атомарно сохраняет value.
func (n *skipListNode) storeValue(v string) {
n.value.Store(&v)
}
// loadForward атомарно загружает указатель на следующий узел на уровне.
func (n *skipListNode) loadForward(level int) *skipListNode {
if level < 0 || level >= len(n.forward) {
return nil
}
val := n.forward[level].Load()
if val == nil {
return nil
}
ptr, ok := val.(*skipListNode)
if !ok {
return nil
}
return ptr
}
// storeForward атомарно сохраняет указатель на следующий узел на уровне.
func (n *skipListNode) storeForward(level int, next *skipListNode) {
if level < 0 || level >= len(n.forward) {
return
}
n.forward[level].Store(next)
}
// casForward атомарно заменяет указатель, если он равен old.
func (n *skipListNode) casForward(level int, old, new *skipListNode) bool {
if level < 0 || level >= len(n.forward) {
return false
}
return n.forward[level].CompareAndSwap(old, new)
}
// SkipList — lock-free список с пропусками.
type SkipList struct {
head *skipListNode
level atomic.Int32
size atomic.Int64
rngMu sync.Mutex
rng *rand.Rand
}
// NewSkipList создаёт новый пустой skip list.
func NewSkipList() *SkipList {
head := &skipListNode{
key: nil,
forward: make([]atomic.Value, maxSkipListLevel),
}
// Инициализируем forward-указатели типизированным nil.
for i := 0; i < maxSkipListLevel; i++ {
head.forward[i].Store((*skipListNode)(nil))
}
// Инициализируем value head-а, чтобы atomic.Value имел
// установленный concrete type.
emptyStr := ""
head.value.Store(&emptyStr)
sl := &SkipList{
head: head,
rng: rand.New(rand.NewSource(int64(0x5eed5eed))),
}
sl.level.Store(0)
sl.size.Store(0)
return sl
}
// randomLevel генерирует случайный уровень для нового узла.
func (sl *SkipList) randomLevel() int {
sl.rngMu.Lock()
defer sl.rngMu.Unlock()
level := 0
for level < maxSkipListLevel-1 && sl.rng.Float64() < skipListProbability {
level++
}
return level
}
// =============================================================================
// ПОИСК ПРЕДШЕСТВЕННИКОВ (вспомогательный метод)
// =============================================================================
// findPredecessors заполняет update[] указателями на узлы, после которых
// нужно вставлять/искать узел с заданным ключом на каждом уровне.
//
// Возвращает:
// - update[] — массив предшественников (длина maxSkipListLevel)
// - existing — узел на уровне 0 с ключом == key, либо nil
//
// Этот метод используется и в Insert, и в Delete.
func (sl *SkipList) findPredecessors(key interface{}) ([]*skipListNode, *skipListNode) {
update := make([]*skipListNode, maxSkipListLevel)
currentLevel := int(sl.level.Load())
// Спускаемся с верхнего уровня до базового.
x := sl.head
for i := currentLevel; i >= 0; i-- {
for {
next := x.loadForward(i)
if next == nil {
break
}
if next.deleted.Load() {
// Пытаемся физически вытолкнуть tombstone.
if x.casForward(i, next, next.loadForward(i)) {
continue
}
continue
}
if compareSkipListKeys(next.key, key) < 0 {
x = next
continue
}
break
}
update[i] = x
}
// Для уровней выше currentLevel предшественник — head.
// (Это может понадобиться, если новый уровень больше текущего.)
for i := currentLevel + 1; i < maxSkipListLevel; i++ {
update[i] = sl.head
}
// Ищем существующий узел на уровне 0.
existing := update[0].loadForward(0)
for existing != nil && compareSkipListKeys(existing.key, key) < 0 {
existing = existing.loadForward(0)
}
if existing != nil && compareSkipListKeys(existing.key, key) != 0 {
existing = nil
}
return update, existing
}
// =============================================================================
// ВСТАВКА (ИСПРАВЛЕНО: баги #6 и #7)
// =============================================================================
// Insert добавляет пару (key, value) в skip list.
// Если ключ уже существует, значение перезаписывается.
//
// РЕАЛИЗАЦИЯ (двухфазная вставка):
//
// Фаза 1: вставка на уровень 0.
// - Если ключ уже есть на уровне 0 — обновляем value (с корректной
// обработкой wasDeleted для size).
// - Иначе — CAS-вставка нового узла на уровень 0. Если CAS провалился
// из-за конкурента — retry.
// - После успешной вставки на уровень 0 ключ ГАРАНТИРОВАННО есть в списке
// (в единственном экземпляре на уровне 0).
//
// Фаза 2: вставка на верхние уровни (best-effort).
// - Для каждого уровня i от 1 до newLevel пытаемся CAS-вставить узел.
// - Если CAS провалился — не страшно: узел уже есть на уровне 0,
// а верхние уровни — только ускорители поиска. Дубликаты на верхних
// уровнях безопасны, потому что Search всегда спускается до уровня 0.
func (sl *SkipList) Insert(key interface{}, value string) {
if key == nil {
return
}
// Внешний цикл — retry при провале CAS на уровне 0.
for attempt := 0; attempt < 1000; attempt++ {
newLevel := sl.randomLevel()
// Поднимаем максимальный уровень, если нужно.
for {
currentLevel := int(sl.level.Load())
if newLevel <= currentLevel {
break
}
if sl.level.CompareAndSwap(int32(currentLevel), int32(newLevel)) {
break
}
}
// Находим предшественников и существующий узел.
update, existing := sl.findPredecessors(key)
// Случай 1: ключ уже есть на уровне 0.
// Обновляем value и, если узел был tombstone, воскрешаем его.
if existing != nil {
wasDeleted := existing.deleted.Load()
existing.storeValue(value)
existing.deleted.Store(false)
// ИСПРАВЛЕНО (баг #6): если узел был tombstone,
// увеличиваем size, потому что Delete его уже уменьшил.
if wasDeleted {
sl.size.Add(1)
}
return
}
// Случай 2: ключа нет — создаём новый узел и вставляем
// сначала на уровень 0.
newNode := &skipListNode{
key: key,
forward: make([]atomic.Value, newLevel+1),
}
for i := 0; i <= newLevel; i++ {
newNode.forward[i].Store((*skipListNode)(nil))
}
newNode.storeValue(value)
// Пытаемся вставить на уровень 0.
next0 := update[0].loadForward(0)
// Защита от ситуации, когда между findPredecessors и сейчас
// кто-то вставил узел с таким же ключом.
for next0 != nil && compareSkipListKeys(next0.key, key) < 0 {
next0 = next0.loadForward(0)
}
// Если между findPredecessors и сейчас появился узел с нашим
// ключом — обновляем его value и выходим.
if next0 != nil && compareSkipListKeys(next0.key, key) == 0 {
wasDeleted := next0.deleted.Load()
next0.storeValue(value)
next0.deleted.Store(false)
if wasDeleted {
sl.size.Add(1)
}
return
}
newNode.storeForward(0, next0)
if !update[0].casForward(0, next0, newNode) {
// CAS провалился — кто-то другой вставил узел между
// findPredecessors и сейчас. Retry.
continue
}
// Узел успешно вставлен на уровень 0. Теперь увеличиваем size
// (единственное место, где мы это делаем для новых узлов).
sl.size.Add(1)
// ============================================================
// ФАЗА 2: вставка на верхние уровни (best-effort).
// ============================================================
//
// Дубликаты на верхних уровнях безопасны: Search всегда спускается
// до уровня 0 и там находит единственный узел с нужным ключом.
// Если CAS на верхнем уровне провалился — просто пропускаем уровень.
for i := 1; i <= newLevel; i++ {
if update[i] == nil {
// Предшественник не определён — пропускаем уровень.
continue
}
// Перечитываем next на этом уровне (мог измениться).
next := update[i].loadForward(i)
newNode.storeForward(i, next)
// CAS best-effort: если провалился — не retry, просто пропускаем.
update[i].casForward(i, next, newNode)
}
return
}
// Если за 1000 попыток не удалось — что-то не так.
// Молча выходим (не паникуем).
}
// =============================================================================
// ПОИСК
// =============================================================================
// Search ищет значение по ключу.
func (sl *SkipList) Search(key interface{}) (string, bool) {
if key == nil {
return "", false
}
x := sl.head
for i := int(sl.level.Load()); i >= 0; i-- {
for {
next := x.loadForward(i)
if next == nil {
break
}
if next.deleted.Load() {
// Пропускаем tombstone.
if x.casForward(i, next, next.loadForward(i)) {
continue
}
continue
}
if compareSkipListKeys(next.key, key) < 0 {
x = next
continue
}
break
}
}
candidate := x.loadForward(0)
if candidate == nil {
return "", false
}
if compareSkipListKeys(candidate.key, key) == 0 {
if candidate.deleted.Load() {
return "", false
}
return candidate.loadValue(), true
}
return "", false
}
// SearchAll возвращает все значения, ключ которых равен заданному.
//
// ВАЖНО: текущая реализация индекса хранит только один docID на ключ,
// поэтому метод возвращает 0 или 1 значение. Для полноценной поддержки
// неуникальных индексов потребуется изменить структуру узла.
func (sl *SkipList) SearchAll(key interface{}) []string {
val, ok := sl.Search(key)
if !ok {
return nil
}
return []string{val}
}
// =============================================================================
// УДАЛЕНИЕ
// =============================================================================
// Delete помечает узел с заданным ключом как удалённый (tombstone).
// Идемпотентен: повторный вызов не уменьшает size.
func (sl *SkipList) Delete(key interface{}) {
if key == nil {
return
}
_, existing := sl.findPredecessors(key)
if existing == nil {
return
}
// CAS как атомарная проверка+установка.
// Уменьшаем size только если мы первыми поставили tombstone.
if existing.deleted.CompareAndSwap(false, true) {
sl.size.Add(-1)
}
}
// =============================================================================
// ОБХОД (RANGE)
// =============================================================================
// Range вызывает fn для каждой пары (key, value) в порядке возрастания ключей.
// Если fn возвращает false — обход прекращается.
func (sl *SkipList) Range(fn func(key interface{}, value string) bool) {
prev := sl.head
tombstoneCount := 0
for {
next := prev.loadForward(0)
if next == nil {
break
}
if next.deleted.Load() {
tombstoneCount++
// Пытаемся физически удалить узел из базового уровня.
if prev.casForward(0, next, next.loadForward(0)) {
if tombstoneCount >= skipListSweepThreshold {
sl.sweepUpperLevels()
tombstoneCount = 0
}
continue
}
// CAS не удался — двигаемся дальше.
prev = next
continue
}
if !fn(next.key, next.loadValue()) {
return
}
prev = next
}
}
// RangeWithKeyPrefix вызывает fn для каждой пары, ключ которой является
// строкой с заданным префиксом.
//
// Благодаря сортированности skip list, обход можно прервать, как только
// ключ перестал соответствовать префиксу.
func (sl *SkipList) RangeWithKeyPrefix(prefix string, fn func(key interface{}, value string) bool) {
sl.Range(func(key interface{}, value string) bool {
keyStr, ok := key.(string)
if !ok {
return true
}
if strings.HasPrefix(keyStr, prefix) {
return fn(key, value)
}
// Если ключ — строка и он больше префикса, дальнейшие ключи
// (отсортированные) тоже не подойдут.
if keyStr > prefix {
return false
}
return true
})
}
// sweepUpperLevels проходит по верхним уровням и физически выталкивает
// tombstone-узлы.
func (sl *SkipList) sweepUpperLevels() {
for i := 1; i < maxSkipListLevel; i++ {
x := sl.head
for {
next := x.loadForward(i)
if next == nil {
break
}
if next.deleted.Load() {
if x.casForward(i, next, next.loadForward(i)) {
continue
}
}
x = next
}
}
}
// =============================================================================
// СЛУЖЕБНЫЕ МЕТОДЫ
// =============================================================================
// Size возвращает приблизительное количество активных элементов.
func (sl *SkipList) Size() int64 { return sl.size.Load() }
// Len — синоним Size.
func (sl *SkipList) Len() int64 { return sl.size.Load() }
// IsEmpty возвращает true, если список пуст.
func (sl *SkipList) IsEmpty() bool { return sl.size.Load() == 0 }
// Clear помечает все узлы как удалённые.
func (sl *SkipList) Clear() {
sl.Range(func(key interface{}, value string) bool {
sl.Delete(key)
return true
})
sl.size.Store(0)
}
// =============================================================================
// СРАВНЕНИЕ КЛЮЧЕЙ
// =============================================================================
// compareSkipListKeys сравнивает два ключа.
func compareSkipListKeys(a, b interface{}) int {
if a == nil && b == nil {
return 0
}
if a == nil {
return -1
}
if b == nil {
return 1
}
switch av := a.(type) {
case string:
if bv, ok := b.(string); ok {
return strings.Compare(av, bv)
}
case int:
if bv, ok := toInt64(b); ok {
return compareInt64(int64(av), bv)
}
case int64:
if bv, ok := toInt64(b); ok {
return compareInt64(av, bv)
}
case float64:
if bv, ok := toFloat64Comparable(b); ok {
if av < bv {
return -1
}
if av > bv {
return 1
}
return 0
}
case bool:
if bv, ok := b.(bool); ok {
if av == bv {
return 0
}
if !av {
return -1
}
return 1
}
case []byte:
if bv, ok := b.([]byte); ok {
return strings.Compare(string(av), string(bv))
}
}
return strings.Compare(fmt.Sprintf("%v", a), fmt.Sprintf("%v", b))
}
func toInt64(v interface{}) (int64, bool) {
switch x := v.(type) {
case int:
return int64(x), true
case int32:
return int64(x), true
case int64:
return x, true
case uint:
return int64(x), true
case uint32:
return int64(x), true
case uint64:
return int64(x), true
case float64:
return int64(x), true
case float32:
return int64(x), true
default:
return 0, false
}
}
func toFloat64Comparable(v interface{}) (float64, bool) {
switch x := v.(type) {
case int:
return float64(x), true
case int32:
return float64(x), true
case int64:
return float64(x), true
case float32:
return float64(x), true
case float64:
return x, true
default:
return 0, false
}
}
func compareInt64(a, b int64) int {
if a < b {
return -1
}
if a > b {
return 1
}
return 0
}
// =============================================================================
// ITERATOR
// =============================================================================
// SkipListIterator — итератор по skip list.
type SkipListIterator struct {
sl *SkipList
current *skipListNode
started bool
}
// NewIterator создаёт новый итератор.
func (sl *SkipList) NewIterator() *SkipListIterator {
return &SkipListIterator{sl: sl}
}
// Next переходит к следующему элементу.
func (it *SkipListIterator) Next() (interface{}, string, bool) {
if it.sl == nil {
return nil, "", false
}
if !it.started {
it.started = true
it.current = it.sl.head
}
for {
next := it.current.loadForward(0)
if next == nil {
return nil, "", false
}
it.current = next
if !next.deleted.Load() {
return next.key, next.loadValue(), true
}
}
}
// Close освобождает ресурсы итератора.
func (it *SkipListIterator) Close() {
it.sl = nil
it.current = nil
}