Merge pull request #4 from guycipher/iss-1-2-3

- minor race condition atomic lru adjustments.. issue #1 #2 #3

Author: guycipher

blockmanager/blockmanager.go

@@ -250,8 +250,9 @@ func (bm *BlockManager) backgroundSync() {
 
 	for {
 		select {
+
 		case <-ticker.C:
-			_ = syscall.Fdatasync(int(bm.fd)) // Use fdatasync for better performance
+			_ = Fdatasync(bm.fd)
 
 		case <-bm.closeChan:
 			return

blockmanager/fdatasync_windows.go

@@ -0,0 +1,31 @@
+//go:build windows
+// +build windows
+
+package blockmanager
+
+var (
+	modntdll                 = syscall.NewLazyDLL("ntdll.dll")
+	procNtFlushBuffersFileEx = modntdll.NewProc("NtFlushBuffersFileEx")
+)
+
+// Fdatasync__Win is a Windows-specific implementation of fdatasync.
+// https://learn.microsoft.com/en-us/windows-hardware/drivers/ddi/ntifs/nf-ntifs-ntflushbuffersfileex
+func Fdatasync(fd uintptr) error {
+
+	// Try to use NtFlushBuffersFileEx with FLUSH_FLAGS_FILE_DATA_SYNC_ONLY flag
+	status, _, err := procNtFlushBuffersFileEx.Call(
+		fd,
+		FLUSH_FLAGS_FILE_DATA_SYNC_ONLY,
+		0,
+		0,
+		0,
+	)
+
+	// Check for error (0 means success in Windows API)
+	if status != 0 {
+		// Fall back to regular FlushFileBuffers if NtFlushBuffersFileEx fails or isn't available
+		return syscall.FlushFileBuffers(syscall.Handle(fd))
+	}
+
+	return nil
+}

blockmanager/fsyncdata.go

@@ -0,0 +1,22 @@
+//go:build darwin || linux || freebsd || netbsd || openbsd
+
+package blockmanager
+
+import (
+	"runtime"
+	"syscall"
+)
+
+func Fdatasync(fd uintptr) error {
+	// On Darwin/macOS, Fdatasync is not available, so we fall back to Fsync..
+	if runtime.GOOS == "darwin" {
+		return syscall.Fsync(int(fd))
+	}
+
+	err := syscall.Fdatasync(int(fd))
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

compactor.go

@@ -436,8 +436,18 @@ func (compactor *Compactor) compactSSTables(sstables []*SSTable, sourceLevel, ta
 	}
 
 	// Add KLog and VLog managers to LRU cache
-	compactor.db.lru.Put(klogPath, klogBm)
-	compactor.db.lru.Put(vlogPath, vlogBm)
+	compactor.db.lru.Put(klogPath, klogBm, func(key, value interface{}) {
+		// Close the block manager when evicted from LRU
+		if bm, ok := value.(*blockmanager.BlockManager); ok {
+			_ = bm.Close()
+		}
+	})
+	compactor.db.lru.Put(vlogPath, vlogBm, func(key, value interface{}) {
+		// Close the block manager when evicted from LRU
+		if bm, ok := value.(*blockmanager.BlockManager); ok {
+			_ = bm.Close()
+		}
+	})
 
 	// Clean up the old SSTable files
 	for _, table := range sstables {
@@ -714,7 +724,12 @@ func newSSTCompactionIterator(sst *SSTable) *sstCompactionIterator {
 		if err != nil {
 			return &sstCompactionIterator{eof: true}
 		}
-		sst.db.lru.Put(klogPath, klogBm)
+		sst.db.lru.Put(klogPath, klogBm, func(key, value interface{}) {
+			// Close the block manager when evicted from LRU
+			if bm, ok := value.(*blockmanager.BlockManager); ok {
+				_ = bm.Close()
+			}
+		})
 	}
 
 	iter := &sstCompactionIterator{
@@ -794,7 +809,12 @@ func (iter *sstCompactionIterator) next() ([]byte, interface{}, int64, bool) {
 			iter.eof = true
 			return nil, nil, 0, false
 		}
-		iter.sstable.db.lru.Put(vlogPath, vlogBm)
+		iter.sstable.db.lru.Put(vlogPath, vlogBm, func(key, value interface{}) {
+			// Close the block manager when evicted from LRU
+			if bm, ok := value.(*blockmanager.BlockManager); ok {
+				_ = bm.Close()
+			}
+		})
 	}
 
 	// Read the value from VLog

db.go

@@ -347,7 +347,12 @@ func (db *DB) reinstate() error {
 		}
 
 		// Add the WAL to the LRU cache
-		db.lru.Put(newWalPath, walBm)
+		db.lru.Put(newWalPath, walBm, func(key, value interface{}) {
+			// Close the block manager when evicted from LRU
+			if bm, ok := value.(*blockmanager.BlockManager); ok {
+				_ = bm.Close()
+			}
+		})
 
 		// Initialize empty transactions slice
 		txns := make([]*Txn, 0)
@@ -381,7 +386,12 @@ func (db *DB) reinstate() error {
 		}
 
 		// Add WAL to LRU cache
-		db.lru.Put(walPath, walBm)
+		db.lru.Put(walPath, walBm, func(key, value interface{}) {
+			// Close the block manager when evicted from LRU
+			if bm, ok := value.(*blockmanager.BlockManager); ok {
+				_ = bm.Close()
+			}
+		})
 
 		// Process all transaction entries in this WAL
 		iter := walBm.Iterator()
@@ -542,7 +552,12 @@ func (db *DB) reinstate() error {
 	}
 
 	// Update or add to LRU cache
-	db.lru.Put(activeWALPath, activeWalBm)
+	db.lru.Put(activeWALPath, activeWalBm, func(key, value interface{}) {
+		// Close the block manager when evicted from LRU
+		if bm, ok := value.(*blockmanager.BlockManager); ok {
+			_ = bm.Close()
+		}
+	})
 
 	// Populate the active memtable with all committed transactions
 	populateMemtableFromTxns(activeMemt, globalTxnMap, false)

flusher.go

@@ -60,7 +60,12 @@ func (flusher *Flusher) queueMemtable() error {
 	}
 
 	// Add the new WAL to the LRU cache
-	flusher.db.lru.Put(newMemtable.wal.path, walBm)
+	flusher.db.lru.Put(newMemtable.wal.path, walBm, func(key, value interface{}) {
+		// Close the block manager when evicted from LRU
+		if bm, ok := value.(*blockmanager.BlockManager); ok {
+			_ = bm.Close()
+		}
+	})
 
 	// Push the current memtable to the immutable stack
 	flusher.immutable.Enqueue(flusher.db.memtable.Load().(*Memtable))
@@ -232,8 +237,18 @@ func (flusher *Flusher) flushMemtable(memt *Memtable) error {
 	flusher.db.log(fmt.Sprintf("Flushed %d entries to SSTable %d", entryCount, sstable.Id))
 
 	// Add both KLog and VLog to the LRU cache
-	flusher.db.lru.Put(klogPath, klogBm)
-	flusher.db.lru.Put(vlogPath, vlogBm)
+	flusher.db.lru.Put(klogPath, klogBm, func(key, value interface{}) {
+		// Close the block manager when evicted from LRU
+		if bm, ok := value.(*blockmanager.BlockManager); ok {
+			_ = bm.Close()
+		}
+	})
+	flusher.db.lru.Put(vlogPath, vlogBm, func(key, value interface{}) {
+		// Close the block manager when evicted from LRU
+		if bm, ok := value.(*blockmanager.BlockManager); ok {
+			_ = bm.Close()
+		}
+	})
 
 	// Add the SSTable to level 1
 	levels := flusher.db.levels.Load()