Part of tarantool#645 bench: load profile patch (select and update)

user@cartridge-cli % ./cartridge bench --select=30 --update=30 --insert=40 --fill=100000
Tarantool 2.8.2 (Binary) f4897ffe-98dd-40fc-a6f2-21ca8bb52fe7

Parameters:
        URL: 127.0.0.1:3301
        user: guest
        connections: 10
        simultaneous requests: 10
        duration: 10 seconds
        key size: 10 bytes
        data size: 20 bytes
        insert: 40 percentages
        select: 30 percentages
        update: 30 percentages

Data schema
| key                            | value
| ------------------------------ | ------------------------------
| random(10)                     | random(20)

The pre-filling of the space has started,
because the insert operation is not specified
or there was an explicit instruction for pre-filling
...
Pre-filling is finished

Benchmark start
...
Benchmark stop

Results:
        Success operations: 1430826
        Failed  operations: 0
        Request count: 1432454
        Time (seconds): 10.000640
        Requests per second: 143236

Author: Kirill-Churkin

cli/bench/bench.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/FZambia/tarantool"
-	"github.com/tarantool/cartridge-cli/cli/common"
 	"github.com/tarantool/cartridge-cli/cli/context"
 )
 
@@ -22,14 +21,25 @@ func printResults(results Results) {
 	fmt.Printf("\tRequests per second: %d\n\n", results.requestsPerSecond)
 }
 
+// verifyOperationsPercentage checks that the amount of operations percentage is 100.
+func verifyOperationsPercentage(ctx *context.BenchCtx) error {
+	entire_percentage := ctx.InsertCount + ctx.SelectCount + ctx.UpdateCount
+	if entire_percentage != 100 {
+		return fmt.Errorf(
+			"The number of operations as a percentage should be equal to 100, " +
+				"note that by default the percentage of inserts is 100")
+	}
+	return nil
+}
+
 // spacePreset prepares space for a benchmark.
 func spacePreset(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection) error {
 	dropBenchmarkSpace(tarantoolConnection)
 	return createBenchmarkSpace(tarantoolConnection)
 }
 
 // incrementRequest increases the counter of successful/failed requests depending on the presence of an error.
-func incrementRequest(err error, results *Results) {
+func (results *Results) incrementRequestsCounters(err error) {
 	if err == nil {
 		results.successResultCount++
 	} else {
@@ -39,29 +49,30 @@ func incrementRequest(err error, results *Results) {
 }
 
 // requestsLoop continuously executes the insert query until the benchmark time runs out.
-func requestsLoop(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection, results *Results, backgroundCtx bctx.Context) {
+func requestsLoop(requestsSequence *RequestsSequence, backgroundCtx bctx.Context) {
 	for {
 		select {
 		case <-backgroundCtx.Done():
 			return
 		default:
-			_, err := tarantoolConnection.Exec(
-				tarantool.Insert(
-					benchSpaceName,
-					[]interface{}{common.RandomString(ctx.KeySize), common.RandomString(ctx.DataSize)}))
-			incrementRequest(err, results)
+			requestsSequence.getNext().execute()
 		}
 	}
 }
 
-// connectionLoop runs "ctx.SimultaneousRequests" requests execution threads through the same connection.
-func connectionLoop(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection, results *Results, backgroundCtx bctx.Context) {
+// connectionLoop runs "ctx.SimultaneousRequests" requests execution threads
+// through the same connection.
+func connectionLoop(
+	ctx *context.BenchCtx,
+	requestsSequence RequestsSequence,
+	backgroundCtx bctx.Context,
+) {
 	var connectionWait sync.WaitGroup
 	for i := 0; i < ctx.SimultaneousRequests; i++ {
 		connectionWait.Add(1)
 		go func() {
 			defer connectionWait.Done()
-			requestsLoop(ctx, tarantoolConnection, results, backgroundCtx)
+			requestsLoop(&requestsSequence, backgroundCtx)
 		}()
 	}
 
@@ -72,6 +83,10 @@ func connectionLoop(ctx context.BenchCtx, tarantoolConnection *tarantool.Connect
 func Run(ctx context.BenchCtx) error {
 	rand.Seed(time.Now().UnixNano())
 
+	if err := verifyOperationsPercentage(&ctx); err != nil {
+		return err
+	}
+
 	// Connect to tarantool and preset space for benchmark.
 	tarantoolConnection, err := tarantool.Connect(ctx.URL, tarantool.Opts{
 		User:     ctx.User,
@@ -90,7 +105,8 @@ func Run(ctx context.BenchCtx) error {
 		return err
 	}
 
-	/// Сreate a "connectionPool" before starting the benchmark to exclude the connection establishment time from measurements.
+	// Сreate a "connectionPool" before starting the benchmark
+	// to exclude the connection establishment time from measurements.
 	connectionPool := make([]*tarantool.Connection, ctx.Connections)
 	for i := 0; i < ctx.Connections; i++ {
 		connectionPool[i], err = tarantool.Connect(ctx.URL, tarantool.Opts{
@@ -103,13 +119,28 @@ func Run(ctx context.BenchCtx) error {
 		defer connectionPool[i].Close()
 	}
 
-	fmt.Println("Benchmark start")
+	// Pre-fill benchmark space if required.
+	if ctx.InsertCount == 0 || ctx.PreFillingCount != PreFillingCount {
+		fmt.Println("The pre-filling of the space has started,\n" +
+			"because the insert operation is not specified\n" +
+			"or there was an explicit instruction for pre-filling.")
+		fmt.Println("...")
+		fillBenchmarkSpace(ctx, connectionPool)
+		fmt.Printf("Pre-filling is finished.\n\n")
+	}
+
+	fmt.Println("Benchmark start.")
 	fmt.Println("...")
 
 	// The "context" will be used to stop all "connectionLoop" when the time is out.
 	backgroundCtx, cancel := bctx.WithCancel(bctx.Background())
 	var waitGroup sync.WaitGroup
 	results := Results{}
+	getRandomTupleCommand := fmt.Sprintf(
+		"box.space.%s.index.%s:random",
+		benchSpaceName,
+		benchSpacePrimaryIndexName,
+	)
 
 	startTime := time.Now()
 	timer := time.NewTimer(time.Duration(ctx.Duration * int(time.Second)))
@@ -119,7 +150,39 @@ func Run(ctx context.BenchCtx) error {
 		waitGroup.Add(1)
 		go func(connection *tarantool.Connection) {
 			defer waitGroup.Done()
-			connectionLoop(ctx, connection, &results, backgroundCtx)
+			requestsSequence := RequestsSequence{
+				[]RequestsGenerator{
+					{
+						InsertRequest{
+							ctx,
+							connection,
+							&results,
+						},
+						ctx.InsertCount,
+					},
+					{
+						SelectRequest{
+							ctx,
+							connection,
+							&results,
+							getRandomTupleCommand,
+						},
+						ctx.SelectCount,
+					},
+					{
+						UpdateRequest{
+							ctx,
+							connection,
+							&results,
+							getRandomTupleCommand,
+						},
+						ctx.UpdateCount,
+					},
+				},
+				0,
+				ctx.InsertCount,
+			}
+			connectionLoop(&ctx, requestsSequence, backgroundCtx)
 		}(connectionPool[i])
 	}
 	// Sends "signal" to all "connectionLoop" and waits for them to complete.
@@ -130,8 +193,10 @@ func Run(ctx context.BenchCtx) error {
 	results.duration = time.Since(startTime).Seconds()
 	results.requestsPerSecond = int(float64(results.handledRequestsCount) / results.duration)
 
-	dropBenchmarkSpace(tarantoolConnection)
-	fmt.Println("Benchmark stop")
+	if err := dropBenchmarkSpace(tarantoolConnection); err != nil {
+		return err
+	}
+	fmt.Println("Benchmark stop.")
 
 	printResults(results)
 	return nil

cli/bench/config.go

@@ -9,9 +9,10 @@ import (
 	"github.com/tarantool/cartridge-cli/cli/context"
 )
 
-var (
+const (
 	benchSpaceName             = "__benchmark_space__"
 	benchSpacePrimaryIndexName = "__bench_primary_key__"
+	PreFillingCount            = 1000000
 )
 
 // printConfig output formatted config parameters.
@@ -25,10 +26,15 @@ func printConfig(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection
 	fmt.Printf("\tduration: %d seconds\n", ctx.Duration)
 	fmt.Printf("\tkey size: %d bytes\n", ctx.KeySize)
 	fmt.Printf("\tdata size: %d bytes\n", ctx.DataSize)
+	fmt.Printf("\tinsert: %d percentages\n", ctx.InsertCount)
+	fmt.Printf("\tselect: %d percentages\n", ctx.SelectCount)
+	fmt.Printf("\tupdate: %d percentages\n\n", ctx.UpdateCount)
+
 	fmt.Printf("Data schema\n")
 	w := tabwriter.NewWriter(os.Stdout, 1, 1, 1, ' ', 0)
 	fmt.Fprintf(w, "|\tkey\t|\tvalue\n")
 	fmt.Fprintf(w, "|\t------------------------------\t|\t------------------------------\n")
 	fmt.Fprintf(w, "|\trandom(%d)\t|\trandom(%d)\n", ctx.KeySize, ctx.DataSize)
 	w.Flush()
+	fmt.Println()
 }

cli/bench/requests.go

@@ -0,0 +1,71 @@
+package bench
+
+import (
+	"math/rand"
+	"reflect"
+
+	"github.com/FZambia/tarantool"
+	"github.com/tarantool/cartridge-cli/cli/common"
+)
+
+// Execute insert operation.
+func (request InsertRequest) execute() {
+	_, err := request.tarantoolConnection.Exec(
+		tarantool.Insert(
+			benchSpaceName,
+			[]interface{}{
+				common.RandomString(request.ctx.KeySize),
+				common.RandomString(request.ctx.DataSize),
+			}))
+	request.results.incrementRequestsCounters(err)
+}
+
+// Execute select operation.
+func (request SelectRequest) execute() {
+	_, err := request.tarantoolConnection.Exec(tarantool.Call(
+		request.getRandomTupleCommand,
+		[]interface{}{rand.Int()}))
+	request.results.incrementRequestsCounters(err)
+}
+
+// Execute update operation.
+func (request UpdateRequest) execute() {
+	getRandomTupleResponse, err := request.tarantoolConnection.Exec(
+		tarantool.Call(request.getRandomTupleCommand,
+			[]interface{}{rand.Int()}))
+	if err == nil {
+		data := getRandomTupleResponse.Data
+		if len(data) > 0 {
+			key := reflect.ValueOf(data[0]).Index(0).Elem().String()
+			_, err := request.tarantoolConnection.Exec(
+				tarantool.Update(
+					benchSpaceName,
+					benchSpacePrimaryIndexName,
+					[]interface{}{key},
+					[]tarantool.Op{tarantool.Op(
+						tarantool.OpAssign(
+							2,
+							common.RandomString(request.ctx.DataSize)))}))
+			request.results.incrementRequestsCounters(err)
+		}
+	}
+}
+
+// getNext return next operation in operations sequence.
+func (requestsSequence *RequestsSequence) getNext() Request {
+	// If at the moment the number of remaining requests = 0,
+	// then find a new generator, which requests count > 0.
+	// If new generator has requests count = 0, then repeat.
+	for requestsSequence.currentCounter == 0 {
+		// Increase the index, which means logical switching to a new generator.
+		requestsSequence.currentRequestIndex++
+		requestsSequence.currentRequestIndex %= len(requestsSequence.requests)
+		// Get new generator by index.
+		nextRequestsGenerator := requestsSequence.requests[requestsSequence.currentRequestIndex]
+		// Get requests count for new operation.
+		requestsSequence.currentCounter = nextRequestsGenerator.count
+	}
+	// Logical taking of a single request.
+	requestsSequence.currentCounter--
+	return requestsSequence.requests[requestsSequence.currentRequestIndex].request
+}

cli/bench/space.go

@@ -3,15 +3,21 @@ package bench
 import (
 	"fmt"
 	"reflect"
+	"sync"
 
 	"github.com/FZambia/tarantool"
+	"github.com/tarantool/cartridge-cli/cli/common"
+	"github.com/tarantool/cartridge-cli/cli/context"
 )
 
 // createBenchmarkSpace creates benchmark space with formatting and primary index.
 func createBenchmarkSpace(tarantoolConnection *tarantool.Connection) error {
 	// Creating space.
 	createCommand := "return box.schema.space.create(...).name"
-	_, err := tarantoolConnection.Exec(tarantool.Eval(createCommand, []interface{}{benchSpaceName, map[string]bool{"if_not_exists": true}}))
+	_, err := tarantoolConnection.Exec(tarantool.Eval(
+		createCommand,
+		[]interface{}{benchSpaceName, map[string]bool{"if_not_exists": true}},
+	))
 	if err != nil {
 		return err
 	}
@@ -54,3 +60,29 @@ func dropBenchmarkSpace(tarantoolConnection *tarantool.Connection) error {
 	}
 	return nil
 }
+
+// fillBenchmarkSpace fills benchmark space
+// with a PreFillingCount number of records
+// using connectionPool for fast filling.
+func fillBenchmarkSpace(ctx context.BenchCtx, connectionPool []*tarantool.Connection) {
+	var waitGroup sync.WaitGroup
+	filledCount := 0
+	for i := 0; i < ctx.Connections; i++ {
+		waitGroup.Add(1)
+		go func(tarantoolConnection *tarantool.Connection) {
+			defer waitGroup.Done()
+			for filledCount < ctx.PreFillingCount {
+				tarantoolConnection.Exec(tarantool.Insert(
+					benchSpaceName,
+					[]interface{}{
+						common.RandomString(ctx.KeySize),
+						common.RandomString(ctx.DataSize),
+					},
+				))
+				filledCount++
+			}
+			return
+		}(connectionPool[i])
+	}
+	waitGroup.Wait()
+}

cli/bench/types.go

@@ -1,5 +1,10 @@
 package bench
 
+import (
+	"github.com/FZambia/tarantool"
+	"github.com/tarantool/cartridge-cli/cli/context"
+)
+
 // Results describes set of benchmark results.
 type Results struct {
 	handledRequestsCount int     // Count of all executed requests.
@@ -8,3 +13,47 @@ type Results struct {
 	duration             float64 // Benchmark duration.
 	requestsPerSecond    int     // Cumber of requests per second - the main measured value.
 }
+
+// Request is nterface for various types of requests.
+type Request interface {
+	execute()
+}
+
+// InsertRequest data structure for insert request.
+type InsertRequest struct {
+	ctx                 context.BenchCtx
+	tarantoolConnection *tarantool.Connection
+	results             *Results
+}
+
+// SelectRequest data structure for select request.
+type SelectRequest struct {
+	ctx                   context.BenchCtx
+	tarantoolConnection   *tarantool.Connection
+	results               *Results
+	getRandomTupleCommand string
+}
+
+// UpdateRequest data structure for update request.
+type UpdateRequest struct {
+	ctx                   context.BenchCtx
+	tarantoolConnection   *tarantool.Connection
+	results               *Results
+	getRandomTupleCommand string
+}
+
+// RequestsGenerator data structure for abstraction of a renewable heap of identical requests.
+type RequestsGenerator struct {
+	request Request // InsertRequest, SelectRequest or UpdateRequest.
+	count   int     // Count of requests.
+}
+
+// RequestsSequence data structure for abstraction for the constant issuance of new requests.
+type RequestsSequence struct {
+	requests []RequestsGenerator
+	// currentRequestIndex describes what type of request will be issued by the sequence.
+	currentRequestIndex int
+	// currentCounter describes how many requests of the same type
+	// are left to issue from RequestsPool.
+	currentCounter int
+}