Merge pull request #140 from wildcatdb/lru-new

I've written a simpler lru implementation for our use case.  No break…

Author: guycipher

c/wildcat_c.go

@@ -34,8 +34,6 @@ typedef struct {
     int level_count;
     int level_multiplier;
     int block_manager_lru_size;
-    double block_manager_lru_evict_ratio;
-    double block_manager_lru_access_weight;
     int permission;
     int bloom_filter;
     int max_compaction_concurrency;
@@ -126,8 +124,6 @@ func fromCOptions(copts *C.wildcat_opts_t) *wildcat.Options {
 		LevelCount:                           int(copts.level_count),
 		LevelMultiplier:                      int(copts.level_multiplier),
 		BlockManagerLRUSize:                  int(copts.block_manager_lru_size),
-		BlockManagerLRUEvictRatio:            float64(copts.block_manager_lru_evict_ratio),
-		BlockManagerLRUAccesWeight:           float64(copts.block_manager_lru_access_weight),
 		Permission:                           os.FileMode(copts.permission),
 		BloomFilter:                          copts.bloom_filter != 0,
 		MaxCompactionConcurrency:             int(copts.max_compaction_concurrency),

compactor.go

@@ -529,14 +529,14 @@ func (compactor *Compactor) compactSSTables(sstables []*SSTable, sourceLevelIdx,
 			if bm, ok := bm.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
 			}
-			compactor.db.lru.Delete(oldKlogPath)
+			compactor.db.lru.Remove(oldKlogPath)
 		}
 
 		if bm, ok := compactor.db.lru.Get(oldVlogPath); ok {
 			if bm, ok := bm.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
 			}
-			compactor.db.lru.Delete(oldVlogPath)
+			compactor.db.lru.Remove(oldVlogPath)
 		}
 
 		_ = os.Remove(oldKlogPath)
@@ -576,13 +576,13 @@ func (compactor *Compactor) compactSSTables(sstables []*SSTable, sourceLevelIdx,
 		return fmt.Errorf("failed to open final VLog block manager: %w", err)
 	}
 
-	compactor.db.lru.Put(klogFinalPath, klogBm, func(key, value interface{}) {
+	compactor.db.lru.Put(klogFinalPath, klogBm, func(key string, value interface{}) {
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
 		}
 	})
 
-	compactor.db.lru.Put(vlogFinalPath, vlogBm, func(key, value interface{}) {
+	compactor.db.lru.Put(vlogFinalPath, vlogBm, func(key string, value interface{}) {
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
 		}
@@ -1121,13 +1121,13 @@ func (compactor *Compactor) redistributeToLevel(tables []*SSTable, targetLevelId
 	targetLevel.sstables.Store(&newSSTables)
 	atomic.AddInt64(&targetLevel.currentSize, newSSTable.Size)
 
-	compactor.db.lru.Put(klogPath, klogBm, func(key, value interface{}) {
+	compactor.db.lru.Put(klogPath, klogBm, func(key string, value interface{}) {
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
 		}
 	})
 
-	compactor.db.lru.Put(vlogPath, vlogBm, func(key, value interface{}) {
+	compactor.db.lru.Put(vlogPath, vlogBm, func(key string, value interface{}) {
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
 		}
@@ -1181,14 +1181,14 @@ func (compactor *Compactor) removeSSTablesFromLevel(tablesToRemove []*SSTable, l
 			if bm, ok := bm.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
 			}
-			compactor.db.lru.Delete(klogPath)
+			compactor.db.lru.Remove(klogPath)
 		}
 
 		if bm, ok := compactor.db.lru.Get(vlogPath); ok {
 			if bm, ok := bm.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
 			}
-			compactor.db.lru.Delete(vlogPath)
+			compactor.db.lru.Remove(vlogPath)
 		}
 
 		_ = os.Remove(klogPath)
@@ -1244,7 +1244,7 @@ func getOrOpenBM(db *DB, path string) (*blockmanager.BlockManager, error) {
 		if err != nil {
 			return nil, fmt.Errorf("failed to open block manager: %w", err)
 		}
-		db.lru.Put(path, bm, func(key, value interface{}) {
+		db.lru.Put(path, bm, func(key string, value interface{}) {
 			if bm, ok := value.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
 			}

db.go

@@ -62,8 +62,6 @@ const (
 	DefaultLevelMultiplier = 10                   // Multiplier for the number of levels
 	// 64MB -> 640MB -> 6.4GB ->  64GB -> 640GB ->  6.4TB
 	DefaultBlockManagerLRUSize                  = 1024                 // Size of the LRU cache for block managers
-	DefaultBlockManagerLRUEvictRatio            = 0.20                 // Eviction ratio for the LRU cache
-	DefaultBlockManagerLRUAccessWeight          = 0.8                  // Access weight for the LRU cache
 	DefaultPermission                           = 0750                 // Default permission for created files
 	DefaultMaxCompactionConcurrency             = 4                    // Default max compaction concurrency
 	DefaultCompactionCooldownPeriod             = 5 * time.Second      // Default cooldown period for compaction
@@ -97,8 +95,6 @@ type Options struct {
 	LevelCount                           int           // Number of levels in the LSM tree
 	LevelMultiplier                      int           // Multiplier for the number of levels
 	BlockManagerLRUSize                  int           // Size of the LRU cache for block managers
-	BlockManagerLRUEvictRatio            float64       // Eviction ratio for the LRU cache
-	BlockManagerLRUAccesWeight           float64       // Access weight for the LRU cache
 	Permission                           os.FileMode   // Permission for created files
 	LogChannel                           chan string   // Channel for logging
 	STDOutLogging                        bool          // Enable logging to standard output (default is false and if set, channel is ignored)
@@ -178,12 +174,12 @@ func Open(opts *Options) (*DB, error) {
 	}
 
 	db := &DB{
-		lru:            lru.New(int64(opts.BlockManagerLRUSize), opts.BlockManagerLRUEvictRatio, opts.BlockManagerLRUAccesWeight), // New block manager LRU cache
-		wg:             &sync.WaitGroup{},                                                                                         // Wait group for background operations
-		opts:           opts,                                                                                                      // Set the options
-		txnBuffer:      buff,                                                                                                      // Create a new buffer for transactions with the specified cap
-		closeCh:        make(chan struct{}),                                                                                       // Channel for closing the database
-		txnTSGenerator: newIDGeneratorWithTimestamp(),                                                                             // We use timestamp generator for monotonic generation (1579134612000000004, next ID will be 1579134612000000005)
+		lru:            lru.New(opts.BlockManagerLRUSize), // New block manager LRU cache
+		wg:             &sync.WaitGroup{},                 // Wait group for background operations
+		opts:           opts,                              // Set the options
+		txnBuffer:      buff,                              // Create a new buffer for transactions with the specified cap
+		closeCh:        make(chan struct{}),               // Channel for closing the database
+		txnTSGenerator: newIDGeneratorWithTimestamp(),     // We use timestamp generator for monotonic generation (1579134612000000004, next ID will be 1579134612000000005)
 	}
 
 	// Initialize flusher and compactor
@@ -360,14 +356,6 @@ func (opts *Options) setDefaults() error {
 		opts.BlockManagerLRUSize = DefaultBlockManagerLRUSize
 	}
 
-	if opts.BlockManagerLRUEvictRatio <= 0 {
-		opts.BlockManagerLRUEvictRatio = DefaultBlockManagerLRUEvictRatio
-	}
-
-	if opts.BlockManagerLRUAccesWeight <= 0 {
-		opts.BlockManagerLRUAccesWeight = DefaultBlockManagerLRUAccessWeight
-	}
-
 	if opts.SSTableBTreeOrder <= 0 {
 		opts.SSTableBTreeOrder = DefaultSSTableBTreeOrder
 	} else {
@@ -433,7 +421,7 @@ func (db *DB) Close() error {
 	db.wg.Wait()
 
 	// Close open block managers
-	db.lru.ForEach(func(key, value interface{}, accessCount uint64) bool {
+	db.lru.ForEach(func(key string, value interface{}) bool {
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
 		}
@@ -501,7 +489,7 @@ func (db *DB) reinstate() error {
 		}
 
 		// Add the WAL to the LRU cache
-		db.lru.Put(newWalPath, walBm, func(key, value interface{}) {
+		db.lru.Put(newWalPath, walBm, func(key string, value interface{}) {
 			// Close the block manager when evicted from LRU
 			if bm, ok := value.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
@@ -536,7 +524,7 @@ func (db *DB) reinstate() error {
 		}
 
 		// Add WAL to LRU cache
-		db.lru.Put(walPath, walBm, func(key, value interface{}) {
+		db.lru.Put(walPath, walBm, func(key string, value interface{}) {
 			// Close the block manager when evicted from LRU
 			if bm, ok := value.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
@@ -698,7 +686,7 @@ func (db *DB) reinstate() error {
 	}
 
 	// Update or add to LRU cache
-	db.lru.Put(activeWALPath, activeWalBm, func(key, value interface{}) {
+	db.lru.Put(activeWALPath, activeWalBm, func(key string, value interface{}) {
 		// Close the block manager when evicted from LRU
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
@@ -1003,7 +991,6 @@ func (db *DB) Stats() string {
 				"Compaction Size Ratio", "Compaction Threshold", "Score Size Weight",
 				"Score Count Weight", "Flusher Interval", "Compactor Interval", "Bloom FPR",
 				"WAL Retry", "WAL Backoff", "SSTable B-Tree Order", "LRU Size",
-				"LRU Evict Ratio", "LRU Access Weight",
 				"File Version",
 				"Magic Number",
 				"Directory",
@@ -1014,8 +1001,7 @@ func (db *DB) Stats() string {
 				db.opts.CompactionSizeRatio, db.opts.CompactionSizeThreshold, db.opts.CompactionScoreSizeWeight,
 				db.opts.CompactionScoreCountWeight, db.opts.FlusherTickerInterval, db.opts.CompactorTickerInterval,
 				db.opts.BloomFilterFPR, db.opts.WalAppendRetry, db.opts.WalAppendBackoff,
-				db.opts.SSTableBTreeOrder, db.opts.BlockManagerLRUSize, db.opts.BlockManagerLRUEvictRatio,
-				db.opts.BlockManagerLRUAccesWeight,
+				db.opts.SSTableBTreeOrder, db.opts.BlockManagerLRUSize,
 				blockmanager.Version,
 				blockmanager.MagicNumber,
 				db.opts.Directory,

flusher.go

@@ -53,7 +53,7 @@ func (flusher *Flusher) queueMemtable() error {
 	}
 
 	// Add the new WAL to the LRU cache
-	flusher.db.lru.Put(newMemtable.wal.path, walBm, func(key, value interface{}) {
+	flusher.db.lru.Put(newMemtable.wal.path, walBm, func(key string, value interface{}) {
 		// Close the block manager when evicted from LRU
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
@@ -276,12 +276,12 @@ func (flusher *Flusher) flushMemtable(memt *Memtable) error {
 	}
 
 	// Add both KLog and VLog to the LRU cache
-	flusher.db.lru.Put(klogFinalPath, klogBm, func(key, value interface{}) {
+	flusher.db.lru.Put(klogFinalPath, klogBm, func(key string, value interface{}) {
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
 		}
 	})
-	flusher.db.lru.Put(vlogFinalPath, vlogBm, func(key, value interface{}) {
+	flusher.db.lru.Put(vlogFinalPath, vlogBm, func(key string, value interface{}) {
 		if bm, ok := value.(*blockmanager.BlockManager); ok {
 			_ = bm.Close()
 		}

go.mod

@@ -4,4 +4,7 @@ go 1.24.5
 
 require go.mongodb.org/mongo-driver v1.17.3
 
-require golang.org/x/sys v0.33.0
+require (
+	github.com/cespare/xxhash/v2 v2.3.0
+	golang.org/x/sys v0.33.0
+)

go.sum

@@ -1,3 +1,5 @@
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=

level.go

@@ -99,7 +99,7 @@ func (l *Level) reopen() error {
 		}
 
 		// Add the KLog to cache
-		l.db.lru.Put(klogPath, klogBm, func(key, value interface{}) {
+		l.db.lru.Put(klogPath, klogBm, func(key string, value interface{}) {
 			if bm, ok := value.(*blockmanager.BlockManager); ok {
 				_ = bm.Close()
 			}