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

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,6 +21,17 @@ 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)
@@ -39,29 +49,25 @@ 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(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection, requestsSequence RequestsSequence, results *Results, 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) {
+func connectionLoop(ctx context.BenchCtx, tarantoolConnection *tarantool.Connection, requestsSequence RequestsSequence, results *Results, 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(ctx, tarantoolConnection, requestsSequence, results, backgroundCtx)
 		}()
 	}
 
@@ -72,15 +78,14 @@ 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,
-		Password: ctx.Password,
-	})
+	tarantoolConnection, err := connectBenchmarkSpace(ctx)
 	if err != nil {
-		return fmt.Errorf(
-			"Couldn't connect to Tarantool %s.",
-			ctx.URL)
+		return err
 	}
 	defer tarantoolConnection.Close()
 
@@ -103,13 +108,24 @@ func Run(ctx context.BenchCtx) error {
 		defer connectionPool[i].Close()
 	}
 
+	// 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 +135,13 @@ func Run(ctx context.BenchCtx) error {
 		waitGroup.Add(1)
 		go func(connection *tarantool.Connection) {
 			defer waitGroup.Done()
-			connectionLoop(ctx, connection, &results, backgroundCtx)
+			requestsSequence := RequestsSequence{
+				[requestTypesCount]RequestsPool{
+					{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, connection, requestsSequence, &results, backgroundCtx)
 		}(connectionPool[i])
 	}
 	// Sends "signal" to all "connectionLoop" and waits for them to complete.
@@ -130,7 +152,9 @@ func Run(ctx context.BenchCtx) error {
 	results.duration = time.Since(startTime).Seconds()
 	results.requestsPerSecond = int(float64(results.handledRequestsCount) / results.duration)
 
-	dropBenchmarkSpace(tarantoolConnection)
+	if err := dropBenchmarkSpace(tarantoolConnection); err != nil {
+		return err
+	}
 	fmt.Println("Benchmark stop")
 
 	printResults(results)

cli/bench/config.go

@@ -9,9 +9,11 @@ import (
 	"github.com/tarantool/cartridge-cli/cli/context"
 )
 
-var (
+const (
 	benchSpaceName             = "__benchmark_space__"
 	benchSpacePrimaryIndexName = "__bench_primary_key__"
+	requestTypesCount          = 3
+	PreFillingCount            = 1000000
 )
 
 // printConfig output formatted config parameters.
@@ -25,10 +27,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,53 @@
+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)}))
+	incrementRequest(err, request.results)
+}
+
+// Execute select operation
+func (request SelectRequest) execute() {
+	_, err := request.tarantoolConnection.Exec(tarantool.Call(request.getRandomTupleCommand, []interface{}{rand.Int()}))
+	incrementRequest(err, request.results)
+}
+
+// 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)))}))
+			incrementRequest(err, request.results)
+		}
+	}
+}
+
+// Return next operation in operations sequence
+func (requestsSequence RequestsSequence) getNext() Request {
+	for requestsSequence.currentCounter == 0 {
+		requestsSequence.currentRequestIndex++
+		requestsSequence.currentRequestIndex %= requestTypesCount
+		requestsSequence.currentCounter = requestsSequence.requests[requestsSequence.currentRequestIndex].count
+	}
+	requestsSequence.currentCounter--
+	return requestsSequence.requests[requestsSequence.currentRequestIndex].request
+}

cli/bench/space.go

@@ -3,10 +3,27 @@ package bench
 import (
 	"fmt"
 	"reflect"
+	"sync"
 
 	"github.com/FZambia/tarantool"
+	"github.com/tarantool/cartridge-cli/cli/common"
+	"github.com/tarantool/cartridge-cli/cli/context"
 )
 
+// connectBenchmarkSpace coonect to benchmark space
+func connectBenchmarkSpace(ctx context.BenchCtx) (*tarantool.Connection, error) {
+	tarantoolConnection, err := tarantool.Connect(ctx.URL, tarantool.Opts{
+		User:     ctx.User,
+		Password: ctx.Password,
+	})
+	if err != nil {
+		return tarantoolConnection, fmt.Errorf(
+			"Couldn't connect to Tarantool %s.",
+			ctx.URL)
+	}
+	return tarantoolConnection, nil
+}
+
 // createBenchmarkSpace creates benchmark space with formatting and primary index.
 func createBenchmarkSpace(tarantoolConnection *tarantool.Connection) error {
 	// Creating space.
@@ -54,3 +71,23 @@ 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,44 @@ type Results struct {
 	duration             float64 // Benchmark duration.
 	requestsPerSecond    int     // Cumber of requests per second - the main measured value.
 }
+
+// Ш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
+}
+
+// RequestsPool data structure for abstraction of a heap of identical requests
+type RequestsPool 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            [requestTypesCount]RequestsPool
+	currentRequestIndex int // currentRequestIndex describes what type of request will be issued by the sequence
+	currentCounter      int // currentCounter describes how many requests of the same type are left to issue from RequestsPool
+}

cli/commands/bench.go

@@ -30,4 +30,10 @@ func init() {
 	benchCmd.Flags().IntVar(&ctx.Bench.Duration, "duration", 10, "Duration of benchmark test (seconds)")
 	benchCmd.Flags().IntVar(&ctx.Bench.KeySize, "keysize", 10, "Size of key part of benchmark data (bytes)")
 	benchCmd.Flags().IntVar(&ctx.Bench.DataSize, "datasize", 20, "Size of value part of benchmark data (bytes)")
+
+	benchCmd.Flags().IntVar(&ctx.Bench.InsertCount, "insert", 100, "percentage of inserts")
+	benchCmd.Flags().IntVar(&ctx.Bench.SelectCount, "select", 0, "percentage of selects")
+	benchCmd.Flags().IntVar(&ctx.Bench.UpdateCount, "update", 0, "percentage of updates")
+	benchCmd.Flags().IntVar(&ctx.Bench.PreFillingCount, "fill", bench.PreFillingCount, "number of records to pre-fill the space")
+
 }

cli/context/context.go

@@ -184,4 +184,8 @@ type BenchCtx struct {
 	Duration             int    // Duration describes test duration in seconds.
 	KeySize              int    // DataSize describes the size of key part of benchmark data (bytes).
 	DataSize             int    // DataSize describes the size of value part of benchmark data (bytes).
+	InsertCount          int    // InsertCount describes the number of insert operations as a percentage
+	SelectCount          int    // SelectCount describes the number of select operations as a percentage
+	UpdateCount          int    // UpdateCount describes the number of update operations as a percentage
+	PreFillingCount      int    // PreFillingCount describes the number of records to pre-fill the space
 }