relayminer performance troubleshooting

kvstore/badger/errors.go

@@ -35,4 +35,7 @@ var (
 	// ErrBadgerGettingStoreLength is returned when the badger store fails to
 	// get the length of the database
 	ErrBadgerGettingStoreLength = errors.New("unable to get database length")
+	// ErrBadgerUnableToCheckExistence is returned when the badger store fails to
+	// check if a key exists
+	ErrBadgerUnableToCheckExistence = errors.New("unable to check key existence")
 )

kvstore/badger/interface.go

@@ -42,7 +42,7 @@ type BadgerKVStore interface {
 	// Exists returns true if the key exists
 	Exists(key []byte) (bool, error)
 	// Len returns the number of key-value pairs in the store
-	Len() int
+	Len() (int, error)
 
 	// --- Data management ---
 

kvstore/badger/kvstore.go

@@ -3,12 +3,15 @@ package badger
 import (
 	"errors"
 	"io"
+	"time"
 
 	badgerv4 "github.com/dgraph-io/badger/v4"
+	badgerv4opts "github.com/dgraph-io/badger/v4/options"
 )
 
 const (
-	maxPendingWrites = 16 // used in backup restoration
+	maxPendingWrites  = 16 // used in backup restoration
+	gcIntervalMinutes = 5 * time.Minute
 )
 
 var _ BadgerKVStore = &badgerKVStore{}
@@ -31,7 +34,13 @@ func NewKVStore(path string) (BadgerKVStore, error) {
 	if err != nil {
 		return nil, errors.Join(ErrBadgerOpeningStore, err)
 	}
-	return &badgerKVStore{db: db}, nil
+
+	store := &badgerKVStore{db: db}
+
+	// Start value log GC in a separate goroutine
+	go store.runValueLogGC()
+
+	return store, nil
 }
 
 // Set sets/updates the value for a given key
@@ -45,6 +54,28 @@ func (store *badgerKVStore) Set(key, value []byte) error {
 	return nil
 }
 
+// TODO: Add/use BatchSet when multiple KV pairs need to be updated for performance benefits
+// BatchSet sets/updates multiple key-value pairs in a single transaction
+// func (store *badgerKVStore) BatchSet(keys, values [][]byte) error {
+// 	if len(keys) != len(values) {
+// 		return errors.New("mismatched number of keys and values")
+// 	}
+//
+// 	wb := store.db.NewWriteBatch()
+// 	defer wb.Cancel()
+//
+// 	for i := range keys {
+// 		if err := wb.Set(keys[i], values[i]); err != nil {
+// 			return errors.Join(ErrBadgerUnableToSetValue, err)
+// 		}
+// 	}
+//
+// 	if err := wb.Flush(); err != nil {
+// 		return errors.Join(ErrBadgerUnableToSetValue, err)
+// 	}
+// 	return nil
+// }
+
 // Get returns the value for a given key
 func (store *badgerKVStore) Get(key []byte) ([]byte, error) {
 	var val []byte
@@ -111,11 +142,23 @@ func (store *badgerKVStore) GetAll(prefix []byte, descending bool) (keys, values
 
 // Exists checks whether the key exists in the store
 func (store *badgerKVStore) Exists(key []byte) (bool, error) {
-	val, err := store.Get(key)
+	var exists bool
+	err := store.db.View(func(tx *badgerv4.Txn) error {
+		_, err := tx.Get(key)
+		if err == badgerv4.ErrKeyNotFound {
+			exists = false
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		exists = true
+		return nil
+	})
 	if err != nil {
-		return false, err
+		return false, errors.Join(ErrBadgerUnableToCheckExistence, err)
 	}
-	return val != nil, nil
+	return exists, nil
 }
 
 // ClearAll deletes all key-value pairs in the store
@@ -159,9 +202,9 @@ func (store *badgerKVStore) Stop() error {
 }
 
 // Len gives the number of keys in the store
-func (store *badgerKVStore) Len() int {
-	count := 0
-	if err := store.db.View(func(tx *badgerv4.Txn) error {
+func (store *badgerKVStore) Len() (int, error) {
+	var count int
+	err := store.db.View(func(tx *badgerv4.Txn) error {
 		opt := badgerv4.DefaultIteratorOptions
 		opt.Prefix = []byte{}
 		opt.Reverse = false
@@ -171,10 +214,26 @@ func (store *badgerKVStore) Len() int {
 			count++
 		}
 		return nil
-	}); err != nil {
-		panic(errors.Join(ErrBadgerGettingStoreLength, err))
+	})
+	if err != nil {
+		return 0, errors.Join(ErrBadgerGettingStoreLength, err)
+	}
+	return count, nil
+}
+
+// runValueLogGC runs the value log garbage collection process periodically
+func (store *badgerKVStore) runValueLogGC() {
+	ticker := time.NewTicker(gcIntervalMinutes)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		err := store.db.RunValueLogGC(0.5)
+		if err != nil && err != badgerv4.ErrNoRewrite {
+			// Log the error, but don't stop the process
+			// We might want to use a proper logging mechanism here
+			println("Error during value log GC:", err)
+		}
 	}
-	return count
 }
 
 // PrefixEndBytes returns the end byteslice for a noninclusive range
@@ -194,7 +253,60 @@ func prefixEndBytes(prefix []byte) []byte {
 
 // badgerOptions returns the badger options for the store being created
 func badgerOptions(path string) badgerv4.Options {
+	// Parameters should be adjusted carefully, depending on the type of load. We need to experiment more to find the best
+	// values, and even then they might need further adjustments as the type of load/environment (e.g. memory dedicated
+	// to the process) changes.
+	//
+	// Good links to read about options:
+	// - https://github.com/dgraph-io/badger/issues/1304#issuecomment-630078745
+	// - https://github.com/dgraph-io/badger/blob/master/options.go#L37
+	// - https://github.com/open-policy-agent/opa/issues/4014#issuecomment-1003700744
 	opts := badgerv4.DefaultOptions(path)
-	opts.Logger = nil // disable badger's logger since it's very noisy
+
+	// Disable badger's logger since it's very noisy
+	opts.Logger = nil
+
+	// Reduce MemTableSize from default 64MB to 32MB
+	// This reduces memory usage but may increase disk I/O due to more frequent flushes
+	opts.MemTableSize = 32 << 20
+
+	// Decrease NumMemtables from default 5 to 3
+	// This reduces memory usage but may slow down writes if set too low
+	opts.NumMemtables = 3
+
+	// Lower ValueThreshold from default 1MB to 256 bytes
+	// This stores more data in LSM trees, reducing memory usage for small values
+	// but may impact write performance for larger values
+	opts.ValueThreshold = 256
+
+	// Reduce BlockCacheSize from default 256MB to 32MB
+	// This reduces memory usage but may slow down read operations
+	opts.BlockCacheSize = 32 << 20
+
+	// Adjust NumLevelZeroTables from default 5 to 3
+	// This triggers compaction more frequently, potentially reducing memory usage
+	// but at the cost of more disk I/O
+	opts.NumLevelZeroTables = 3
+
+	// Adjust NumLevelZeroTablesStall from default 15 to 8
+	// This also helps in triggering compaction more frequently
+	opts.NumLevelZeroTablesStall = 8
+
+	// Change Compression from default Snappy to ZSTD
+	// This can reduce memory and disk usage at the cost of higher CPU usage
+	opts.Compression = badgerv4opts.ZSTD
+
+	// Set ZSTDCompressionLevel to 3 (default is 1)
+	// This provides better compression but increases CPU usage
+	opts.ZSTDCompressionLevel = 3
+
+	// Reduce BaseTableSize from default 2MB to 1MB
+	// This might help in reducing memory usage, especially for many small keys
+	opts.BaseTableSize = 1 << 20
+
+	// Enable dynamic thresholding for value log (default is 0.0, which is disabled)
+	// This might help in optimizing memory usage for specific workloads
+	opts.VLogPercentile = 0.9
+
 	return opts
 }

kvstore/badger/kvstore_test.go

@@ -389,7 +389,9 @@ func TestBadger_KVStore_Len(t *testing.T) {
 
 	for _, tc := range tests {
 		require.NoError(t, store.Set(tc.key, tc.value))
-		require.Equal(t, tc.size, store.Len())
+		len, err := store.Len()
+		require.NoError(t, err)
+		require.Equal(t, tc.size, len)
 	}
 }
 

kvstore/interfaces.go

@@ -13,7 +13,7 @@ type MapStore interface {
 	// Delete removes a key
 	Delete(key []byte) error
 	// Len returns the number of key-value pairs in the store
-	Len() int
+	Len() (int, error)
 
 	// --- Debug ---
 

kvstore/simplemap/simplemap.go

@@ -63,8 +63,8 @@ func (sm *simpleMap) Delete(key []byte) error {
 }
 
 // Len returns the number of key-value pairs in the store.
-func (sm *simpleMap) Len() int {
-	return len(sm.m)
+func (sm *simpleMap) Len() (int, error) {
+	return len(sm.m), nil
 }
 
 // ClearAll clears all key-value pairs

kvstore/simplemap/simplemap_test.go

@@ -158,7 +158,9 @@ func TestSimpleMap_Len(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.desc, func(t *testing.T) {
 			tt.setup(store)
-			require.Equal(t, tt.expectedLen, store.Len())
+			len, err := store.Len()
+			require.NoError(t, err)
+			require.Equal(t, tt.expectedLen, len)
 		})
 	}
 }
@@ -172,6 +174,7 @@ func TestSimpleMap_ClearAll(t *testing.T) {
 
 	// Clear all elements
 	require.NoError(t, store.ClearAll())
-
-	require.Equal(t, 0, store.Len())
+	len, err := store.Len()
+	require.NoError(t, err)
+	require.Equal(t, 0, len)
 }

smst_test.go

@@ -348,7 +348,9 @@ func TestSMST_OrphanRemoval(t *testing.T) {
 
 	nodeCount := func(t *testing.T) int {
 		require.NoError(t, impl.Commit())
-		return smn.Len()
+		len, err := smn.Len()
+		require.NoError(t, err)
+		return len
 	}
 	setup := func() {
 		smn = simplemap.NewSimpleMap()

smt_test.go

@@ -316,7 +316,9 @@ func TestSMT_OrphanRemoval(t *testing.T) {
 
 	nodeCount := func(t *testing.T) int {
 		require.NoError(t, impl.Commit())
-		return smn.Len()
+		len, err := smn.Len()
+		require.NoError(t, err)
+		return len
 	}
 	setup := func() {
 		smn = simplemap.NewSimpleMap()