Added CSC hits/evicts counter (#35)

* Included CSC docs with examples

* Added more context for CSC invalidation

* Added CSC hits/evicts counter

Author: filipecosta90

cli.go

@@ -32,12 +32,14 @@ func prepareCommandsDistribution(queries arrayStringParameters, cmds [][]string,
 	totalDifferentCommands = len(cmds)
 	var totalRateSum = 0.0
 	var err error
+	if len(benchmarkCommandsRatios) == 0 {
+		for i, _ := range queries {
+			cmdRates[i] = 1.0 / float64(len(queries))
+		}
+	}
 	for i, rawCmdString := range queries {
 		cmds[i], _ = shellwords.Parse(rawCmdString)
-		if i >= len(benchmarkCommandsRatios) {
-			cmdRates[i] = 1
-
-		} else {
+		if i < len(benchmarkCommandsRatios) {
 			cmdRates[i], err = strconv.ParseFloat(benchmarkCommandsRatios[i], 64)
 			if err != nil {
 				log.Fatalf("Error while converting query-rate param %s: %v", benchmarkCommandsRatios[i], err)

common.go

@@ -9,15 +9,20 @@ import (
 	"math"
 	"math/rand"
 	"strings"
+	"sync"
 )
 
 var totalCommands uint64
 var totalCached uint64
 var totalErrors uint64
+var totalCachedInvalidations uint64
 var latencies *hdrhistogram.Histogram
+var latenciesTick *hdrhistogram.Histogram
 var benchmarkCommands arrayStringParameters
 var benchmarkCommandsRatios arrayStringParameters
 
+var cscInvalidationMutex sync.Mutex
+
 const Inf = rate.Limit(math.MaxFloat64)
 const charset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 

redis-bechmark-go.go

@@ -157,13 +157,25 @@ func sendCmdLogicRadix(conn Client, newCmdS []string, enableMultiExec bool, key
 	datapointsChan <- datapoint{!(err != nil), duration.Microseconds(), cacheHit}
 }
 
+func onInvalidations(messages []rueidis.RedisMessage) {
+	if messages != nil {
+		cscInvalidationMutex.Lock()
+		for range messages {
+			totalCachedInvalidations++
+		}
+		cscInvalidationMutex.Unlock()
+	}
+
+}
+
 func main() {
 	host := flag.String("h", "127.0.0.1", "Server hostname.")
 	port := flag.Int("p", 12000, "Server port.")
 	rps := flag.Int64("rps", 0, "Max rps. If 0 no limit is applied and the DB is stressed up to maximum.")
 	rpsburst := flag.Int64("rps-burst", 0, "Max rps burst. If 0 the allowed burst will be the ammount of clients.")
 	username := flag.String("u", "", "Username for Redis Auth.")
 	password := flag.String("a", "", "Password for Redis Auth.")
+	jsonOutFile := flag.String("json-out-file", "", "Results file. If empty will not save.")
 	seed := flag.Int64("random-seed", 12345, "random seed to be used.")
 	clients := flag.Uint64("c", 50, "number of clients.")
 	keyspacelen := flag.Uint64("r", 1000000, "keyspace length. The benchmark will expand the string __key__ inside an argument with a number in the specified range from 0 to keyspacelen-1. The substitution changes every time a command is executed.")
@@ -248,6 +260,7 @@ func main() {
 	samplesPerClient := *numberRequests / *clients
 	client_update_tick := 1
 	latencies = hdrhistogram.New(1, 90000000, 3)
+	latenciesTick = hdrhistogram.New(1, 90000000, 3)
 	opts := radix.Dialer{}
 	if *password != "" {
 		opts.AuthPass = *password
@@ -310,6 +323,10 @@ func main() {
 		cmd := make([]string, len(args))
 		copy(cmd, args)
 		if *cscEnabled || *useRuedis {
+			var invalidationFunction func([]rueidis.RedisMessage) = nil
+			if *cscEnabled {
+				invalidationFunction = onInvalidations
+			}
 			clientOptions := rueidis.ClientOption{
 				InitAddress:         []string{connectionStr},
 				Username:            *username,
@@ -323,6 +340,7 @@ func main() {
 				ReadBufferEachConn:  1024,
 				WriteBufferEachConn: 1024,
 				CacheSizeEachConn:   *cscSizeBytes,
+				OnInvalidations:     invalidationFunction,
 			}
 			clientOptions.Dialer.KeepAlive = *clientKeepAlive
 			ruedisClient, err = rueidis.NewClient(clientOptions)
@@ -350,8 +368,10 @@ func main() {
 	signal.Notify(c, os.Interrupt)
 
 	tick := time.NewTicker(time.Duration(client_update_tick) * time.Second)
-	closed, _, duration, totalMessages, _ := updateCLI(tick, c, *numberRequests, *loop, datapointsChan)
-	messageRate := float64(totalMessages) / float64(duration.Seconds())
+	closed, startT, endT, duration, _, _, _, messageRateTs, cacheRateTs, cacheInvalidationsTs, percentilesTs := updateCLI(tick, c, *numberRequests, *loop, datapointsChan, int(*clients))
+	messageRate := float64(totalCommands) / float64(duration.Seconds())
+	CSCHitRate := float64(totalCached) / float64(duration.Seconds())
+	CSCInvalidationRate := float64(totalCachedInvalidations) / float64(duration.Seconds())
 	avgMs := float64(latencies.Mean()) / 1000.0
 	p50IngestionMs := float64(latencies.ValueAtQuantile(50.0)) / 1000.0
 	p95IngestionMs := float64(latencies.ValueAtQuantile(95.0)) / 1000.0
@@ -362,10 +382,22 @@ func main() {
 	fmt.Printf("Total Duration %.3f Seconds\n", duration.Seconds())
 	fmt.Printf("Total Errors %d\n", totalErrors)
 	fmt.Printf("Throughput summary: %.0f requests per second\n", messageRate)
+	fmt.Printf("                    %.0f CSC Hits per second\n", CSCHitRate)
+	fmt.Printf("                    %.0f CSC Evicts per second\n", CSCInvalidationRate)
 	fmt.Printf("Latency summary (msec):\n")
 	fmt.Printf("    %9s %9s %9s %9s\n", "avg", "p50", "p95", "p99")
 	fmt.Printf("    %9.3f %9.3f %9.3f %9.3f\n", avgMs, p50IngestionMs, p95IngestionMs, p99IngestionMs)
 
+	testResult := NewTestResult("", uint(*clients), 0)
+	testResult.FillDurationInfo(startT, endT, duration)
+	testResult.OverallClientLatencies = percentilesTs
+	testResult.OverallCommandRate = messageRateTs
+	testResult.OverallCSCHitRate = cacheRateTs
+	testResult.OverallCSCInvalidationRate = cacheInvalidationsTs
+	_, overallLatencies := generateLatenciesMap(latencies, duration)
+	testResult.Totals = overallLatencies
+	saveJsonResult(testResult, *jsonOutFile)
+
 	if closed {
 		return
 	}
@@ -378,21 +410,27 @@ func main() {
 	os.Exit(0)
 }
 
-func updateCLI(tick *time.Ticker, c chan os.Signal, message_limit uint64, loop bool, datapointsChan chan datapoint) (bool, time.Time, time.Duration, uint64, []float64) {
+func updateCLI(tick *time.Ticker, c chan os.Signal, message_limit uint64, loop bool, datapointsChan chan datapoint, totalClients int) (bool, time.Time, time.Time, time.Duration, uint64, uint64, uint64, []float64, []float64, []float64, []map[string]float64) {
 	var currentErr uint64 = 0
 	var currentCount uint64 = 0
 	var currentCachedCount uint64 = 0
 	start := time.Now()
 	prevTime := time.Now()
 	prevMessageCount := uint64(0)
+	prevMessageCached := uint64(0)
+	previousCachedInvalidations := uint64(0)
 	messageRateTs := []float64{}
+	cacheRateTs := []float64{}
+	cacheInvalidationsTs := []float64{}
+	percentilesTs := []map[string]float64{}
 	var dp datapoint
-	fmt.Printf("%26s %7s %25s %25s %7s %25s %25s %7s %25s\n", "Test time", " ", "Total Commands", "Total Errors", "", "Command Rate", "Client Cache Hits", "", "p50 lat. (msec)")
+	fmt.Printf("%26s %7s %25s %25s %7s %25s %25s %25s %25s\n", "Test time", " ", "Total Commands", "Total Errors", "", "Command Rate", "CSC Hits/sec", "CSC Invalidations/sec", "p50 lat. (msec)")
 	for {
 		select {
 		case dp = <-datapointsChan:
 			{
 				latencies.RecordValue(dp.duration_ms)
+				latenciesTick.RecordValue(dp.duration_ms)
 				if !dp.success {
 					currentErr++
 				}
@@ -412,41 +450,48 @@ func updateCLI(tick *time.Ticker, c chan os.Signal, message_limit uint64, loop b
 				now := time.Now()
 				took := now.Sub(prevTime)
 				messageRate := float64(totalCommands-prevMessageCount) / float64(took.Seconds())
+				InvalidationMessageRate := float64(totalCachedInvalidations-previousCachedInvalidations) / float64(took.Seconds())
+				CacheHitRate := float64(totalCached-prevMessageCached) / float64(took.Seconds())
 				completionPercentStr := "[----%]"
 				if !loop {
 					completionPercent := float64(totalCommands) / float64(message_limit) * 100.0
 					completionPercentStr = fmt.Sprintf("[%3.1f%%]", completionPercent)
 				}
 				errorPercent := float64(totalErrors) / float64(totalCommands) * 100.0
-				cachedPercent := 0.0
-				if totalCached > 0 {
-					cachedPercent = float64(totalCached) / float64(totalCommands) * 100.0
-				}
-
 				p50 := float64(latencies.ValueAtQuantile(50.0)) / 1000.0
 
 				if prevMessageCount == 0 && totalCommands != 0 {
 					start = time.Now()
 				}
 				if totalCommands != 0 {
 					messageRateTs = append(messageRateTs, messageRate)
+					cacheRateTs = append(cacheRateTs, CacheHitRate)
+					cacheInvalidationsTs = append(cacheInvalidationsTs, InvalidationMessageRate)
+					_, perTickLatencies := generateLatenciesMap(latenciesTick, took)
+					percentilesTs = append(percentilesTs, perTickLatencies)
+					latenciesTick.Reset()
 				}
+
 				prevMessageCount = totalCommands
+				prevMessageCached = totalCached
+				previousCachedInvalidations = totalCachedInvalidations
 				prevTime = now
 
-				fmt.Printf("%25.0fs %s %25d %25d [%3.1f%%] %25.0f %25d [%3.1f%%] %25.3f\t", time.Since(start).Seconds(), completionPercentStr, totalCommands, totalErrors, errorPercent, messageRate, totalCached, cachedPercent, p50)
+				fmt.Printf("%25.0fs %s %25d %25d [%3.1f%%] %25.0f %25.0f %25.0f %25.3f\t", time.Since(start).Seconds(), completionPercentStr, totalCommands, totalErrors, errorPercent, messageRate, CacheHitRate, InvalidationMessageRate, p50)
 				fmt.Printf("\r")
 				//w.Flush()
 				if message_limit > 0 && totalCommands >= uint64(message_limit) && !loop {
-					return true, start, time.Since(start), totalCommands, messageRateTs
+					end := time.Now()
+					return true, start, end, time.Since(start), totalCommands, totalCached, totalErrors, messageRateTs, cacheRateTs, cacheInvalidationsTs, percentilesTs
 				}
 
 				break
 			}
 
 		case <-c:
 			fmt.Println("\nreceived Ctrl-c - shutting down")
-			return true, start, time.Since(start), totalCommands, messageRateTs
+			end := time.Now()
+			return true, start, end, time.Since(start), totalCommands, totalCached, totalErrors, messageRateTs, cacheRateTs, cacheInvalidationsTs, percentilesTs
 		}
 	}
 }

test_result.go

@@ -0,0 +1,93 @@
+package main
+
+import (
+	"encoding/json"
+	"github.com/HdrHistogram/hdrhistogram-go"
+	"io/ioutil"
+	"log"
+	"time"
+)
+
+const resultFormatVersion = "0.0.1"
+
+type TestResult struct {
+
+	// Test Configs
+	resultFormatVersion string `json:"ResultFormatVersion"`
+	Metadata            string `json:"Metadata"`
+	Clients             uint   `json:"Clients"`
+	MaxRps              uint64 `json:"MaxRps"`
+	RandomSeed          int64  `json:"RandomSeed"`
+
+	StartTime      int64 `json:"StartTime"`
+	EndTime        int64 `json:"EndTime"`
+	DurationMillis int64 `json:"DurationMillis"`
+
+	// Populated after benchmark
+	// Benchmark Totals
+	Totals map[string]float64 `json:"Totals"`
+
+	// Overall Rates
+	OverallCommandRate         []float64 `json:"OverallCommandRate"`
+	OverallCSCHitRate          []float64 `json:"OverallCSCHitRate"`
+	OverallCSCInvalidationRate []float64 `json:"OverallCSCInvalidationRate"`
+
+	// Overall Client Quantiles
+	OverallClientLatencies []map[string]float64 `json:"OverallClientLatencies"`
+}
+
+func NewTestResult(metadata string, clients uint, maxRps uint64) *TestResult {
+	return &TestResult{resultFormatVersion: resultFormatVersion, Metadata: metadata, Clients: clients, MaxRps: maxRps}
+}
+
+func (r *TestResult) SetUsedRandomSeed(seed int64) *TestResult {
+	r.RandomSeed = seed
+	return r
+}
+
+func (r *TestResult) FillDurationInfo(startTime time.Time, endTime time.Time, duration time.Duration) {
+	r.StartTime = startTime.UTC().UnixNano() / 1000000
+	r.EndTime = endTime.UTC().UnixNano() / 1000000
+	r.DurationMillis = duration.Milliseconds()
+}
+
+func saveJsonResult(testResult *TestResult, jsonOutputFile string) {
+	if jsonOutputFile != "" {
+		file, err := json.MarshalIndent(testResult, "", " ")
+		if err != nil {
+			log.Fatal(err)
+		}
+		log.Printf("Saving JSON results file to %s\n", jsonOutputFile)
+		err = ioutil.WriteFile(jsonOutputFile, file, 0644)
+		if err != nil {
+			log.Fatal(err)
+		}
+	}
+}
+
+func calculateRateMetrics(current, prev int64, took time.Duration) (rate float64) {
+	rate = float64(current-prev) / float64(took.Seconds())
+	return
+}
+
+func generateLatenciesMap(hist *hdrhistogram.Histogram, tick time.Duration) (int64, map[string]float64) {
+	ops := hist.TotalCount()
+	percentilesTrack := []float64{0.0, 50.0, 95.0, 99.0, 99.9, 100.0}
+	q0 := 0.0
+	q50 := 0.0
+	q95 := 0.0
+	q99 := 0.0
+	q999 := 0.0
+	q100 := 0.0
+	if ops > 0 {
+		percentilesMap := hist.ValueAtPercentiles(percentilesTrack)
+		q0 = float64(percentilesMap[0.0]) / 10e2
+		q50 = float64(percentilesMap[50.0]) / 10e2
+		q95 = float64(percentilesMap[95.0]) / 10e2
+		q99 = float64(percentilesMap[99.0]) / 10e2
+		q999 = float64(percentilesMap[99.9]) / 10e2
+		q100 = float64(percentilesMap[100.0]) / 10e2
+	}
+	mp := map[string]float64{"q0": q0, "q50": q50, "q95": q95, "q99": q99, "q999": q999, "q100": q100, "ops/sec": float64(ops) / tick.Seconds()}
+	return ops, mp
+}