WIP: add perf tests

Author: oleg-jukovec

tarantool_test.go

@@ -15,6 +15,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -95,6 +96,298 @@ var opts = Opts{
 
 const N = 500
 
+func BenchmarkPerformance_naive(b *testing.B) {
+	var err error
+
+	conn := test_helpers.ConnectWithValidation(b, dialer, opts)
+	defer conn.Close()
+
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Fatalf("failed to inialize database: %s", err)
+	}
+
+	b.ResetTimer()
+
+	for b.Loop() {
+		req := NewSelectRequest(spaceNo).
+			Index(indexNo).
+			Iterator(IterEq).
+			Key([]interface{}{uint(1111)})
+		data, err := conn.Do(req).Get()
+		if err != nil {
+			b.Errorf("request error: %s", err)
+		}
+
+		tuple := data[0].([]any)
+		if tuple[0].(uint16) != uint16(1111) {
+			b.Errorf("invalid result")
+		}
+	}
+}
+
+func BenchmarkPerformance_naive_with_single_request(b *testing.B) {
+	var err error
+
+	conn := test_helpers.ConnectWithValidation(b, dialer, opts)
+	defer conn.Close()
+
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Fatalf("failed to inialize database: %s", err)
+	}
+
+	req := NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Iterator(IterEq).
+		Key(UintKey{I: 1111})
+
+	b.ResetTimer()
+
+	for b.Loop() {
+		data, err := conn.Do(req).Get()
+		if err != nil {
+			b.Errorf("request error: %s", err)
+		}
+
+		tuple := data[0].([]any)
+		if tuple[0].(uint16) != uint16(1111) {
+			b.Errorf("invalid result")
+		}
+	}
+}
+
+type benchTuple struct {
+	id uint
+}
+
+func (t *benchTuple) DecodeMsgpack(dec *msgpack.Decoder) error {
+	l, err := dec.DecodeArrayLen()
+	if err != nil {
+		return fmt.Errorf("failed to decode tuples array: %w", err)
+	}
+
+	if l != 1 {
+		return fmt.Errorf("unexpected tuples array with len %d", l)
+	}
+
+	l, err = dec.DecodeArrayLen()
+	if err != nil {
+		return fmt.Errorf("failed to decode tuple array: %w", err)
+	}
+
+	if l < 1 {
+		return fmt.Errorf("too smal tuple have 0 fields")
+	}
+
+	t.id, err = dec.DecodeUint()
+	if err != nil {
+		return fmt.Errorf("failed to decode id: %w", err)
+	}
+
+	return nil
+}
+
+func BenchmarkPerformance_naive_with_custom_type(b *testing.B) {
+	var err error
+
+	conn := test_helpers.ConnectWithValidation(b, dialer, opts)
+	defer conn.Close()
+
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Fatalf("failed to inialize database: %s", err)
+	}
+
+	req := NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Iterator(IterEq).
+		Key(UintKey{I: 1111})
+
+	var tuple benchTuple
+
+	b.ResetTimer()
+
+	for b.Loop() {
+		err := conn.Do(req).GetTyped(&tuple)
+		if err != nil {
+			b.Errorf("request error: %s", err)
+		}
+
+		if tuple.id != 1111 {
+			b.Errorf("invalid result")
+		}
+	}
+}
+
+func BenchmarkPerformance_multithread(b *testing.B) {
+	var err error
+
+	conn := test_helpers.ConnectWithValidation(b, dialer, opts)
+	defer conn.Close()
+
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Fatalf("failed to inialize database: %s", err)
+	}
+
+	req := NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Iterator(IterEq).
+		Key(UintKey{I: 1111})
+
+	b.ResetTimer()
+
+	b.RunParallel(func(pb *testing.PB) {
+		var tuple benchTuple
+
+		for pb.Next() {
+			err := conn.Do(req).GetTyped(&tuple)
+			if err != nil {
+				b.Errorf("request error: %s", err)
+			}
+
+			if tuple.id != 1111 {
+				b.Errorf("invalid result")
+			}
+		}
+	})
+}
+
+func BenchmarkPerformance_multithread_parallelism(b *testing.B) {
+	var err error
+
+	conn := test_helpers.ConnectWithValidation(b, dialer, opts)
+	defer conn.Close()
+
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Fatalf("failed to inialize database: %s", err)
+	}
+
+	req := NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Iterator(IterEq).
+		Key(UintKey{I: 1111})
+
+	b.ResetTimer()
+
+	for p := 1; p <= 1024; p *= 2 {
+		b.Run(fmt.Sprintf("%d", p), func(b *testing.B) {
+			b.SetParallelism(p)
+			b.ResetTimer()
+
+			b.RunParallel(func(pb *testing.PB) {
+				var tuple benchTuple
+
+				for pb.Next() {
+					err := conn.Do(req).GetTyped(&tuple)
+					if err != nil {
+						b.Errorf("request error: %s", err)
+					}
+
+					if tuple.id != 1111 {
+						b.Errorf("invalid result")
+					}
+				}
+			})
+		})
+	}
+}
+
+func TestPerformance(t *testing.T) {
+	requests := int64(10_000_000)
+	connections := 16
+
+	ops := opts
+	ops.Concurrency = 8
+
+	conns := make([]*Connection, 0, connections)
+	for range connections {
+		conn := test_helpers.ConnectWithValidation(t, dialer, ops)
+		defer conn.Close()
+
+		conns = append(conns, conn)
+	}
+
+	_, err := conns[0].Replace(spaceNo, []interface{}{uint(1111)})
+	if err != nil {
+		t.Fatalf("failed to inialize database: %s", err)
+	}
+
+	req := NewSelectRequest(spaceNo).
+		Index(indexNo).
+		Iterator(IterEq).
+		Key(UintKey{I: 1111})
+
+	maxRps := float64(0)
+	maxConnections := 0
+	maxConcurrency := 0
+
+	for cn := 1; cn <= connections; cn *= 2 {
+		for cc := 1; cc <= 512; cc *= 2 {
+			var wg sync.WaitGroup
+
+			curRequests := requests
+
+			start := time.Now()
+
+			for i := range cc {
+				wg.Add(1)
+
+				ch := make(chan *Future, 1024)
+
+				go func(i int) {
+					defer close(ch)
+
+					for atomic.AddInt64(&curRequests, -1) >= 0 {
+						ch <- conns[i%cn].Do(req)
+					}
+				}(i)
+
+				go func() {
+					defer wg.Done()
+
+					var tuple benchTuple
+
+					for fut := range ch {
+						err := fut.GetTyped(&tuple)
+						if err != nil {
+							t.Errorf("request error: %s", err)
+						}
+
+						if tuple.id != 1111 {
+							t.Errorf("invalid result")
+						}
+					}
+				}()
+			}
+
+			wg.Wait()
+
+			duration := time.Since(start)
+
+			rps := float64(requests)/duration.Seconds()
+			fmt.Println("requests   :", requests)
+			fmt.Println("concurrency:", cc)
+			fmt.Println("connections:", cn)
+			fmt.Printf("duration   : %.2f\n", duration.Seconds())
+			fmt.Printf("requests/s : %.2f\n", rps)
+			fmt.Println("============")
+
+			if maxRps < rps {
+				maxRps = rps
+				maxConnections = cn
+				maxConcurrency = cc
+			}
+		}
+	}
+
+	fmt.Println("max connections:", maxConnections)
+	fmt.Println("max concurrency:", maxConcurrency)
+	fmt.Printf("max requests/s : %.2f", maxRps)
+}
+
 func BenchmarkClientSerial(b *testing.B) {
 	var err error