Flashring externalize

.vscode/launch.json

@@ -26,7 +26,7 @@
             "mode": "debug",
             "program": "${workspaceFolder}/flashring/cmd/flashringtest",
             "env": {
-                "PLAN": "readthrough-batched"
+                "PLAN": "readthrough"
             }
         }
 

flashring/cmd/flashringtest/plan_badger.go

@@ -10,7 +10,7 @@ import (
 	"strings"
 	"sync"
 
-	cachepkg "github.com/Meesho/BharatMLStack/flashring/internal/cache"
+	cachepkg "github.com/Meesho/BharatMLStack/flashring/pkg/cache"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )

flashring/cmd/flashringtest/plan_freecache.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 	"sync"
 
-	cachepkg "github.com/Meesho/BharatMLStack/flashring/internal/cache"
+	cachepkg "github.com/Meesho/BharatMLStack/flashring/pkg/cache"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )

flashring/cmd/flashringtest/plan_lockless.go

@@ -13,7 +13,7 @@ import (
 	"sync"
 	"time"
 
-	cachepkg "github.com/Meesho/BharatMLStack/flashring/internal/cache"
+	cachepkg "github.com/Meesho/BharatMLStack/flashring/pkg/cache"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )
@@ -35,13 +35,13 @@ func planLockless() {
 		cpuProfile         string
 	)
 
-	flag.StringVar(&mountPoint, "mount", "/media/a0d00kc/trishul/", "data directory for shard files")
-	flag.IntVar(&numShards, "shards", 500, "number of shards")
-	flag.IntVar(&keysPerShard, "keys-per-shard", 10_00_00, "keys per shard")
-	flag.IntVar(&memtableMB, "memtable-mb", 16, "memtable size in MiB")
+	flag.StringVar(&mountPoint, "mount", "/mnt/disks/nvme", "data directory for shard files")
+	flag.IntVar(&numShards, "shards", 100, "number of shards")
+	flag.IntVar(&keysPerShard, "keys-per-shard", 3_00_000, "keys per shard")
+	flag.IntVar(&memtableMB, "memtable-mb", 2, "memtable size in MiB")
 	flag.IntVar(&fileSizeMultiplier, "file-size-multiplier", 2, "file size in GiB per shard")
-	flag.IntVar(&readWorkers, "readers", 8, "number of read workers")
-	flag.IntVar(&writeWorkers, "writers", 8, "number of write workers")
+	flag.IntVar(&readWorkers, "readers", 16, "number of read workers")
+	flag.IntVar(&writeWorkers, "writers", 16, "number of write workers")
 	flag.IntVar(&sampleSecs, "sample-secs", 30, "predictor sampling window in seconds")
 	flag.Int64Var(&iterations, "iterations", 100_000_000, "number of iterations")
 	flag.Float64Var(&aVal, "a", 0.4, "a value for the predictor")
@@ -84,7 +84,7 @@ func planLockless() {
 	}
 
 	memtableSizeInBytes := int32(memtableMB) * 1024 * 1024
-	fileSizeInBytes := int64(fileSizeMultiplier) * int64(memtableSizeInBytes)
+	fileSizeInBytes := int64(fileSizeMultiplier) * 1024 * 1024 * 1024 // fileSizeMultiplier in GiB
 
 	cfg := cachepkg.WrapCacheConfig{
 		NumShards:             numShards,
@@ -95,21 +95,11 @@ func planLockless() {
 		GridSearchEpsilon:     0.0001,
 		SampleDuration:        time.Duration(sampleSecs) * time.Second,
 
-		// Pass the metrics collector to record cache metrics
-		MetricsRecorder: InitMetricsCollector(),
+		//lockless mode for PutLL/GetLL
+		EnableLockless: true,
 	}
 
-	// Set additional input parameters that the cache doesn't know about
-	metricsCollector.SetShards(numShards)
-	metricsCollector.SetKeysPerShard(keysPerShard)
-	metricsCollector.SetReadWorkers(readWorkers)
-	metricsCollector.SetWriteWorkers(writeWorkers)
-	metricsCollector.SetPlan("lockless")
-
-	// Start background goroutine to wait for shutdown signal and export CSV
-	go RunmetricsWaitForShutdown()
-
-	pc, err := cachepkg.NewWrapCache(cfg, mountPoint, logStats)
+	pc, err := cachepkg.NewWrapCache(cfg, mountPoint)
 	if err != nil {
 		panic(err)
 	}
@@ -121,7 +111,7 @@ func planLockless() {
 		missedKeyChanList[i] = make(chan int)
 	}
 
-	totalKeys := keysPerShard * numShards
+	totalKeys := 30_000_000
 	str1kb := strings.Repeat("a", 1024)
 	str1kb = "%d" + str1kb
 

flashring/cmd/flashringtest/plan_random_gausian.go

@@ -12,7 +12,7 @@ import (
 	"sync"
 	"time"
 
-	cachepkg "github.com/Meesho/BharatMLStack/flashring/internal/cache"
+	cachepkg "github.com/Meesho/BharatMLStack/flashring/pkg/cache"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )
@@ -83,7 +83,7 @@ func planRandomGaussian() {
 	}
 
 	memtableSizeInBytes := int32(memtableMB) * 1024 * 1024
-	fileSizeInBytes := int64(fileSizeMultiplier) * int64(memtableSizeInBytes)
+	fileSizeInBytes := int64(fileSizeMultiplier) * 1024 * 1024 * 1024 // fileSizeMultiplier in GiB
 
 	cfg := cachepkg.WrapCacheConfig{
 		NumShards:             numShards,
@@ -95,7 +95,7 @@ func planRandomGaussian() {
 		SampleDuration:        time.Duration(sampleSecs) * time.Second,
 	}
 
-	pc, err := cachepkg.NewWrapCache(cfg, mountPoint, logStats)
+	pc, err := cachepkg.NewWrapCache(cfg, mountPoint)
 	if err != nil {
 		panic(err)
 	}

flashring/cmd/flashringtest/plan_readthrough_gausian.go

@@ -13,7 +13,7 @@ import (
 	"sync"
 	"time"
 
-	cachepkg "github.com/Meesho/BharatMLStack/flashring/internal/cache"
+	cachepkg "github.com/Meesho/BharatMLStack/flashring/pkg/cache"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )
@@ -24,7 +24,7 @@ func planReadthroughGaussian() {
 		numShards          int
 		keysPerShard       int
 		memtableMB         int
-		fileSizeMultiplier int
+		fileSizeMultiplier float64
 		readWorkers        int
 		writeWorkers       int
 		sampleSecs         int
@@ -35,13 +35,13 @@ func planReadthroughGaussian() {
 		cpuProfile         string
 	)
 
-	flag.StringVar(&mountPoint, "mount", "/media/a0d00kc/trishul/", "data directory for shard files")
-	flag.IntVar(&numShards, "shards", 500, "number of shards")
-	flag.IntVar(&keysPerShard, "keys-per-shard", 4_00_00, "keys per shard")
-	flag.IntVar(&memtableMB, "memtable-mb", 16, "memtable size in MiB")
-	flag.IntVar(&fileSizeMultiplier, "file-size-multiplier", 2, "file size in GiB per shard")
-	flag.IntVar(&readWorkers, "readers", 8, "number of read workers")
-	flag.IntVar(&writeWorkers, "writers", 8, "number of write workers")
+	flag.StringVar(&mountPoint, "mount", "/mnt/disks/nvme/", "data directory for shard files")
+	flag.IntVar(&numShards, "shards", 50, "number of shards")
+	flag.IntVar(&keysPerShard, "keys-per-shard", 6_00_000, "keys per shard")
+	flag.IntVar(&memtableMB, "memtable-mb", 2, "memtable size in MiB")
+	flag.Float64Var(&fileSizeMultiplier, "file-size-multiplier", 0.25, "file size in GiB per shard")
+	flag.IntVar(&readWorkers, "readers", 16, "number of read workers")
+	flag.IntVar(&writeWorkers, "writers", 16, "number of write workers")
 	flag.IntVar(&sampleSecs, "sample-secs", 30, "predictor sampling window in seconds")
 	flag.Int64Var(&iterations, "iterations", 100_000_000, "number of iterations")
 	flag.Float64Var(&aVal, "a", 0.4, "a value for the predictor")
@@ -84,7 +84,7 @@ func planReadthroughGaussian() {
 	}
 
 	memtableSizeInBytes := int32(memtableMB) * 1024 * 1024
-	fileSizeInBytes := int64(fileSizeMultiplier) * int64(memtableSizeInBytes)
+	fileSizeInBytes := int64(float64(fileSizeMultiplier) * 1024 * 1024 * 1024) // fileSizeMultiplier in GiB
 
 	cfg := cachepkg.WrapCacheConfig{
 		NumShards:             numShards,
@@ -94,22 +94,9 @@ func planReadthroughGaussian() {
 		ReWriteScoreThreshold: 0.8,
 		GridSearchEpsilon:     0.0001,
 		SampleDuration:        time.Duration(sampleSecs) * time.Second,
-
-		// Pass the metrics collector to record cache metrics
-		MetricsRecorder: InitMetricsCollector(),
 	}
 
-	// Set additional input parameters that the cache doesn't know about
-	metricsCollector.SetShards(numShards)
-	metricsCollector.SetKeysPerShard(keysPerShard)
-	metricsCollector.SetReadWorkers(readWorkers)
-	metricsCollector.SetWriteWorkers(writeWorkers)
-	metricsCollector.SetPlan("readthrough")
-
-	// Start background goroutine to wait for shutdown signal and export CSV
-	go RunmetricsWaitForShutdown()
-
-	pc, err := cachepkg.NewWrapCache(cfg, mountPoint, logStats)
+	pc, err := cachepkg.NewWrapCache(cfg, mountPoint)
 	if err != nil {
 		panic(err)
 	}
@@ -121,7 +108,7 @@ func planReadthroughGaussian() {
 		missedKeyChanList[i] = make(chan int)
 	}
 
-	totalKeys := keysPerShard * numShards
+	totalKeys := 30_000_000
 	str1kb := strings.Repeat("a", 1024)
 	str1kb = "%d" + str1kb
 
@@ -139,7 +126,7 @@ func planReadthroughGaussian() {
 		key := fmt.Sprintf("key%d", k)
 		val := []byte(fmt.Sprintf(str1kb, k))
 		if err := pc.Put(key, val, 60); err != nil {
-			panic(err)
+			log.Error().Err(err).Msgf("error putting key %s", key)
 		}
 		if k%5000000 == 0 {
 			fmt.Printf("----------------------------------------------prepopulated %d keys\n", k)
@@ -158,7 +145,7 @@ func planReadthroughGaussian() {
 					key := fmt.Sprintf("key%d", mk)
 					val := []byte(fmt.Sprintf(str1kb, mk))
 					if err := pc.Put(key, val, 60); err != nil {
-						panic(err)
+						log.Error().Err(err).Msgf("error putting key %s", key)
 					}
 				}
 			}(w)
@@ -183,13 +170,14 @@ func planReadthroughGaussian() {
 					}
 
 					if expired {
-						panic("key expired")
+						log.Error().Msgf("key %s expired", key)
+						// panic("key expired")
 
 					}
 					if found && string(val) != fmt.Sprintf(str1kb, randomval) {
 						panic("value mismatch")
 					}
-					if k%5000000 == 0 {
+					if k%50000 == 0 {
 						fmt.Printf("----------------------------------------------read %d keys %d readerid\n", k, workerID)
 					}
 				}

flashring/cmd/flashringtest/plan_readthrough_gausian_batched.go

@@ -13,7 +13,7 @@ import (
 	"sync"
 	"time"
 
-	cachepkg "github.com/Meesho/BharatMLStack/flashring/internal/cache"
+	cachepkg "github.com/Meesho/BharatMLStack/flashring/pkg/cache"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )
@@ -93,7 +93,7 @@ func planReadthroughGaussianBatched() {
 	}
 
 	memtableSizeInBytes := int32(memtableMB) * 1024 * 1024
-	fileSizeInBytes := int64(fileSizeMultiplier) * int64(memtableSizeInBytes)
+	fileSizeInBytes := int64(fileSizeMultiplier) * 1024 * 1024 * 1024 // fileSizeMultiplier in GiB
 
 	cfg := cachepkg.WrapCacheConfig{
 		NumShards:             numShards,
@@ -108,22 +108,9 @@ func planReadthroughGaussianBatched() {
 		EnableBatching:    enableBatching,
 		BatchWindowMicros: batchWindowMicros,
 		MaxBatchSize:      maxBatchSize,
-
-		// Pass the metrics collector to record cache metrics
-		MetricsRecorder: InitMetricsCollector(),
 	}
 
-	// Set additional input parameters that the cache doesn't know about
-	metricsCollector.SetShards(numShards)
-	metricsCollector.SetKeysPerShard(keysPerShard)
-	metricsCollector.SetReadWorkers(readWorkers)
-	metricsCollector.SetWriteWorkers(writeWorkers)
-	metricsCollector.SetPlan("readthrough-batched")
-
-	// Start background goroutine to wait for shutdown signal and export CSV
-	go RunmetricsWaitForShutdown()
-
-	pc, err := cachepkg.NewWrapCache(cfg, mountPoint, logStats)
+	pc, err := cachepkg.NewWrapCache(cfg, mountPoint)
 	if err != nil {
 		panic(err)
 	}