Updates to perf tests

Author: Tim Middleton

.github/workflows/build-compatability-v1-1412.yaml

@@ -23,7 +23,7 @@ jobs:
       fail-fast: false
       matrix:
         coherenceVersion:
-          - 14.1.2-0-1
+          - 14.1.2-0-2
           - 14.1.2-0-2-SNAPSHOT
         go-version:
           - 1.23.x

test/e2e/perf/run_test.go

@@ -8,16 +8,151 @@ package perf
 
 import (
 	"github.com/onsi/gomega"
+	"github.com/oracle/coherence-go-client/v2/coherence"
+	"github.com/oracle/coherence-go-client/v2/coherence/aggregators"
+	"github.com/oracle/coherence-go-client/v2/coherence/filters"
 	"testing"
 )
 
-// TestPerformance1 tests for discovery
-func TestPerformance1(t *testing.T) {
+var (
+	filterIDBetween = filters.Between(idExtractor, 125_000, 150_000)
+	filterCountryAU = filters.Equal(countryExtractor, "Australia")
+	filterMultiple  = filters.Equal(countryExtractor, "Australia").And(filterIDBetween)
+	countryInFilter = filters.In(countryExtractor, []string{"Mexico", "Australia"})
+)
+
+// TestStreamingPerformance tests streaming.
+func TestStreamingPerformance(t *testing.T) {
+	var iterations int64 = 10
+
+	testCases := []struct {
+		testName string
+		test     func(t *testing.T, filter filters.Filter, iterations int64) *PerformanceResult
+		filter   filters.Filter
+		count    int64
+	}{
+		{"FilterBetween", RunTestFilterEntrySet, filterIDBetween, iterations},
+		{"FilterEquals", RunTestFilterEntrySet, filterCountryAU, iterations},
+		{"FilterMultiple", RunTestFilterEntrySet, filterMultiple, iterations},
+		{"FilterAlways", RunTestFilterEntrySet, filters.Always(), iterations},
+		{"FilterIn", RunTestFilterEntrySet, countryInFilter, iterations},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.testName, func(t *testing.T) {
+			result := tc.test(t, tc.filter, tc.count)
+			mapResults[tc.testName] = result
+		})
+	}
+}
+
+// TestStreamingPerformance tests streaming.
+func TestAggregatorPerformance(t *testing.T) {
+	var iterations int64 = 10
+
+	testCases := []struct {
+		testName string
+		test     func(t *testing.T, filter filters.Filter, iterations int64) *PerformanceResult
+		filter   filters.Filter
+		count    int64
+	}{
+		{"AggregatorCountAllFilter", RunCountAggregationTest, nil, iterations},
+		{"AggregatorCountCountryFilter", RunCountAggregationTest, filterCountryAU, iterations},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.testName, func(t *testing.T) {
+			result := tc.test(t, tc.filter, tc.count)
+			mapResults[tc.testName] = result
+		})
+	}
+}
+
+// TestKeyOperators tests key operations.
+func TestKeyOperators(t *testing.T) {
+	var iterations int64 = 100_000
+
+	testCases := []struct {
+		testName  string
+		test      func(t *testing.T, operation string, iterations int64) *PerformanceResult
+		operation string
+		count     int64
+	}{
+		{"KeyGet", RunTestKeyOperation, "get", iterations},
+		{"KeyPut", RunTestKeyOperation, "put", iterations},
+		{"KeyContainsKey", RunTestKeyOperation, "containsKey", iterations},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.testName, func(t *testing.T) {
+			result := tc.test(t, tc.operation, tc.count)
+			mapResults[tc.testName] = result
+		})
+	}
+}
+
+// RunTestFilterEntrySet runs tests against various filters.
+func RunTestFilterEntrySet(t *testing.T, filter filters.Filter, count int64) *PerformanceResult {
 	var (
 		g = gomega.NewWithT(t)
+		i int64
 	)
 
+	timer := newTestTimer(count)
+	for i = 0; i < count; i++ {
+		timer.Start()
+		for ch := range config.Students.EntrySetFilter(ctx, filter) {
+			g.Expect(ch.Err).To(gomega.BeNil())
+			_ = ch.Value
+		}
+		timer.End()
+	}
+
+	return timer.Complete()
+}
+
+// RunCountAggregationTest runs tests against various aggregators.
+func RunCountAggregationTest(t *testing.T, filter filters.Filter, count int64) *PerformanceResult {
+	var (
+		g    = gomega.NewWithT(t)
+		i    int64
+		fltr = filters.Always()
+	)
+
+	if filter != nil {
+		fltr = filter
+	}
+
+	timer := newTestTimer(count)
+	for i = 0; i < count; i++ {
+		timer.Start()
+		_, err := coherence.AggregateFilter[int, Student](ctx, config.Students, fltr, aggregators.Count())
+		g.Expect(err).To(gomega.BeNil())
+		timer.End()
+	}
+
+	return timer.Complete()
+}
+
+// RunTestKeyOperation runs key based tests.
+func RunTestKeyOperation(t *testing.T, operation string, count int64) *PerformanceResult {
+	g := gomega.NewWithT(t)
+
 	size, err := config.Students.Size(ctx)
-	g.Expect(err).To(gomega.BeNil())
-	g.Expect(size).To(gomega.Equal(maxStudents))
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+
+	timer := newTestTimer(count)
+	for i := 0; i < int(count); i++ {
+		id := rnd.Intn(size) + 1
+		timer.Start()
+		if operation == "get" {
+			_, err = config.Students.Get(ctx, id)
+		} else if operation == "put" {
+			_, err = config.Students.Put(ctx, id, getRandomStudent(id))
+		} else if operation == "containsKey" {
+			_, err = config.Students.ContainsKey(ctx, id)
+		}
+
+		g.Expect(err).To(gomega.BeNil())
+		timer.End()
+	}
+
+	return timer.Complete()
 }

test/e2e/perf/suite_test.go

@@ -10,9 +10,11 @@ import (
 	"context"
 	"fmt"
 	"github.com/oracle/coherence-go-client/v2/coherence"
+	"github.com/oracle/coherence-go-client/v2/coherence/extractors"
 	"log"
 	"math/rand"
 	"os"
+	"sort"
 	"testing"
 	"time"
 )
@@ -30,11 +32,23 @@ type Config struct {
 	Students coherence.NamedCache[int, Student]
 }
 
+type PerformanceResult struct {
+	Executions int64
+	TotalTime  int64
+	MinTime    time.Duration
+	MaxTime    time.Duration
+}
+
 var (
 	ctx = context.Background()
-	//nolint:gosec // just a test
-	rnd    = rand.New(rand.NewSource(time.Now().UnixNano()))
+	//nolint:gosec // just a test - have something consistent
+	rnd    = rand.New(rand.NewSource(123_456_789))
 	config = Config{}
+
+	courseExtractor  = extractors.Extract[string]("course")
+	countryExtractor = extractors.Extract[string]("country")
+	idExtractor      = extractors.Extract[int]("id")
+	mapResults       = make(map[string]*PerformanceResult)
 )
 
 const (
@@ -59,24 +73,66 @@ func TestMain(m *testing.M) {
 		errorAndExit("unable to clear", err)
 	}
 
+	// Add Indexes
+	err = coherence.AddIndex[int, Student](ctx, config.Students, courseExtractor, true)
+	if err != nil {
+		errorAndExit("unable to add index", err)
+	}
+
+	err = coherence.AddIndex[int, Student](ctx, config.Students, countryExtractor, true)
+	if err != nil {
+		errorAndExit("unable to add index", err)
+	}
+
+	err = coherence.AddIndex[int, Student](ctx, config.Students, idExtractor, true)
+	if err != nil {
+		errorAndExit("unable to add index", err)
+	}
+
 	log.Println("Populating cache with", maxStudents, "students")
 	if populateCache(config.Students, maxStudents) != nil {
 		errorAndExit("failed to populate cache", err)
 	}
 
 	size, err = config.Students.Size(ctx)
-	if err != nil || size != maxStudents {
-		errorAndExit("size not correct", err)
+	if err != nil {
+		errorAndExit("error", err)
+	}
+	if size != maxStudents {
+		errorAndExit("count", fmt.Errorf("invalid number of students: %d", size))
 	}
 	log.Println("Cache size is", size)
 
 	exitCode := m.Run()
 
 	fmt.Printf("Tests completed with return code %d\n", exitCode)
+	printResults()
 
 	os.Exit(exitCode)
 }
 
+func printResults() {
+	keys := make([]string, 0, len(mapResults))
+	for k := range mapResults {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	// Sort keys
+	sort.Strings(keys)
+	fmt.Println()
+	fmt.Println("RESULTS START")
+	fmt.Println()
+	fmt.Printf("%-30s %15s %15s %15s %15s %15s\n", "TEST", "TOTAL TIME", "EXECUTIONS", "MIN", "MAX", "AVERAGE")
+	for _, k := range keys {
+		v := mapResults[k]
+		fmt.Printf("%-30s %15v %15d %15v %15v %15v\n", k, time.Duration(v.TotalTime), v.Executions,
+			v.MinTime, v.MaxTime, time.Duration(v.TotalTime/v.Executions))
+	}
+	fmt.Println()
+	fmt.Println("RESULTS END")
+}
+
 func errorAndExit(message string, err error) {
 	fmt.Println(message, err)
 	os.Exit(1)
@@ -101,23 +157,20 @@ func InitializeCoherence(ctx context.Context, address string) (Config, error) {
 	return config, nil
 }
 
+var (
+	courses   = []string{"C1", "C2", "C3", "C4"}
+	countries = []string{"Australia", "USA", "Canada", "Mexico"}
+)
+
 func populateCache(cache coherence.NamedMap[int, Student], count int) error {
 	var (
 		buffer    = make(map[int]Student)
 		err       error
-		courses   = []string{"C1", "C2", "C3", "C4"}
-		countries = []string{"Australia", "USA", "Canada", "Mexico"}
 		batchSize = 10_000
 	)
 
 	for i := 1; i <= count; i++ {
-		buffer[i] = Student{
-			ID:      i,
-			Name:    fmt.Sprintf("student%d", i),
-			Address: fmt.Sprintf("address%d", i),
-			Country: randomize(countries),
-			Course:  randomize(courses),
-		}
+		buffer[i] = getRandomStudent(i)
 		if i%batchSize == 0 {
 			err = cache.PutAll(ctx, buffer)
 			if err != nil {
@@ -133,9 +186,61 @@ func populateCache(cache coherence.NamedMap[int, Student], count int) error {
 	return nil
 }
 
+func getRandomStudent(i int) Student {
+	return Student{
+		ID:      i,
+		Name:    fmt.Sprintf("student%d", i),
+		Address: fmt.Sprintf("address%d", i),
+		Country: randomize(countries),
+		Course:  randomize(courses),
+	}
+}
+
 func randomize(arr []string) string {
 	if len(arr) == 0 {
 		return ""
 	}
 	return arr[rnd.Intn(len(arr))]
 }
+
+type testTimer struct {
+	startTime    time.Time
+	endTime      time.Time
+	minDuration  time.Duration
+	maxDuration  time.Duration
+	currentStart time.Time
+	count        int64
+}
+
+func (t *testTimer) Start() {
+	t.currentStart = time.Now()
+}
+
+func (t *testTimer) End() {
+	duration := time.Since(t.currentStart)
+	if duration < t.minDuration {
+		t.minDuration = duration * time.Nanosecond
+	}
+	if duration > t.maxDuration {
+		t.maxDuration = duration * time.Nanosecond
+	}
+}
+
+func (t *testTimer) Complete() *PerformanceResult {
+	t.endTime = time.Now()
+	return &PerformanceResult{
+		Executions: t.count,
+		TotalTime:  t.endTime.Sub(t.startTime).Nanoseconds(),
+		MaxTime:    t.maxDuration,
+		MinTime:    t.minDuration,
+	}
+}
+
+func newTestTimer(count int64) *testTimer {
+	return &testTimer{
+		startTime:   time.Now(),
+		count:       count,
+		minDuration: time.Duration(10000) * time.Second,
+		maxDuration: 0,
+	}
+}