Merge pull request #21488 from serathius/robustness-benmarks

Improve robustness benchmarks

Author: serathius

tests/robustness/model/replay.go

@@ -18,8 +18,18 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+
+	"github.com/anishathalye/porcupine"
 )
 
+func NewReplayFromOperations(ops []porcupine.Operation) *EtcdReplay {
+	requests := []EtcdRequest{}
+	for _, op := range ops {
+		requests = append(requests, op.Input.(EtcdRequest))
+	}
+	return NewReplay(requests)
+}
+
 func NewReplay(persistedRequests []EtcdRequest) *EtcdReplay {
 	state := freshEtcdState()
 	// Padding for index 0 and 1, so the index matches the revision.

tests/robustness/validate/operations_test.go

@@ -296,56 +296,60 @@ func keyValueRevision(key, value string, rev int64) model.KeyValue {
 
 func BenchmarkValidateLinearizableOperations(b *testing.B) {
 	lg := zap.NewNop()
-	b.Run("Successes", func(b *testing.B) {
-		history := allPutSuccesses(1000)
+	b.Run("SequentialSuccessPuts", func(b *testing.B) {
+		history := sequentialSuccessPuts(7000, 2)
 		shuffles := shuffleHistory(history, b.N)
 		b.ResetTimer()
 		validateShuffles(b, lg, shuffles, time.Second)
 	})
-	b.Run("AllFailures", func(b *testing.B) {
-		history := allPutFailures(10)
+	b.Run("SequentialFailedPuts", func(b *testing.B) {
+		history := sequentialFailedPuts(14, 1)
 		shuffles := shuffleHistory(history, b.N)
 		b.ResetTimer()
 		validateShuffles(b, lg, shuffles, time.Second)
 	})
-	b.Run("PutFailuresWithRead", func(b *testing.B) {
-		history := putFailuresWithRead(b, 8)
+	b.Run("ConcurrentFailedPutsWithRead", func(b *testing.B) {
+		history := concurrentFailedPutsWithRead(b, 13)
 		shuffles := shuffleHistory(history, b.N)
 		b.ResetTimer()
 		validateShuffles(b, lg, shuffles, time.Second)
 	})
+	b.Run("BacktrackingHeavy", func(b *testing.B) {
+		history := backtrackingHeavy(b)
+		shuffles := shuffleHistory(history, b.N)
+		b.ResetTimer()
+		for i := 0; i < len(shuffles); i++ {
+			validateLinearizableOperationsAndVisualize(lg, shuffles[i], time.Second)
+		}
+	})
 }
 
-func allPutSuccesses(concurrencyCount int) []porcupine.Operation {
+func sequentialSuccessPuts(count int, startRevision int64) []porcupine.Operation {
 	ops := []porcupine.Operation{}
-	for i := 0; i < concurrencyCount; i++ {
+	for i := 0; i < count; i++ {
 		ops = append(ops, porcupine.Operation{
 			ClientId: i,
 			Input:    putRequest("key", "value"),
-			Output:   txnResponse(int64(i)+2, model.EtcdOperationResult{}),
-			Call:     int64(i),
-			Return:   int64(i) + int64(concurrencyCount),
+			Output:   txnResponse(startRevision+int64(i), model.EtcdOperationResult{}),
+			Call:     int64(i * 2),
+			Return:   int64(i*2 + 1),
 		})
 	}
 	return ops
 }
 
-func putFailuresWithRead(b *testing.B, concurrencyCount int) []porcupine.Operation {
+func concurrentFailedPutsWithRead(b *testing.B, concurrencyCount int) []porcupine.Operation {
 	ops := []porcupine.Operation{}
 	for i := 0; i < concurrencyCount; i++ {
 		ops = append(ops, porcupine.Operation{
 			ClientId: i,
-			Input:    putRequest(fmt.Sprintf("key%d", i), "value"),
+			Input:    putRequest("key", "value"),
 			Output:   errorResponse(fmt.Errorf("timeout")),
 			Call:     int64(i),
 			Return:   int64(i) + int64(concurrencyCount),
 		})
 	}
-	requests := []model.EtcdRequest{}
-	for _, op := range ops {
-		requests = append(requests, op.Input.(model.EtcdRequest))
-	}
-	replay := model.NewReplay(requests)
+	replay := model.NewReplayFromOperations(ops)
 	state, err := replay.StateForRevision(int64(concurrencyCount) + 1)
 	if err != nil {
 		b.Fatal(err)
@@ -362,17 +366,72 @@ func putFailuresWithRead(b *testing.B, concurrencyCount int) []porcupine.Operati
 	return ops
 }
 
-func allPutFailures(concurrencyCount int) []porcupine.Operation {
+func sequentialFailedPuts(count int, keyCount int) []porcupine.Operation {
 	ops := []porcupine.Operation{}
-	for i := 0; i < concurrencyCount; i++ {
+	for i := 0; i < count; i++ {
+		key := "key0"
+		if keyCount > 1 {
+			key = fmt.Sprintf("key%d", i%keyCount)
+		}
 		ops = append(ops, porcupine.Operation{
 			ClientId: i,
-			Input:    putRequest("key", "value"),
+			Input:    putRequest(key, "value"),
 			Output:   errorResponse(fmt.Errorf("timeout")),
+			Call:     int64(i * 2),
+			Return:   int64(i*2 + 1),
+		})
+	}
+	return ops
+}
+
+func backtrackingHeavy(b *testing.B) (ops []porcupine.Operation) {
+	for i := 0; i < 30; i++ {
+		ops = append(ops, porcupine.Operation{
+			ClientId: -1,
+			Input:    putRequest(fmt.Sprintf("key%d", i+1000), "value"),
+			Output:   txnResponse(int64(i+2), model.EtcdOperationResult{}),
 			Call:     int64(i),
-			Return:   int64(i) + int64(concurrencyCount),
+			Return:   int64(i) + 1,
+		})
+	}
+	startTime := int64(1000)
+
+	failedPuts := 4
+	for i := 0; i < failedPuts; i++ {
+		ops = append(ops, porcupine.Operation{
+			ClientId: i,
+			Input:    putRequest(fmt.Sprintf("key%d", i), "value"),
+			Output:   errorResponse(fmt.Errorf("timeout")),
+			Call:     startTime + int64(i),
+			Return:   startTime + 1000 + int64(i),
+		})
+	}
+	replay := model.NewReplayFromOperations(ops)
+	state, err := replay.StateForRevision(int64(30 + 1))
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	concurrentReads := 3
+	for i := 0; i < concurrentReads; i++ {
+		request := rangeRequest(fmt.Sprintf("key%d", i), "", 0, 0)
+		_, resp := state.Step(request)
+		ops = append(ops, porcupine.Operation{
+			ClientId: failedPuts + i,
+			Input:    request,
+			Output:   resp,
+			Call:     startTime + 1100,
+			Return:   startTime + 2100,
 		})
 	}
+
+	ops = append(ops, porcupine.Operation{
+		ClientId: 99,
+		Input:    rangeRequest("key0", "", 0, 0),
+		Output:   rangeResponse(0, keyValueRevision("key0", "wrong", 9999)),
+		Call:     startTime + 3000,
+		Return:   startTime + 4000,
+	})
 	return ops
 }