Is my performance what i should expect?

[BolverBlitz]

I tested with 3.9 Million Keys (8KB - 12KB) strings

All servers where running on the same Proxmox VM: NUMA = 1, 1 CPU (E7-8880v4) with 256GB Memory
The clients where running on a smaler VM on the same server.
Nither the client nor server VM had any high CPU usage, less than 25% total, 56% peak on the server on a single core
The hardware server runs 4x E7-8880v4 with 1.5TB DDR4 2400mhz Memory

I´ve also used the memtier_benchmark dragonfly uses in the readme, it did about 90K Keys/s but they are a lot smaler.
I do expect to be slower with my large keys, it avrages 2.5 - 3 gbit network traffic (10Gbit Nic)
Benchmark 1 uses Scan to find my keys, Benchmark 2 just selects the keys directly.

KeyDB: Default config, without snapshots
Dragonfly: ./dragonfly --logtostderr --requirepass=example --dbfilename= --cache_mode=true --bind 0.0.0.0 --port 6600
Garned: dotnet run -c Release -f net8.0 -- --password example --port 6500

I would realy like to understand what i did wrong, since nothing seems fully used up.

My Results:
Rading Benchmark 1:
KeyDB #1: 29992.28 Keys/Sec - 129 Seconds
KeyDB #2: 27210.58 Keys/Sec - 143 Seconds
Garnet #1: 30235.26 Keys/Sec - 71 Seconds (Only 2170297 Keys where read)
Garnet #2: 29251.03 Keys/Sec - 74 Seconds (Only 2170297 Keys where read)
Dragonfly #1: 27023.49 Keys/Sec - 144 Seconds
Dragonfly #2: 26491.59 Keys/Sec - 147 Seconds

Reading Benchmark 2:
KeyDB #1: 41565 Keys/Sec - 1m34s
KeyDB #2: 42085 Keys/Sec - 1m32s
Dragonfly #1: 40778 Keys/Sec - 1m35s
Dragonfly #2: 41848 Keys/Sec - 1m32s

Insertion: (This is most likely caped by the database i get my data from)
KeyDB #1: 14164.97 Keys/Sec - 4m34s
Garnet #2: 13947.97 Keys/Sec - 4m39s
Garnet #2: 13625.38 Keys/Sec - 4m45s
Dragonfly #1: 13973.42 Keys/Sec - 4m38s

Benchmark 1 Code

package main

import (
	"context"
	"log"
	"os"
	"sync"
	"time"

	"github.com/joho/godotenv"
	"github.com/redis/go-redis/v9"
)

var totalKeysProcessed int64
var totalKeysMutex sync.Mutex

func fetchAndProcessKeys(rdb *redis.Client, batchSize int, pattern string) {
	var cursor uint64
	var wg sync.WaitGroup
	ctx := context.Background()

	for {
		// Fetch a batch of keys
		keys, nextCursor, err := rdb.Scan(ctx, cursor, pattern, int64(batchSize)).Result()
		if err != nil {
			log.Fatalf("Error fetching keys: %v", err)
		}

		if len(keys) > 0 {
			wg.Add(1)
			go func(batchKeys []string) {
				defer wg.Done()
				processBatch(rdb, batchKeys)
			}(keys)
		}

		if nextCursor == 0 {
			break
		}
		cursor = nextCursor
	}

	wg.Wait()
	log.Println("All batches processed")
}

func processBatch(rdb *redis.Client, keys []string) {
	ctx := context.Background()
	startTime := time.Now()
	var wg sync.WaitGroup

	for _, key := range keys {
		wg.Add(1)
		go func(k string) {
			defer wg.Done()
			value, err := rdb.Get(ctx, k).Result()
			if err != nil {
				log.Printf("Error fetching value for key %s: %v", k, err)
				return
			}

			_ = value // GEO DO SOME PROCESSING HERE
		}(key)
	}

	wg.Wait()
	duration := time.Since(startTime).Seconds()

	// Update the total keys processed count after processing the batch
	totalKeysMutex.Lock()
	totalKeysProcessed += int64(len(keys))
	totalKeysMutex.Unlock()

	log.Printf("Processed %d keys in %.2f seconds (%.2f keys/second)", len(keys), duration, float64(len(keys))/duration)
}

func main() {
	err := godotenv.Load()
	if err != nil {
		log.Fatalf("Error loading .env file: %v", err)
	}

	redisAddr := os.Getenv("REDIS_ADDR")
	redisPassword := os.Getenv("REDIS_PASSWORD")
	disableTotalKeys := os.Getenv("DISABLE_REDIS_TOTALKEYS") == "true"

	ctx := context.Background()

	log.Println("Connecting to Redis...")
	rdb := redis.NewClient(&redis.Options{
		Addr:         redisAddr,
		Password:     redisPassword,
		DialTimeout:  10 * time.Second,
		ReadTimeout:  30 * time.Second,
		WriteTimeout: 30 * time.Second,
	})
	defer rdb.Close()

	_, err = rdb.Ping(ctx).Result()
	if err != nil {
		log.Fatalf("Failed to connect to Redis: %v", err)
	}
	log.Println("Connected to Redis.")

	// Conditionally get the total number of keys in Redis
	var totalKeys int64
	if !disableTotalKeys {
		totalKeys, err = rdb.DBSize(ctx).Result()
		if err != nil {
			log.Fatalf("Failed to get total keys in Redis: %v", err)
		}
	}

	batchSize := 1000
	keyPattern := "JP1:*"

	log.Printf("Starting to process keys with pattern: %s", keyPattern)
	startTime := time.Now()

	fetchAndProcessKeys(rdb, batchSize, keyPattern)

	// Calculate total processing time and average keys processed per second
	totalDuration := time.Since(startTime).Seconds()
	averageKeysPerSecond := float64(totalKeysProcessed) / totalDuration

	log.Printf("Total processing time: %.2f seconds", totalDuration)
	if !disableTotalKeys {
		log.Printf("Total keys processed: %d/%d", totalKeysProcessed, totalKeys)
	} else {
		log.Printf("Total keys processed: %d", totalKeysProcessed)
	}
	log.Printf("Average keys processed per second: %.2f", averageKeysPerSecond)
}

Benchmark 1 Code

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"strconv"
	"sync"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/joho/godotenv"
	"github.com/redis/go-redis/v9"
)

func main() {
	err := godotenv.Load()
	if err != nil {
		log.Fatalf("Error loading .env file: %v", err)
	}

	// Convert numGoroutines to int
	numGoroutines, err := strconv.Atoi(os.Getenv("NUM_GOROUTINES"))
	if err != nil {
		log.Fatalf("Invalid NUM_GOROUTINES value: %v", err)
	}

	postgresURL := os.Getenv("POSTGRES_URL")
	redisAddr := os.Getenv("REDIS_ADDR")
	redisPassword := os.Getenv("REDIS_PASSWORD")

	ctx := context.Background()

	// Connect to PostgreSQL
	log.Println("Connecting to PostgreSQL...")
	conn, err := pgx.Connect(ctx, postgresURL)
	if err != nil {
		log.Fatalf("Failed to connect to PostgreSQL: %v", err)
	}
	defer conn.Close(ctx)
	log.Println("Connected to PostgreSQL.")

	// Connect to Redis
	log.Println("Connecting to Redis...")
	rdb := redis.NewClient(&redis.Options{
		Addr:     redisAddr,
		Password: redisPassword,
	})
	defer rdb.Close()

	_, err = rdb.Ping(ctx).Result()
	if err != nil {
		log.Fatalf("Failed to connect to Redis: %v", err)
	}
	log.Println("Connected to Redis.")

	// Retrieve match IDs
	log.Println("Retrieving match IDs...")
	matchIDs, err := fetchMatchIDs(ctx, conn)
	if err != nil {
		log.Fatalf("Failed to fetch match IDs: %v", err)
	}
	log.Printf("Retrieved %d match IDs.", len(matchIDs))

	var wg sync.WaitGroup
	matchChan := make(chan int)

	// Track progress, speed, and metrics
	processed := 0
	iterationCount := 0
	speedWindow := []int{}
	windowSize := 60
	progressMutex := &sync.Mutex{}
	speedTicker := time.NewTicker(1 * time.Second)
	defer speedTicker.Stop()

	// Display progress and speed
	go func() {
		for range speedTicker.C {
			progressMutex.Lock()

			speedWindow = append(speedWindow, iterationCount)
			if len(speedWindow) > windowSize {
				speedWindow = speedWindow[1:]
			}

			averageSpeed := 0
			for _, speed := range speedWindow {
				averageSpeed += speed
			}
			if len(speedWindow) > 0 {
				averageSpeed /= len(speedWindow)
			}

			remainingMatches := len(matchIDs) - processed
			var timeRemaining string
			if averageSpeed > 0 {
				secondsRemaining := remainingMatches / averageSpeed
				timeRemaining = formatDuration(secondsRemaining)
			} else {
				timeRemaining = "Calculating..."
			}

			// Update progress display
			percentage := float64(processed) / float64(len(matchIDs)) * 100
			fmt.Printf("\rReading matches: [%-50s] %.2f%% (%d/%d) - %d reads/s (avg) - ETA: %s",
				createBar(percentage), percentage, processed, len(matchIDs), averageSpeed, timeRemaining)

			// Reset iteration count for the next second
			iterationCount = 0
			progressMutex.Unlock()
		}
	}()

	// Start timing
	startTime := time.Now()

	// Worker goroutines to read matches
	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for matchID := range matchChan {
				err := readMatch(ctx, rdb, matchID)

				if err != nil {
					log.Printf("Error reading match %d: %v", matchID, err)
				}

				// Update progress
				progressMutex.Lock()
				processed++
				iterationCount++
				progressMutex.Unlock()
			}
		}()
	}

	// Send match IDs to goroutines
	for _, matchID := range matchIDs {
		matchChan <- matchID
	}
	close(matchChan)
	wg.Wait()

	totalTime := time.Since(startTime)
	averageReadsPerSecond := float64(len(matchIDs)) / totalTime.Seconds()

	fmt.Printf("\nAll matches read succesfully in %v!\n", totalTime)
	fmt.Printf("Average reads per second: %.2f\n", averageReadsPerSecond)
}

// Fetch match IDs from PostgreSQL
func fetchMatchIDs(ctx context.Context, conn *pgx.Conn) ([]int, error) {
	rows, err := conn.Query(ctx, `SELECT id FROM database.my_table WHERE condition = TRUE`)
	if err != nil {
		return nil, fmt.Errorf("Error querying match IDs: %w", err)
	}
	defer rows.Close()

	var matchIDs []int
	for rows.Next() {
		var matchID int
		if err := rows.Scan(&matchID); err != nil {
			return nil, fmt.Errorf("Error scanning match ID: %w", err)
		}
		matchIDs = append(matchIDs, matchID)
	}
	return matchIDs, rows.Err()
}

// Read match data from Redis
func readMatch(ctx context.Context, rdb *redis.Client, matchID int) error {
	key := fmt.Sprintf("JP1:%d", matchID)
	_, err := rdb.Get(ctx, key).Result()
	if err != nil {
		return fmt.Errorf("Error reading match %d from Redis: %w", matchID, err)
	}
	return nil
}

// createBar creates a progress bar with the given percentage
func createBar(percentage float64) string {
	barWidth := 50
	progress := int(percentage / 2)
	bar := ""
	for i := 0; i < barWidth; i++ {
		if i < progress {
			bar += "="
		} else {
			bar += " "
		}
	}
	return bar
}

// formatDuration formats seconds as "mm:ss"
func formatDuration(seconds int) string {
	minutes := seconds / 60
	seconds = seconds % 60
	return fmt.Sprintf("%02d:%02d", minutes, seconds)
}