vitalif
/
etcd
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request #15159 from serathius/linearizability-separate-request-type 

tests: Separate request type from Txn operation type
dependabot/go_modules/go.uber.org/atomic-1.10.0
Marek Siarkowicz 2023-01-24 13:47:55 +01:00 committed by GitHub
parent 0d7b624fae 844ac9c76d
commit 207e0f74aa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 249 additions and 189 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/linearizability/client.go
View File

@ -78,7 +78,7 @@ func (c *recordingClient) Delete(ctx context.Context, key string) error {
return nil
}
func (c *recordingClient) Txn(ctx context.Context, key, expectedValue, newValue string) error {
func (c *recordingClient) CompareAndSet(ctx context.Context, key, expectedValue, newValue string) error {
callTime := time.Now()
txn := c.client.Txn(ctx)
var cmp clientv3.Cmp

57
tests/linearizability/linearizability_test.go
View File

@ -46,13 +46,19 @@ var (
minimalQPS: 100,
maximalQPS: 200,
clientCount: 8,
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []opChance{{operation: model.Put, chance: 50}, {operation: model.Delete, chance: 10}, {operation: model.PutWithLease, chance: 10}, {operation: model.LeaseRevoke, chance: 10}, {operation: model.Txn, chance: 20}}},
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []requestChance{
{operation: Put, chance: 50},
{operation: Delete, chance: 10},
{operation: PutWithLease, chance: 10},
{operation: LeaseRevoke, chance: 10},
{operation: CompareAndSet, chance: 20},
}},
}
HighTrafficPut = trafficConfig{
minimalQPS: 200,
maximalQPS: 1000,
clientCount: 12,
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []opChance{{operation: model.Put, chance: 100}}},
traffic: readWriteSingleKey{keyCount: 4, leaseTTL: DefaultLeaseTTL, writes: []requestChance{{operation: Put, chance: 100}}},
}
)
@ -193,13 +199,14 @@ func patchOperationBasedOnWatchEvents(operations []porcupine.Operation, watchEve
lastObservedOperation := lastOperationObservedInWatch(operations, persisted)
for _, op := range operations {
request := op.Input.(model.EtcdRequest)
resp := op.Output.(model.EtcdResponse)
if resp.Err == nil || op.Call > lastObservedOperation.Call {
// No need to patch successfully requests and cannot patch requests outside observed window.
if resp.Err == nil || op.Call > lastObservedOperation.Call || request.Type != model.Txn {
// Cannot patch those requests.
newOperations = append(newOperations, op)
continue
}
event, hasUniqueWriteOperation := matchWatchEvent(op, persisted)
event := matchWatchEvent(request.Txn, persisted)
if event != nil {
// Set revision and time based on watchEvent.
op.Return = event.Time.UnixNano()
@ -210,7 +217,7 @@ func patchOperationBasedOnWatchEvents(operations []porcupine.Operation, watchEve
newOperations = append(newOperations, op)
continue
}
if hasWriteOperation(op) && !hasUniqueWriteOperation {
if hasNonUniqueWriteOperation(request.Txn) && !hasUniqueWriteOperation(request.Txn) {
// Leave operation as it is as we cannot match non-unique operations to watch events.
newOperations = append(newOperations, op)
continue
@ -224,7 +231,11 @@ func lastOperationObservedInWatch(operations []porcupine.Operation, watchEvents
var maxCallTime int64
var lastOperation porcupine.Operation
for _, op := range operations {
event, _ := matchWatchEvent(op, watchEvents)
request := op.Input.(model.EtcdRequest)
if request.Type != model.Txn {
continue
}
event := matchWatchEvent(request.Txn, watchEvents)
if event != nil && op.Call > maxCallTime {
maxCallTime = op.Call
lastOperation = op
@ -233,33 +244,35 @@ func lastOperationObservedInWatch(operations []porcupine.Operation, watchEvents
return lastOperation
}
func matchWatchEvent(op porcupine.Operation, watchEvents map[model.EtcdOperation]watchEvent) (event *watchEvent, hasUniqueWriteOperation bool) {
request := op.Input.(model.EtcdRequest)
func matchWatchEvent(request *model.TxnRequest, watchEvents map[model.EtcdOperation]watchEvent) *watchEvent {
for _, etcdOp := range request.Ops {
if model.IsWrite(etcdOp.Type) && model.IsUnique(etcdOp.Type) {
// We expect all put to be unique as they write unique value.
hasUniqueWriteOperation = true
opType := etcdOp.Type
if opType == model.PutWithLease {
opType = model.Put
}
if etcdOp.Type == model.Put {
// Remove LeaseID which is not exposed in watch.
event, ok := watchEvents[model.EtcdOperation{
Type: opType,
Type: etcdOp.Type,
Key: etcdOp.Key,
Value: etcdOp.Value,
}]
if ok {
return &event, hasUniqueWriteOperation
return &event
}
}
}
return nil, hasUniqueWriteOperation
return nil
}
func hasWriteOperation(op porcupine.Operation) bool {
request := op.Input.(model.EtcdRequest)
func hasNonUniqueWriteOperation(request *model.TxnRequest) bool {
for _, etcdOp := range request.Ops {
if model.IsWrite(etcdOp.Type) {
if etcdOp.Type == model.Put || etcdOp.Type == model.Delete {
return true
}
}
return false
}
func hasUniqueWriteOperation(request *model.TxnRequest) bool {
for _, etcdOp := range request.Ops {
if etcdOp.Type == model.Put {
return true
}
}

26
tests/linearizability/model/history.go
View File

@ -194,11 +194,11 @@ func (h *AppendableHistory) appendFailed(request EtcdRequest, start time.Time, e
}
func getRequest(key string) EtcdRequest {
return EtcdRequest{Ops: []EtcdOperation{{Type: Get, Key: key}}}
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Get, Key: key}}}}
}
func getResponse(value string, revision int64) EtcdResponse {
return EtcdResponse{OpsResult: []EtcdOperationResult{{Value: value}}, Revision: revision}
return EtcdResponse{Txn: &TxnResponse{OpsResult: []EtcdOperationResult{{Value: value}}}, Revision: revision}
}
func failedResponse(err error) EtcdResponse {
@ -210,23 +210,23 @@ func unknownResponse(revision int64) EtcdResponse {
}
func putRequest(key, value string) EtcdRequest {
return EtcdRequest{Ops: []EtcdOperation{{Type: Put, Key: key, Value: value}}}
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Put, Key: key, Value: value}}}}
}
func putResponse(revision int64) EtcdResponse {
return EtcdResponse{OpsResult: []EtcdOperationResult{{}}, Revision: revision}
return EtcdResponse{Txn: &TxnResponse{OpsResult: []EtcdOperationResult{{}}}, Revision: revision}
}
func deleteRequest(key string) EtcdRequest {
return EtcdRequest{Ops: []EtcdOperation{{Type: Delete, Key: key}}}
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Delete, Key: key}}}}
}
func deleteResponse(deleted int64, revision int64) EtcdResponse {
return EtcdResponse{OpsResult: []EtcdOperationResult{{Deleted: deleted}}, Revision: revision}
return EtcdResponse{Txn: &TxnResponse{OpsResult: []EtcdOperationResult{{Deleted: deleted}}}, Revision: revision}
}
func txnRequest(key, expectValue, newValue string) EtcdRequest {
return EtcdRequest{Conds: []EtcdCondition{{Key: key, ExpectedValue: expectValue}}, Ops: []EtcdOperation{{Type: Put, Key: key, Value: newValue}}}
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Conds: []EtcdCondition{{Key: key, ExpectedValue: expectValue}}, Ops: []EtcdOperation{{Type: Put, Key: key, Value: newValue}}}}
}
func txnResponse(succeeded bool, revision int64) EtcdResponse {
@ -234,27 +234,27 @@ func txnResponse(succeeded bool, revision int64) EtcdResponse {
if succeeded {
result = []EtcdOperationResult{{}}
}
return EtcdResponse{OpsResult: result, TxnResult: !succeeded, Revision: revision}
return EtcdResponse{Txn: &TxnResponse{OpsResult: result, TxnResult: !succeeded}, Revision: revision}
}
func putWithLeaseRequest(key, value string, leaseID int64) EtcdRequest {
return EtcdRequest{Ops: []EtcdOperation{{Type: PutWithLease, Key: key, Value: value, LeaseID: leaseID}}}
return EtcdRequest{Type: Txn, Txn: &TxnRequest{Ops: []EtcdOperation{{Type: Put, Key: key, Value: value, LeaseID: leaseID}}}}
}
func leaseGrantRequest(leaseID int64) EtcdRequest {
return EtcdRequest{Ops: []EtcdOperation{{Type: LeaseGrant, LeaseID: leaseID}}}
return EtcdRequest{Type: LeaseGrant, LeaseGrant: &LeaseGrantRequest{LeaseID: leaseID}}
}
func leaseGrantResponse(revision int64) EtcdResponse {
return EtcdResponse{OpsResult: []EtcdOperationResult{{}}, Revision: revision}
return EtcdResponse{LeaseGrant: &LeaseGrantReponse{}, Revision: revision}
}
func leaseRevokeRequest(leaseID int64) EtcdRequest {
return EtcdRequest{Ops: []EtcdOperation{{Type: LeaseRevoke, LeaseID: leaseID}}}
return EtcdRequest{Type: LeaseRevoke, LeaseRevoke: &LeaseRevokeRequest{LeaseID: leaseID}}
}
func leaseRevokeResponse(revision int64) EtcdResponse {
return EtcdResponse{OpsResult: []EtcdOperationResult{{}}, Revision: revision}
return EtcdResponse{LeaseRevoke: &LeaseRevokeResponse{}, Revision: revision}
}
type History struct {

302
tests/linearizability/model/model.go
View File

@ -17,22 +17,17 @@ package model
import (
"encoding/json"
"fmt"
"github.com/anishathalye/porcupine"
"reflect"
"strings"
"github.com/anishathalye/porcupine"
)
type OperationType string
const (
Get OperationType = "get"
Put OperationType = "put"
Delete OperationType = "delete"
Txn OperationType = "txn"
PutWithLease OperationType = "putWithLease"
LeaseGrant OperationType = "leaseGrant"
LeaseRevoke OperationType = "leaseRevoke"
Get OperationType = "get"
Put OperationType = "put"
Delete OperationType = "delete"
)
var Etcd = porcupine.Model{
@ -57,15 +52,22 @@ var Etcd = porcupine.Model{
},
}
func IsWrite(t OperationType) bool {
return t == Put || t == Delete || t == PutWithLease || t == LeaseRevoke || t == LeaseGrant
}
type RequestType string
func IsUnique(t OperationType) bool {
return t == Put || t == PutWithLease
}
const (
Txn RequestType = "txn"
LeaseGrant RequestType = "leaseGrant"
LeaseRevoke RequestType = "leaseRevoke"
)
type EtcdRequest struct {
Type RequestType
LeaseGrant *LeaseGrantRequest
LeaseRevoke *LeaseRevokeRequest
Txn *TxnRequest
}
type TxnRequest struct {
Conds []EtcdCondition
Ops []EtcdOperation
}
@ -82,14 +84,32 @@ type EtcdOperation struct {
LeaseID int64
}
type LeaseGrantRequest struct {
LeaseID int64
}
type LeaseRevokeRequest struct {
LeaseID int64
}
type EtcdResponse struct {
Err error
Revision int64
ResultUnknown bool
TxnResult bool
OpsResult []EtcdOperationResult
Txn *TxnResponse
LeaseGrant *LeaseGrantReponse
LeaseRevoke *LeaseRevokeResponse
}
type TxnResponse struct {
TxnResult bool
OpsResult []EtcdOperationResult
}
type LeaseGrantReponse struct {
LeaseID int64
}
type LeaseRevokeResponse struct{}
func Match(r1, r2 EtcdResponse) bool {
return ((r1.ResultUnknown || r2.ResultUnknown) && (r1.Revision == r2.Revision)) || reflect.DeepEqual(r1, r2)
}
@ -115,12 +135,38 @@ type EtcdState struct {
}
func describeEtcdRequestResponse(request EtcdRequest, response EtcdResponse) string {
prefix := describeEtcdOperations(request.Ops)
if len(request.Conds) != 0 {
prefix = fmt.Sprintf("if(%s).then(%s)", describeEtcdConditions(request.Conds), prefix)
}
return fmt.Sprintf("%s -> %s", describeEtcdRequest(request), describeEtcdResponse(request, response))
}
return fmt.Sprintf("%s -> %s", prefix, describeEtcdResponse(request.Ops, response))
func describeEtcdResponse(request EtcdRequest, response EtcdResponse) string {
if response.Err != nil {
return fmt.Sprintf("err: %q", response.Err)
}
if response.ResultUnknown {
return fmt.Sprintf("unknown, rev: %d", response.Revision)
}
if request.Type == Txn {
return fmt.Sprintf("%s, rev: %d", describeTxnResponse(request.Txn, response.Txn), response.Revision)
} else {
return fmt.Sprintf("ok, rev: %d", response.Revision)
}
}
func describeEtcdRequest(request EtcdRequest) string {
switch request.Type {
case Txn:
describeOperations := describeEtcdOperations(request.Txn.Ops)
if len(request.Txn.Conds) != 0 {
return fmt.Sprintf("if(%s).then(%s)", describeEtcdConditions(request.Txn.Conds), describeOperations)
}
return describeOperations
case LeaseGrant:
return fmt.Sprintf("leaseGrant(%d)", request.LeaseGrant.LeaseID)
case LeaseRevoke:
return fmt.Sprintf("leaseRevoke(%d)", request.LeaseRevoke.LeaseID)
default:
return fmt.Sprintf("<! unknown request type: %q !>", request.Type)
}
}
func describeEtcdConditions(conds []EtcdCondition) string {
@ -139,21 +185,14 @@ func describeEtcdOperations(ops []EtcdOperation) string {
return strings.Join(opsDescription, ", ")
}
func describeEtcdResponse(ops []EtcdOperation, response EtcdResponse) string {
if response.Err != nil {
return fmt.Sprintf("err: %q", response.Err)
}
if response.ResultUnknown {
return fmt.Sprintf("unknown, rev: %d", response.Revision)
}
func describeTxnResponse(request *TxnRequest, response *TxnResponse) string {
if response.TxnResult {
return fmt.Sprintf("txn failed, rev: %d", response.Revision)
return fmt.Sprintf("txn failed")
}
respDescription := make([]string, len(response.OpsResult))
for i := range response.OpsResult {
respDescription[i] = describeEtcdOperationResponse(ops[i].Type, response.OpsResult[i])
respDescription[i] = describeEtcdOperationResponse(request.Ops[i].Type, response.OpsResult[i])
}
respDescription = append(respDescription, fmt.Sprintf("rev: %d", response.Revision))
return strings.Join(respDescription, ", ")
}
@ -162,17 +201,12 @@ func describeEtcdOperation(op EtcdOperation) string {
case Get:
return fmt.Sprintf("get(%q)", op.Key)
case Put:
return fmt.Sprintf("put(%q, %q)", op.Key, op.Value)
if op.LeaseID != 0 {
return fmt.Sprintf("put(%q, %q, %d)", op.Key, op.Value, op.LeaseID)
}
return fmt.Sprintf("put(%q, %q, nil)", op.Key, op.Value)
case Delete:
return fmt.Sprintf("delete(%q)", op.Key)
case Txn:
return "<! unsupported: nested transaction !>"
case LeaseGrant:
return fmt.Sprintf("leaseGrant(%d)", op.LeaseID)
case LeaseRevoke:
return fmt.Sprintf("leaseRevoke(%d)", op.LeaseID)
case PutWithLease:
return fmt.Sprintf("putWithLease(%q, %q, %d)", op.Key, op.Value, op.LeaseID)
default:
return fmt.Sprintf("<! unknown op: %q !>", op.Type)
}
@ -189,14 +223,6 @@ func describeEtcdOperationResponse(op OperationType, resp EtcdOperationResult) s
return fmt.Sprintf("ok")
case Delete:
return fmt.Sprintf("deleted: %d", resp.Deleted)
case Txn:
return "<! unsupported: nested transaction !>"
case LeaseGrant:
return fmt.Sprintf("ok")
case LeaseRevoke:
return fmt.Sprintf("ok")
case PutWithLease:
return fmt.Sprintf("ok")
default:
return fmt.Sprintf("<! unknown op: %q !>", op)
}
@ -226,31 +252,34 @@ func initState(request EtcdRequest, response EtcdResponse) EtcdState {
KeyLeases: map[string]int64{},
Leases: map[int64]EtcdLease{},
}
if response.TxnResult {
return state
}
for i, op := range request.Ops {
opResp := response.OpsResult[i]
switch op.Type {
case Get:
if opResp.Value != "" {
state.KeyValues[op.Key] = opResp.Value
}
case Put:
state.KeyValues[op.Key] = op.Value
case Delete:
case PutWithLease:
//nop here since lease wont be there
case LeaseGrant:
lease := EtcdLease{
LeaseID: op.LeaseID,
Keys: map[string]struct{}{},
}
state.Leases[op.LeaseID] = lease
case LeaseRevoke:
default:
panic("Unknown operation")
switch request.Type {
case Txn:
if response.Txn.TxnResult {
return state
}
for i, op := range request.Txn.Ops {
opResp := response.Txn.OpsResult[i]
switch op.Type {
case Get:
if opResp.Value != "" {
state.KeyValues[op.Key] = opResp.Value
}
case Put:
state.KeyValues[op.Key] = op.Value
case Delete:
default:
panic("Unknown operation")
}
}
case LeaseGrant:
lease := EtcdLease{
LeaseID: request.LeaseGrant.LeaseID,
Keys: map[string]struct{}{},
}
state.Leases[request.LeaseGrant.LeaseID] = lease
case LeaseRevoke:
default:
panic(fmt.Sprintf("Unknown request type: %v", request.Type))
}
return state
}
@ -278,81 +307,84 @@ func applyRequest(states PossibleStates, request EtcdRequest, response EtcdRespo
// applyRequestToSingleState handles a successful request, returning updated state and response it would generate.
func applyRequestToSingleState(s EtcdState, request EtcdRequest) (EtcdState, EtcdResponse) {
success := true
for _, cond := range request.Conds {
if val := s.KeyValues[cond.Key]; val != cond.ExpectedValue {
success = false
break
}
}
if !success {
return s, EtcdResponse{Revision: s.Revision, TxnResult: true}
}
newKVs := map[string]string{}
for k, v := range s.KeyValues {
newKVs[k] = v
}
s.KeyValues = newKVs
opResp := make([]EtcdOperationResult, len(request.Ops))
increaseRevision := false
for i, op := range request.Ops {
switch op.Type {
case Get:
opResp[i].Value = s.KeyValues[op.Key]
case Put:
s.KeyValues[op.Key] = op.Value
increaseRevision = true
s = detachFromOldLease(s, op.Key)
case Delete:
if _, ok := s.KeyValues[op.Key]; ok {
delete(s.KeyValues, op.Key)
increaseRevision = true
s = detachFromOldLease(s, op.Key)
opResp[i].Deleted = 1
switch request.Type {
case Txn:
success := true
for _, cond := range request.Txn.Conds {
if val := s.KeyValues[cond.Key]; val != cond.ExpectedValue {
success = false
break
}
case PutWithLease:
if _, ok := s.Leases[op.LeaseID]; ok {
//handle put op.
}
if !success {
return s, EtcdResponse{Revision: s.Revision, Txn: &TxnResponse{TxnResult: true}}
}
opResp := make([]EtcdOperationResult, len(request.Txn.Ops))
increaseRevision := false
for i, op := range request.Txn.Ops {
switch op.Type {
case Get:
opResp[i].Value = s.KeyValues[op.Key]
case Put:
_, leaseExists := s.Leases[op.LeaseID]
if op.LeaseID != 0 && !leaseExists {
break
}
s.KeyValues[op.Key] = op.Value
increaseRevision = true
s = detachFromOldLease(s, op.Key)
s = attachToNewLease(s, op.LeaseID, op.Key)
}
case LeaseRevoke:
//Delete the keys attached to the lease
keyDeleted := false
for key, _ := range s.Leases[op.LeaseID].Keys {
//same as delete.
if _, ok := s.KeyValues[key]; ok {
if !keyDeleted {
keyDeleted = true
}
delete(s.KeyValues, key)
delete(s.KeyLeases, key)
if leaseExists {
s = attachToNewLease(s, op.LeaseID, op.Key)
}
case Delete:
if _, ok := s.KeyValues[op.Key]; ok {
delete(s.KeyValues, op.Key)
increaseRevision = true
s = detachFromOldLease(s, op.Key)
opResp[i].Deleted = 1
}
default:
panic("unsupported operation")
}
//delete the lease
delete(s.Leases, op.LeaseID)
if keyDeleted {
increaseRevision = true
}
case LeaseGrant:
lease := EtcdLease{
LeaseID: op.LeaseID,
Keys: map[string]struct{}{},
}
s.Leases[op.LeaseID] = lease
default:
panic("unsupported operation")
}
if increaseRevision {
s.Revision += 1
}
return s, EtcdResponse{Txn: &TxnResponse{OpsResult: opResp}, Revision: s.Revision}
case LeaseGrant:
lease := EtcdLease{
LeaseID: request.LeaseGrant.LeaseID,
Keys: map[string]struct{}{},
}
s.Leases[request.LeaseGrant.LeaseID] = lease
return s, EtcdResponse{Revision: s.Revision, LeaseGrant: &LeaseGrantReponse{}}
case LeaseRevoke:
//Delete the keys attached to the lease
keyDeleted := false
for key, _ := range s.Leases[request.LeaseRevoke.LeaseID].Keys {
//same as delete.
if _, ok := s.KeyValues[key]; ok {
if !keyDeleted {
keyDeleted = true
}
delete(s.KeyValues, key)
delete(s.KeyLeases, key)
}
}
//delete the lease
delete(s.Leases, request.LeaseRevoke.LeaseID)
if keyDeleted {
s.Revision += 1
}
return s, EtcdResponse{Revision: s.Revision, LeaseRevoke: &LeaseRevokeResponse{}}
default:
panic(fmt.Sprintf("Unknown request type: %v", request.Type))
}
if increaseRevision {
s.Revision += 1
}
return s, EtcdResponse{OpsResult: opResp, Revision: s.Revision}
}
func detachFromOldLease(s EtcdState, key string) EtcdState {

17
tests/linearizability/model/model_test.go
View File

@ -562,17 +562,22 @@ func TestModelDescribe(t *testing.T) {
{
req: putRequest("key3", "3"),
resp: putResponse(3),
expectDescribe: `put("key3", "3") -> ok, rev: 3`,
expectDescribe: `put("key3", "3", nil) -> ok, rev: 3`,
},
{
req: putWithLeaseRequest("key3b", "3b", 3),
resp: putResponse(3),
expectDescribe: `put("key3b", "3b", 3) -> ok, rev: 3`,
},
{
req: putRequest("key4", "4"),
resp: failedResponse(errors.New("failed")),
expectDescribe: `put("key4", "4") -> err: "failed"`,
expectDescribe: `put("key4", "4", nil) -> err: "failed"`,
},
{
req: putRequest("key4b", "4b"),
resp: unknownResponse(42),
expectDescribe: `put("key4b", "4b") -> unknown, rev: 42`,
expectDescribe: `put("key4b", "4b", nil) -> unknown, rev: 42`,
},
{
req: deleteRequest("key5"),
@ -587,17 +592,17 @@ func TestModelDescribe(t *testing.T) {
{
req: txnRequest("key7", "7", "77"),
resp: txnResponse(false, 7),
expectDescribe: `if(key7=="7").then(put("key7", "77")) -> txn failed, rev: 7`,
expectDescribe: `if(key7=="7").then(put("key7", "77", nil)) -> txn failed, rev: 7`,
},
{
req: txnRequest("key8", "8", "88"),
resp: txnResponse(true, 8),
expectDescribe: `if(key8=="8").then(put("key8", "88")) -> ok, rev: 8`,
expectDescribe: `if(key8=="8").then(put("key8", "88", nil)) -> ok, rev: 8`,
},
{
req: txnRequest("key9", "9", "99"),
resp: failedResponse(errors.New("failed")),
expectDescribe: `if(key9=="9").then(put("key9", "99")) -> err: "failed"`,
expectDescribe: `if(key9=="9").then(put("key9", "99", nil)) -> err: "failed"`,
},
}
for _, tc := range tcs {

34
tests/linearizability/traffic.go
View File

@ -24,7 +24,6 @@ import (
"go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/tests/v3/linearizability/identity"
"go.etcd.io/etcd/tests/v3/linearizability/model"
)
var (
@ -32,18 +31,29 @@ var (
RequestTimeout = 40 * time.Millisecond
)
type TrafficRequestType string
const (
Get TrafficRequestType = "get"
Put TrafficRequestType = "put"
Delete TrafficRequestType = "delete"
PutWithLease TrafficRequestType = "putWithLease"
LeaseRevoke TrafficRequestType = "leaseRevoke"
CompareAndSet TrafficRequestType = "compareAndSet"
)
type Traffic interface {
Run(ctx context.Context, clientId int, c *recordingClient, limiter *rate.Limiter, ids identity.Provider, lm identity.LeaseIdStorage)
}
type readWriteSingleKey struct {
keyCount int
writes []opChance
writes []requestChance
leaseTTL int64
}
type opChance struct {
operation model.OperationType
type requestChance struct {
operation TrafficRequestType
chance int
}
@ -80,18 +90,18 @@ func (t readWriteSingleKey) Write(ctx context.Context, c *recordingClient, limit
writeCtx, cancel := context.WithTimeout(ctx, RequestTimeout)
var err error
switch t.pickWriteOperation() {
case model.Put:
switch t.pickWriteRequest() {
case Put:
err = c.Put(writeCtx, key, newValue)
case model.Delete:
case Delete:
err = c.Delete(writeCtx, key)
case model.Txn:
case CompareAndSet:
var expectValue string
if len(lastValues) != 0 {
expectValue = string(lastValues[0].Value)
}
err = c.Txn(writeCtx, key, expectValue, newValue)
case model.PutWithLease:
err = c.CompareAndSet(writeCtx, key, expectValue, newValue)
case PutWithLease:
leaseId := lm.LeaseId(cid)
if leaseId == 0 {
leaseId, err = c.LeaseGrant(writeCtx, t.leaseTTL)
@ -105,7 +115,7 @@ func (t readWriteSingleKey) Write(ctx context.Context, c *recordingClient, limit
err = c.PutWithLease(putCtx, key, newValue, leaseId)
putCancel()
}
case model.LeaseRevoke:
case LeaseRevoke:
leaseId := lm.LeaseId(cid)
if leaseId != 0 {
err = c.LeaseRevoke(writeCtx, leaseId)
@ -124,7 +134,7 @@ func (t readWriteSingleKey) Write(ctx context.Context, c *recordingClient, limit
return err
}
func (t readWriteSingleKey) pickWriteOperation() model.OperationType {
func (t readWriteSingleKey) pickWriteRequest() TrafficRequestType {
sum := 0
for _, op := range t.writes {
sum += op.chance