Skip to content

Commit

Permalink
Fix the issue that leader change with new leader info cause invalidSt…
Browse files Browse the repository at this point in the history
…ore state and lead to unnecessary backoff (tikv#1115) (tikv#1337)

* Fix the issue that leader change with nea leader info cause invalidStore state and lead to unnecessary backoff (tikv#1115)

* Fix the issue that leader change with nea leader info cause invalidStore
state and lead to unnecessary backoff

Signed-off-by: MyonKeminta <[email protected]>

* address comments

Signed-off-by: MyonKeminta <[email protected]>

---------

Signed-off-by: MyonKeminta <[email protected]>
Co-authored-by: MyonKeminta <[email protected]>

* Fix build on go1.18

Signed-off-by: MyonKeminta <[email protected]>

---------

Signed-off-by: MyonKeminta <[email protected]>
Co-authored-by: MyonKeminta <[email protected]>
  • Loading branch information
2 people authored and Ari Ekmekji committed Jul 19, 2024
1 parent 4d4dcd3 commit 9e1b9eb
Show file tree
Hide file tree
Showing 2 changed files with 90 additions and 1 deletion.
15 changes: 14 additions & 1 deletion internal/locate/region_request.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,13 @@ import (
"fmt"
"math"
"math/rand"
"reflect"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"unsafe"

"go.uber.org/zap"
"google.golang.org/grpc/codes"
Expand Down Expand Up @@ -1005,6 +1007,17 @@ func (s *replicaSelector) proxyReplica() *replica {
return nil
}

// sliceIdentical checks whether two slices are referencing the same block of memory. Two `nil`s are also considered
// the same.
func sliceIdentical[T any](a, b []T) bool {
if len(a) != len(b) {
return false
}
aHeader := (*reflect.SliceHeader)(unsafe.Pointer(&a))
bHeader := (*reflect.SliceHeader)(unsafe.Pointer(&b))
return aHeader.Data == bHeader.Data
}

func (s *replicaSelector) refreshRegionStore() {
oldRegionStore := s.regionStore
newRegionStore := s.region.getStore()
Expand All @@ -1017,7 +1030,7 @@ func (s *replicaSelector) refreshRegionStore() {
// So we just compare the address here.
// When stores change, we mark this replicaSelector as invalid to let the caller
// recreate a new replicaSelector.
if &oldRegionStore.stores != &newRegionStore.stores {
if !sliceIdentical(oldRegionStore.stores, newRegionStore.stores) {
s.state = &invalidStore{}
return
}
Expand Down
76 changes: 76 additions & 0 deletions internal/locate/region_request3_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,82 @@ func (s *testRegionRequestToThreeStoresSuite) TestSwitchPeerWhenNoLeader() {
resp, _, err := s.regionRequestSender.SendReq(bo, req, loc.Region, time.Second)
s.Nil(err)
s.NotNil(resp)
s.Nil(resp.GetRegionError())
}

func (s *testRegionRequestToThreeStoresSuite) TestSwitchPeerWhenNoLeaderErrorWithNewLeaderInfo() {
cnt := 0
var location *KeyLocation
cli := &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
cnt++
switch cnt {
case 1:
region := s.cache.GetCachedRegionWithRLock(location.Region)
s.NotNil(region)
leaderPeerIdx := int(region.getStore().workTiKVIdx)
peers := region.meta.Peers
// return no leader with new leader info
response = &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
RegionError: &errorpb.Error{NotLeader: &errorpb.NotLeader{
RegionId: req.RegionId,
Leader: peers[(leaderPeerIdx+1)%len(peers)],
}},
}}
case 2:
response = &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
Value: []byte("a"),
}}
default:
return nil, fmt.Errorf("unexpected request")
}
return response, err
}}

req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: []byte("a")}, kvrpcpb.Context{})
req.ReplicaReadType = kv.ReplicaReadLeader
var err error
location, err = s.cache.LocateKey(s.bo, []byte("a"))
s.Nil(err)
s.NotNil(location)
bo := retry.NewBackoffer(context.Background(), 1000)
resp, _, err := NewRegionRequestSender(s.cache, cli).SendReqCtx(bo, req, location.Region, time.Second, tikvrpc.TiKV)
s.Nil(err)
s.NotNil(resp)
regionErr, err := resp.GetRegionError()
s.Nil(err)
s.Nil(regionErr)
// It's unreasoneable to retry in upper layer, such as cop request, the upper layer will need to rebuild cop request and retry, there are some unnecessary overhead.
s.Equal(cnt, 2)
r := s.cache.GetCachedRegionWithRLock(location.Region)
s.True(r.isValid())
}

func (s *testRegionRequestToThreeStoresSuite) TestSliceIdentical() {
a := make([]int, 0)
b := a
s.True(sliceIdentical(a, b))
// This is not guaranteed.
// b = make([]int, 0)
// s.False(sliceIdentical(a, b))

a = append(a, 1, 2, 3)
b = a
s.True(sliceIdentical(a, b))
b = a[:2]
s.False(sliceIdentical(a, b))
b = a[1:]
s.False(sliceIdentical(a, b))
a = a[1:]
s.True(sliceIdentical(a, b))

a = nil
b = nil

s.True(sliceIdentical(a, b))
a = make([]int, 0)
s.False(sliceIdentical(a, b))
a = append(a, 1)
s.False(sliceIdentical(a, b))
}

func (s *testRegionRequestToThreeStoresSuite) loadAndGetLeaderStore() (*Store, string) {
Expand Down

0 comments on commit 9e1b9eb

Please sign in to comment.