diff --git a/internal/locate/region_cache.go b/internal/locate/region_cache.go index 38b7727e3c..9403adacf9 100644 --- a/internal/locate/region_cache.go +++ b/internal/locate/region_cache.go @@ -780,6 +780,13 @@ func (c *RegionCache) Close() { c.bg.shutdown(true) } +// IsBackgroundRunnerClosed returns whether RegionCache's background runner context has been canceled. +// +// It is intended for tests/debugging only. +func (c *RegionCache) IsBackgroundRunnerClosed() bool { + return c.bg.closed() +} + // checkAndResolve checks and resolve addr of failed stores. // this method isn't thread-safe and only be used by one goroutine. func (c *RegionCache) checkAndResolve(needCheckStores []*Store, needCheck func(*Store) bool) []*Store { @@ -1827,18 +1834,6 @@ func (c *RegionCache) BatchLoadRegionsFromKey(bo *retry.Backoffer, startKey []by return regions[len(regions)-1].EndKey(), nil } -// AsyncInvalidateCachedRegion marks a cached region for reload on next access -// without invalidating it, so in-flight retries can still proceed. -// It sets needDelayedReloadReady directly, skipping the GC promotion from -// needDelayedReloadPending for faster reload. -func (c *RegionCache) AsyncInvalidateCachedRegion(id RegionVerID) { - cachedRegion := c.GetCachedRegionWithRLock(id) - if cachedRegion == nil { - return - } - cachedRegion.setSyncFlags(needDelayedReloadReady) -} - // InvalidateCachedRegion removes a cached Region. func (c *RegionCache) InvalidateCachedRegion(id RegionVerID) { c.InvalidateCachedRegionWithReason(id, Other) diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go index 9ded6f3169..31b5549723 100644 --- a/internal/locate/region_request3_test.go +++ b/internal/locate/region_request3_test.go @@ -1410,9 +1410,6 @@ func (s *testRegionRequestToThreeStoresSuite) TestDoNotTryUnreachableLeader() { follower, _, _, _ := region.FollowerStorePeer(regionStore, 0, &storeSelectorOp{}) s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) { - if req.StaleRead && addr == follower.addr { - return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{DataIsNotReady: &errorpb.DataIsNotReady{}}}}, nil - } return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{ Value: []byte(addr), }}, nil diff --git a/internal/locate/region_request_state_test.go b/internal/locate/region_request_state_test.go index 18b297e915..cc9fd75074 100644 --- a/internal/locate/region_request_state_test.go +++ b/internal/locate/region_request_state_test.go @@ -303,9 +303,9 @@ func testRegionCacheStaleRead(t *testing.T) { leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z1"}, leaderSuccessReadType: SuccessStaleRead, - followerRegionValid: true, - followerAsyncReload: util.Some(true), - // may async reload region and access it from leader. + followerRegionValid: false, + followerAsyncReload: util.Some(false), + // region is hard-invalidated but the current request succeeds via leader retry. followerSuccessReplica: []string{"z1"}, followerSuccessReadType: SuccessStaleRead, }, @@ -348,7 +348,7 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(true), leaderSuccessReplica: []string{"z2", "z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.None[bool](), followerSuccessReplica: []string{"z2"}, @@ -383,7 +383,7 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z2", "z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z2"}, @@ -408,11 +408,11 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z3"}, - followerSuccessReadType: SuccessFollowerRead, + followerSuccessReadType: SuccessStaleRead, }, { do: leaderServerIsBusy, @@ -421,11 +421,11 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z2", "z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z2", "z3"}, - followerSuccessReadType: SuccessFollowerRead, + followerSuccessReadType: SuccessStaleRead, }, { do: leaderServerIsBusy, @@ -434,11 +434,11 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z3"}, - followerSuccessReadType: SuccessFollowerRead, + followerSuccessReadType: SuccessStaleRead, }, { do: leaderDown, diff --git a/internal/locate/replica_selector.go b/internal/locate/replica_selector.go index 499b0fd10d..65f9947083 100644 --- a/internal/locate/replica_selector.go +++ b/internal/locate/replica_selector.go @@ -31,13 +31,14 @@ import ( type replicaSelector struct { baseReplicaSelector - replicaReadType kv.ReplicaReadType - isStaleRead bool - isReadOnlyReq bool - option storeSelectorOp - target *replica - proxy *replica - attempts int + replicaReadType kv.ReplicaReadType + isStaleRead bool + isReadOnlyReq bool + option storeSelectorOp + target *replica + proxy *replica + attempts int + regionInvalidatedForRetry bool // set when region is hard-invalidated but this selector should still retry on leader } func newReplicaSelector( @@ -98,7 +99,13 @@ func buildTiKVReplicas(region *Region) []*replica { } func (s *replicaSelector) next(bo *retry.Backoffer, req *tikvrpc.Request) (rpcCtx *RPCContext, err error) { - if !s.region.isValid() { + if s.regionInvalidatedForRetry { + // Allow one more attempt on this selector even though the region is + // invalidated. This is set by onRegionNotFound so that the current + // request can retry on the leader while concurrent requests are + // stopped by the hard invalidation. + s.regionInvalidatedForRetry = false + } else if !s.region.isValid() { metrics.TiKVReplicaSelectorFailureCounter.WithLabelValues("invalid").Inc() return nil, nil } @@ -182,17 +189,21 @@ func (s *replicaSelector) nextForReplicaReadMixed(req *tikvrpc.Request) { } s.target = strategy.next(s) if s.target != nil { - if s.isStaleRead && s.attempts == 1 { - // stale-read request first access. - if !s.target.store.IsLabelsMatch(s.option.labels) && s.target.peer.Id != s.region.GetLeaderPeerID() { - // If the target replica's labels is not match and not leader, use replica read. - // This is for compatible with old version. - req.StaleRead = false - req.ReplicaRead = true - } else { - // use stale read. + if s.isStaleRead { + isStaleRead := true + if s.attempts != 1 || (!s.target.store.IsLabelsMatch(s.option.labels) && s.target.peer.Id != s.region.GetLeaderPeerID()) { + // retry or target replica's labels does not match and not leader + if strategy.canSendReplicaRead(s) { + // use replica read. + isStaleRead = false + } + } + if isStaleRead { req.StaleRead = true req.ReplicaRead = false + } else { + req.StaleRead = false + req.ReplicaRead = true } } else { // use replica read only if the target is not leader. @@ -300,6 +311,16 @@ func (s *ReplicaSelectMixedStrategy) next(selector *replicaSelector) *replica { return nil } +func (s *ReplicaSelectMixedStrategy) canSendReplicaRead(selector *replicaSelector) bool { + replicas := selector.replicas + replica := replicas[s.leaderIdx] + if replica.attempts == 0 || replica.hasFlag(deadlineErrUsingConfTimeoutFlag) || replica.hasFlag(serverIsBusyFlag) { + // don't do the replica read if leader is not exhausted or leader has timeout or server busy error. + return false + } + return true +} + func hasDeadlineExceededError(replicas []*replica) bool { for _, replica := range replicas { if replica.hasFlag(deadlineErrUsingConfTimeoutFlag) { @@ -509,11 +530,15 @@ func (s *replicaSelector) onRegionNotFound( leaderIdx := s.region.getStore().workTiKVIdx leader := s.replicas[leaderIdx] if !leader.isExhausted(1, 0) { - // if the request is not sent to leader, we can retry it with leader and invalidate the region cache asynchronously. It helps in the scenario - // where region is split by the leader but not yet created in replica due to replica down. + // if the request is not sent to leader, we can retry it with leader and invalidate the region cache + // immediately. It helps in the scenario where region is split by the leader but not yet created + // in replica due to replica down. + // Hard-invalidate the region so that concurrent requests stop using the stale region, + // but allow this selector to retry on the leader via regionInvalidatedForRetry. req.SetReplicaReadType(kv.ReplicaReadLeader) s.replicaReadType = kv.ReplicaReadLeader - s.regionCache.AsyncInvalidateCachedRegion(ctx.Region) + s.regionInvalidatedForRetry = true + s.regionCache.InvalidateCachedRegion(ctx.Region) return true, nil } s.regionCache.InvalidateCachedRegion(ctx.Region) @@ -546,6 +571,9 @@ func (s *replicaSelector) onServerIsBusy( backoffErr := errors.Errorf("server is busy, ctx: %v", ctx) if s.canFastRetry() { s.addPendingBackoff(store, retry.BoTiKVServerBusy, backoffErr) + if s.target != nil { + s.target.addFlag(serverIsBusyFlag) + } return true, nil } err = bo.Backoff(retry.BoTiKVServerBusy, backoffErr) diff --git a/internal/locate/replica_selector_test.go b/internal/locate/replica_selector_test.go index cb688831fa..0df3ef4c85 100644 --- a/internal/locate/replica_selector_test.go +++ b/internal/locate/replica_selector_test.go @@ -376,7 +376,7 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store2, replica-read: true, stale-read: false}"}, + "{addr: store2, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: nil, backoffCnt: 0, @@ -418,7 +418,7 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}"}, // store2 has DeadLineExceededErr, so don't retry store2 even it is new leader. + "{addr: store3, replica-read: false, stale-read: true}"}, // store2 has DeadLineExceededErr, so don't retry store2 even it is new leader. respErr: "", respRegionError: nil, backoffCnt: 0, @@ -518,8 +518,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { expect: &accessPathResult{ accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}"}, + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: fakeEpochNotMatch, backoffCnt: 1, @@ -562,7 +562,7 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store3, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store2, replica-read: true, stale-read: false}"}, + "{addr: store2, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: nil, backoffCnt: 0, @@ -582,8 +582,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -604,8 +604,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", "{addr: store2, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store3, replica-read: true, stale-read: false}", - "{addr: store1, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", + "{addr: store1, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -625,8 +625,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", "{addr: store2, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store3, replica-read: true, stale-read: false}", - "{addr: store1, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", + "{addr: store1, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -668,8 +668,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { expect: &accessPathResult{ accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -948,7 +948,7 @@ func testReplicaReadAccessPathByCase2(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}"}, + "{addr: store3, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: nil, backoffCnt: 0, @@ -1094,9 +1094,16 @@ func testReplicaReadAccessPathByBasicCase(s *testReplicaSelectorSuite) { backoff = []string{} } if staleRead { - accessPath = []string{ - "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", + if tp == ServerIsBusyErr || tp == ServerIsBusyWithEstimatedWaitMsErr { + accessPath = []string{ + "{addr: store1, replica-read: false, stale-read: true}", + "{addr: store2, replica-read: false, stale-read: true}", + } + } else { + accessPath = []string{ + "{addr: store1, replica-read: false, stale-read: true}", + "{addr: store2, replica-read: true, stale-read: false}", + } } } default: @@ -1149,14 +1156,16 @@ func testReplicaReadAccessPathByBasicCase(s *testReplicaSelectorSuite) { case kv.ReplicaReadLeader: accessPath = []string{"{addr: store1, replica-read: false, stale-read: false}"} case kv.ReplicaReadFollower: - // For RegionNotFoundErr from follower, it will retry on leader + // For RegionNotFoundErr from follower, it will retry on leader. + // The region is hard-invalidated to stop concurrent requests, + // but the current selector bypasses the validity check. if tp == RegionNotFoundErr { accessPath = []string{ "{addr: store2, replica-read: true, stale-read: false}", "{addr: store1, replica-read: false, stale-read: false}", } respRegionError = nil - regionIsValid = true + regionIsValid = false } else { accessPath = []string{"{addr: store2, replica-read: true, stale-read: false}"} } @@ -2014,8 +2023,8 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -2036,8 +2045,8 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", "{addr: store2, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store3, replica-read: true, stale-read: false}", - "{addr: store1, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", + "{addr: store1, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -2146,8 +2155,8 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { expect: &accessPathResult{ accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -2169,7 +2178,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -2191,7 +2200,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -2233,7 +2242,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { beforeRun: func() { /* don't resetStoreState */ }, expect: &accessPathResult{ accessPath: []string{ - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -2762,7 +2771,11 @@ func TestReplicaReadAvoidSlowStore(t *testing.T) { if expectedFirstStore == 3 { // Retry on store 2 which is a follower. // Stale-read mode falls back to replica-read mode. - expectedSecondPath = "{addr: store2, replica-read: true, stale-read: false}" + if staleRead { + expectedSecondPath = "{addr: store2, replica-read: false, stale-read: true}" + } else { + expectedSecondPath = "{addr: store2, replica-read: true, stale-read: false}" + } } else { // Retry in leader read mode will not use replica-read. expectedSecondPath = "{addr: store3, replica-read: false, stale-read: false}" diff --git a/internal/logutil/hex.go b/internal/logutil/hex.go index 1accd622ce..066a496357 100644 --- a/internal/logutil/hex.go +++ b/internal/logutil/hex.go @@ -87,7 +87,7 @@ func prettyPrint(w io.Writer, val reflect.Value) { prettyPrint(w, fv) } fmt.Fprintf(w, "}") - case reflect.Ptr: + case reflect.Pointer: if val.IsNil() { fmt.Fprintf(w, "%v", val.Interface()) } else { diff --git a/tikv/kv.go b/tikv/kv.go index 49066c7921..15f6a9eee7 100644 --- a/tikv/kv.go +++ b/tikv/kv.go @@ -98,6 +98,12 @@ func createEtcdKV(addrs []string, tlsConfig *tls.Config) (*clientv3.Client, erro if err != nil { return nil, errors.WithStack(err) } + ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second) + defer cancel() + if err := cli.Sync(ctx); err != nil { + _ = cli.Close() + return nil, errors.WithStack(err) + } return cli, nil } @@ -442,20 +448,25 @@ func (s *KVStore) Close() error { s.wg.Wait() s.oracle.Close() - s.pdClient.Close() - if s.pdHttpClient != nil { - s.pdHttpClient.Close() + if s.txnLatches != nil { + s.txnLatches.Close() } s.lockResolver.Close() + // Close region cache before closing clients it depends on. Otherwise, its background tasks might still try + // to access PD/TiKV and encounter "grpc: the client connection is closing". + s.regionCache.Close() if err := s.GetTiKVClient().Close(); err != nil { return err } - if s.txnLatches != nil { - s.txnLatches.Close() + if s.pdHttpClient != nil { + s.pdHttpClient.Close() } - s.regionCache.Close() + if _, err := util.EvalFailpoint("checkRegionCacheClosedBeforePDClose"); err == nil && !s.regionCache.IsBackgroundRunnerClosed() { + panic("region cache is not closed before closing pd client") + } + s.pdClient.Close() if err := s.kv.Close(); err != nil { return err diff --git a/tikv/kv_test.go b/tikv/kv_test.go index 8e609c4974..8a7042819d 100644 --- a/tikv/kv_test.go +++ b/tikv/kv_test.go @@ -22,6 +22,7 @@ import ( "testing" "time" + "github.com/pingcap/failpoint" "github.com/pingcap/kvproto/pkg/kvrpcpb" "github.com/pingcap/kvproto/pkg/metapb" "github.com/stretchr/testify/require" @@ -280,3 +281,19 @@ func (s *testKVSuite) TestErrorHalfwayInNewKVStore() { _, err := NewKVStore("TestErrorHalfwayInNewKVStore", s.store.pdClient, NewMockSafePointKV(), &mocktikv.RPCClient{}) require.Error(s.T(), err) } + +func TestKVStoreCloseCheckRegionCacheClosedBeforePDClose(t *testing.T) { + util.EnableFailpoints() + require.NoError(t, failpoint.Enable("tikvclient/checkRegionCacheClosedBeforePDClose", "return(true)")) + t.Cleanup(func() { + require.NoError(t, failpoint.Disable("tikvclient/checkRegionCacheClosedBeforePDClose")) + }) + + client, cluster, pdClient, err := testutils.NewMockTiKV("", nil) + require.NoError(t, err) + _, _, _, _ = mocktikv.BootstrapWithMultiStores(cluster, 1) + + store, err := NewTestTiKVStore(client, pdClient, nil, nil, 0) + require.NoError(t, err) + require.NoError(t, store.Close()) +}