fix: getCurrentRps

internal/boomer/output.go

@@ -81,12 +81,8 @@ func getAvgContentLength(numRequests int64, totalContentLength int64) (avgConten
 	return avgContentLength
 }
 
-func getCurrentRps(numRequests int64, numReqsPerSecond map[int64]int64) (currentRps int64) {
+func getCurrentRps(numRequests int64) (currentRps float64) {
-	currentRps = int64(0)
+	currentRps = float64(numRequests) / float64(reportStatsInterval/time.Second)
-	numReqsPerSecondLength := int64(len(numReqsPerSecond))
-	if numReqsPerSecondLength != 0 {
-		currentRps = numRequests / numReqsPerSecondLength
-	}
 	return currentRps
 }
 
@@ -125,7 +121,7 @@ func (o *ConsoleOutput) OnEvent(data map[string]interface{}) {
 	}
 
 	currentTime := time.Now()
-	println(fmt.Sprintf("Current time: %s, Users: %d, State: %d, Total RPS: %d, Total Fail Ratio: %.1f%%",
+	println(fmt.Sprintf("Current time: %s, Users: %d, State: %d, Total RPS: %.1f, Total Fail Ratio: %.1f%%",
 		currentTime.Format("2006/01/02 15:04:05"), output.UserCount, output.State, output.TotalRPS, output.TotalFailRatio*100))
 	println(fmt.Sprintf("Accumulated Transactions: %d Passed, %d Failed",
 		output.TransactionsPassed, output.TransactionsFailed))
@@ -143,7 +139,7 @@ func (o *ConsoleOutput) OnEvent(data map[string]interface{}) {
 		row[6] = strconv.FormatInt(stat.MinResponseTime, 10)
 		row[7] = strconv.FormatInt(stat.MaxResponseTime, 10)
 		row[8] = strconv.FormatInt(stat.avgContentLength, 10)
-		row[9] = strconv.FormatInt(stat.currentRps, 10)
+		row[9] = strconv.FormatFloat(stat.currentRps, 'f', 2, 64)
 		row[10] = strconv.FormatInt(stat.currentFailPerSec, 10)
 		table.Append(row)
 	}
@@ -157,7 +153,7 @@ type statsEntryOutput struct {
 	medianResponseTime int64   // median response time
 	avgResponseTime    float64 // average response time, round float to 2 decimal places
 	avgContentLength   int64   // average content size
-	currentRps         int64   // # reqs/sec
+	currentRps         float64 // # reqs/sec
 	currentFailPerSec  int64   // # fails/sec
 }
 
@@ -167,7 +163,7 @@ type dataOutput struct {
 	TotalStats         *statsEntryOutput                 `json:"stats_total"`
 	TransactionsPassed int64                             `json:"transactions_passed"`
 	TransactionsFailed int64                             `json:"transactions_failed"`
-	TotalRPS           int64                             `json:"total_rps"`
+	TotalRPS           float64                           `json:"total_rps"`
 	TotalFailRatio     float64                           `json:"total_fail_ratio"`
 	Stats              []*statsEntryOutput               `json:"stats"`
 	Errors             map[string]map[string]interface{} `json:"errors"`
@@ -210,7 +206,7 @@ func convertData(data map[string]interface{}) (output *dataOutput, err error) {
 		TotalStats:         entryTotalOutput,
 		TransactionsPassed: transactionsPassed,
 		TransactionsFailed: transactionsFailed,
-		TotalRPS:           getCurrentRps(entryTotalOutput.NumRequests, entryTotalOutput.NumReqsPerSec),
+		TotalRPS:           getCurrentRps(entryTotalOutput.NumRequests),
 		TotalFailRatio:     getTotalFailRatio(entryTotalOutput.NumRequests, entryTotalOutput.NumFailures),
 		Stats:              make([]*statsEntryOutput, 0, len(stats)),
 	}
@@ -246,7 +242,7 @@ func deserializeStatsEntry(stat interface{}) (entryOutput *statsEntryOutput, err
 		medianResponseTime: getMedianResponseTime(numRequests, entry.ResponseTimes),
 		avgResponseTime:    getAvgResponseTime(numRequests, entry.TotalResponseTime),
 		avgContentLength:   getAvgContentLength(numRequests, entry.TotalContentLength),
-		currentRps:         getCurrentRps(numRequests, entry.NumReqsPerSec),
+		currentRps:         getCurrentRps(numRequests),
 		currentFailPerSec:  getCurrentFailPerSec(entry.NumFailures, entry.NumFailPerSec),
 	}
 	return
@@ -418,7 +414,7 @@ func (o *PrometheusPusherOutput) OnEvent(data map[string]interface{}) {
 	gaugeState.Set(float64(output.State))
 
 	// rps in total
-	gaugeTotalRPS.Set(float64(output.TotalRPS))
+	gaugeTotalRPS.Set(output.TotalRPS)
 
 	// failure ratio in total
 	gaugeTotalFailRatio.Set(output.TotalFailRatio)
@@ -437,7 +433,7 @@ func (o *PrometheusPusherOutput) OnEvent(data map[string]interface{}) {
 		gaugeMinResponseTime.WithLabelValues(method, name).Set(float64(stat.MinResponseTime))
 		gaugeMaxResponseTime.WithLabelValues(method, name).Set(float64(stat.MaxResponseTime))
 		gaugeAverageContentLength.WithLabelValues(method, name).Set(float64(stat.avgContentLength))
-		gaugeCurrentRPS.WithLabelValues(method, name).Set(float64(stat.currentRps))
+		gaugeCurrentRPS.WithLabelValues(method, name).Set(stat.currentRps)
 		gaugeCurrentFailPerSec.WithLabelValues(method, name).Set(float64(stat.currentFailPerSec))
 	}
 

internal/boomer/output_test.go

@@ -1,6 +1,7 @@
 package boomer
 
 import (
+	"fmt"
 	"math"
 	"testing"
 )
@@ -57,23 +58,17 @@ func TestGetAvgContentLength(t *testing.T) {
 }
 
 func TestGetCurrentRps(t *testing.T) {
-	numRequests := int64(10)
+	numRequests := int64(6)
-	numReqsPerSecond := map[int64]int64{}
+	currentRps := getCurrentRps(numRequests)
-
-	currentRps := getCurrentRps(numRequests, numReqsPerSecond)
-	if currentRps != 0 {
-		t.Error("currentRps should be 0")
-	}
-
-	numReqsPerSecond[1] = 2
-	numReqsPerSecond[2] = 3
-	numReqsPerSecond[3] = 2
-	numReqsPerSecond[4] = 3
-
-	currentRps = getCurrentRps(numRequests, numReqsPerSecond)
 	if currentRps != 2 {
 		t.Error("currentRps should be 2")
 	}
+
+	numRequests = int64(8)
+	currentRps = getCurrentRps(numRequests)
+	if fmt.Sprintf("%.2f", currentRps) != "2.67" {
+		t.Error("currentRps should be 2.67")
+	}
 }
 
 func TestConsoleOutput(t *testing.T) {

internal/boomer/ratelimiter.go

@@ -164,7 +164,7 @@ func (limiter *RampUpRateLimiter) Start() {
 			case <-quitChannel:
 				return
 			default:
-				atomic.StoreInt64(&limiter.currentThreshold, limiter.nextThreshold)
+				atomic.StoreInt64(&limiter.currentThreshold, atomic.LoadInt64(&limiter.nextThreshold))
 				time.Sleep(limiter.refillPeriod)
 				close(limiter.broadcastChannel)
 				limiter.broadcastChannel = make(chan bool)
@@ -178,7 +178,7 @@ func (limiter *RampUpRateLimiter) Start() {
 			case <-quitChannel:
 				return
 			default:
-				nextValue := limiter.nextThreshold + limiter.rampUpStep
+				nextValue := atomic.LoadInt64(&limiter.nextThreshold) + limiter.rampUpStep
 				if nextValue < 0 {
 					// int64 overflow
 					nextValue = int64(math.MaxInt64)
@@ -208,6 +208,6 @@ func (limiter *RampUpRateLimiter) Acquire() (blocked bool) {
 
 // Stop the rate limiter.
 func (limiter *RampUpRateLimiter) Stop() {
-	limiter.nextThreshold = 0
+	atomic.StoreInt64(&limiter.nextThreshold, 0)
 	close(limiter.quitChannel)
 }