bench: add workers into rate mode

We need to run multiple workers for the rate mode and share the load
between them.

Close #170

Signed-off-by: Ekaterina Pavlova <ekt@morphbits.io>

README.md

@@ -276,7 +276,7 @@ The following default configurations are available:
        --to                         Number of fund receivers (default: 1)
        --vote                       Whether or not candidates should be voted for before the bench.
    -d                               Benchmark description.
-   -m                               Benchmark mode.
+   -m                               Benchmark mode. Possible values: rate, wrk. In rate mode, -q and -w flags should be specified. In wrk mode, only -w flag should be specified.
                                     Example: -m wrk -m rate
    -w                               Number of used workers.
                                     Example: -w 10 -w 15 -w 40

cmd/bench/main.go

@@ -33,7 +33,7 @@ func main() {
 	defer cancel()
 
 	var (
-		workers         int
+		workers         = v.GetInt("workers")
 		rate            int
 		msPerBlock      int
 		mempoolOOMDelay time.Duration
@@ -45,18 +45,12 @@ func main() {
 		client          *internal.RPCClient
 	)
 
-	switch mode {
-	case internal.ModeWorker:
-		workers = v.GetInt("workers")
-		client = internal.NewRPCClient(v, workers)
-
-	case internal.ModeRate:
-		workers = 1
+	if mode == internal.ModeRate {
 		rate = v.GetInt("rateLimit")
-		threshold = time.Duration(time.Second.Nanoseconds() / int64(rate))
-		client = internal.NewRPCClient(v, 1)
+		threshold = time.Duration(time.Second.Nanoseconds() / int64(rate) * int64(workers))
 	}
 
+	client = internal.NewRPCClient(v, workers)
 	version, err := client.GetVersion(ctx)
 	if err != nil {
 		log.Fatalf("could not receive RPC Node version: %v", err)

cmd/internal/worker.go

@@ -197,7 +197,7 @@ func NewWorkers(opts ...WorkerOption) (Worker, error) {
 
 	switch p.mode {
 	case ModeRate:
-		log.Printf("Init worker with %d QPS / %s time limit (%d txs will try to send)", p.rate, p.timeLimit, ln)
+		log.Printf("Init %d workers with %d QPS / %s time limit (%d txs will try to send)", p.wrkCount, p.rate, p.timeLimit, ln)
 	case ModeWorker:
 		log.Printf("Init %d workers / %s time limit (%d txs will try to send)", p.wrkCount, p.timeLimit, ln)
 	}
@@ -220,8 +220,9 @@ func NewWorkers(opts ...WorkerOption) (Worker, error) {
 // idx defines the order of the transaction being sent and can be more than overall transactions count, because retransmission is supported.
 func (d *doer) worker(ctx context.Context, idx *atomic.Int64, start time.Time) {
 	var (
-		done  = ctx.Done()
-		timer = time.NewTimer(d.timeLimit)
+		done           = ctx.Done()
+		timer          = time.NewTimer(d.timeLimit)
+		localTxCounter int64
 	)
 
 	defer func() {
@@ -237,7 +238,7 @@ loop:
 		case <-timer.C:
 			return
 		default:
-			i := idx.Add(1)
+			idx.Add(1)
 			if d.dump.TransactionsQueue.Len() == 0 {
 				return
 			}
@@ -264,10 +265,11 @@ loop:
 
 			since := time.Since(start)
 			count := d.countTxs.Add(1)
+			localTxCounter++
 			d.rpsReporter(float64(count) / since.Seconds())
 
 			if d.threshold > 0 {
-				waitFor := time.Until(start.Add(time.Duration(d.threshold.Nanoseconds() * (i + 1))))
+				waitFor := time.Until(start.Add(time.Duration(d.threshold.Nanoseconds() * (localTxCounter + 1))))
 				if waitFor > 0 {
 					time.Sleep(waitFor)
 				}

runner.sh

@@ -6,7 +6,9 @@ OUTPUT=""
 ARGS=(-i "/dump.txs")
 FILES=()
 MODE=""
-COUNT=""
+TARGET_RPS=""
+# Count of workers
+WORKERS_COUNT="30"
 IR_TYPE=go
 RPC_TYPE=
 RPC_ADDR=()
@@ -31,7 +33,7 @@ show_help() {
 	echo "       --to                         Number of fund receivers (default: 1)"
 	echo "       --vote                       Whether or not candidates should be voted for before the bench."
 	echo "   -d                               Benchmark description."
-	echo "   -m                               Benchmark mode."
+	echo "   -m                               Benchmark mode. Possible values: rate, wrk. In rate mode, -q and -w flags should be specified. In wrk mode, only -w flag should be specified."
 	echo "                                    Example: -m wrk -m rate"
 	echo "   -w                               Number of used workers."
 	echo "                                    Example: -w 10 -w 15 -w 40"
@@ -150,7 +152,7 @@ while test $# -gt 0; do
 	-w)
 		test $# -gt 0 || fatal "workers count should be specified"
 		ARGS+=(-w "$1")
-		COUNT="$1"
+		WORKERS_COUNT="$1"
 		shift
 		;;
 
@@ -163,7 +165,7 @@ while test $# -gt 0; do
 	-q)
 		test $# -gt 0 || fatal "benchmark rate limit should be specified"
 		ARGS+=(-q "$1")
-		COUNT="$1"
+		TARGET_RPS="$1"
 		shift
 		;;
 
@@ -261,7 +263,12 @@ else
 	fatal "Invalid validator count: $NEOBENCH_VALIDATOR_COUNT"
 fi
 
-OUTPUT="/out/${OUTPUT}_${MODE}_${COUNT}.log"
+if [ "rate" = "$MODE" ]; then
+  OUTPUT="/out/${OUTPUT}_${MODE}_${TARGET_RPS}_workers_${WORKERS_COUNT}.log"
+else
+  OUTPUT="/out/${OUTPUT}_${MODE}_${WORKERS_COUNT}.log"
+fi
+
 if [ ${#RPC_ADDR[@]} -eq 0 ]; then
 	ARGS+=("${DEFAULT_RPC_ADDR[@]}")
 else