fix inconsistent store pointers between the region-cache and store-ca…

X-Original-Commit: 3c2a4f5cca9cbab6350652da29057bd54cd90874

Author: ClamChowderTiDB

internal/locate/region_cache.go

@@ -348,7 +348,7 @@ func newRegion(bo *retry.Backoffer, c *RegionCache, pdRegion *pd.Region) (*Regio
 			leaderAccessIdx = AccessIndex(len(rs.accessIndex[tiKVOnly]))
 		}
 		availablePeers = append(availablePeers, p)
-		switch store.storeType {
+		switch store.StoreType() {
 		case tikvrpc.TiKV:
 			rs.accessIndex[tiKVOnly] = append(rs.accessIndex[tiKVOnly], len(rs.stores))
 		case tikvrpc.TiFlash:
@@ -702,7 +702,7 @@ func NewRegionCache(pdClient pd.Client, opt ...RegionCacheOpt) *RegionCache {
 	c.bg.scheduleWithTrigger(func(ctx context.Context, t time.Time) bool {
 		// check and resolve normal stores periodically by default.
 		filter := func(state resolveState) bool {
-			return state != unresolved && state != tombstone && state != deleted
+			return state != unresolved && state != tombstone
 		}
 		if t.IsZero() {
 			// check and resolve needCheck stores because it's triggered by a CheckStoreEvent this time.
@@ -757,13 +757,9 @@ func refreshFullStoreList(ctx context.Context, stores storeCache) {
 			continue
 		}
 		s := stores.getOrInsertDefault(store.GetId())
-		// TODO: maybe refactor this, together with other places initializing Store
-		s.addr = addr
-		s.peerAddr = store.GetPeerAddress()
-		s.saddr = store.GetStatusAddress()
-		s.storeType = tikvrpc.GetStoreTypeByMeta(store)
-		s.labels = store.GetLabels()
-		s.changeResolveStateTo(unresolved, resolved)
+		if err := s.initResolveLite(store); err != nil {
+			continue
+		}
 	}
 }
 
@@ -805,7 +801,7 @@ func (c *RegionCache) checkAndResolve(needCheckStores []*Store, needCheck func(*
 
 	needCheckStores = c.stores.filter(needCheckStores, needCheck)
 	for _, store := range needCheckStores {
-		_, err := store.reResolve(c.stores, c.bg)
+		_, err := store.reResolve(c.stores)
 		tikverr.Log(err)
 	}
 	return needCheckStores
@@ -841,12 +837,12 @@ type RPCContext struct {
 func (c *RPCContext) String() string {
 	var runStoreType string
 	if c.Store != nil {
-		runStoreType = c.Store.storeType.Name()
+		runStoreType = c.Store.StoreType().Name()
 	}
 	res := fmt.Sprintf("region ID: %d, meta: %s, peer: %s, addr: %s, idx: %d, reqStoreType: %s, runStoreType: %s",
 		c.Region.GetID(), c.Meta, c.Peer, c.Addr, c.AccessIdx, c.AccessMode, runStoreType)
 	if c.ProxyStore != nil {
-		res += fmt.Sprintf(", proxy store id: %d, proxy addr: %s", c.ProxyStore.storeID, c.ProxyStore.addr)
+		res += fmt.Sprintf(", proxy store id: %d, proxy addr: %s", c.ProxyStore.storeID, c.ProxyStore.GetAddr())
 	}
 	return res
 }
@@ -942,7 +938,7 @@ func (c *RegionCache) GetTiKVRPCContext(bo *retry.Backoffer, id RegionVerID, rep
 		cachedRegion.invalidate(Other)
 		logutil.Logger(bo.GetCtx()).Info("invalidate current region, because others failed on same store",
 			zap.Uint64("region", id.GetID()),
-			zap.String("store", store.addr))
+			zap.String("store", store.GetAddr()))
 		return nil, nil
 	}
 
@@ -1011,7 +1007,7 @@ func (c *RegionCache) GetAllValidTiFlashStores(id RegionVerID, currentStore *Sto
 		if storeFailEpoch != regionStore.storeEpochs[storeIdx] {
 			continue
 		}
-		if !labelFilter(store.labels) {
+		if !labelFilter(store.GetLabels()) {
 			continue
 		}
 		allStores = append(allStores, store.storeID)
@@ -1046,7 +1042,7 @@ func (c *RegionCache) GetTiFlashRPCContext(bo *retry.Backoffer, id RegionVerID,
 	for i := 0; i < regionStore.accessStoreNum(tiFlashOnly); i++ {
 		accessIdx := AccessIndex((sIdx + i) % regionStore.accessStoreNum(tiFlashOnly))
 		storeIdx, store := regionStore.accessStore(tiFlashOnly, accessIdx)
-		if !labelFilter(store.labels) {
+		if !labelFilter(store.GetLabels()) {
 			continue
 		}
 		addr, err := c.getStoreAddr(bo, cachedRegion, store)
@@ -1058,7 +1054,7 @@ func (c *RegionCache) GetTiFlashRPCContext(bo *retry.Backoffer, id RegionVerID,
 			return nil, nil
 		}
 		if store.getResolveState() == needCheck {
-			_, err := store.reResolve(c.stores, c.bg)
+			_, err := store.reResolve(c.stores)
 			tikverr.Log(err)
 		}
 		regionStore.workTiFlashIdx.Store(int32(accessIdx))
@@ -1068,7 +1064,7 @@ func (c *RegionCache) GetTiFlashRPCContext(bo *retry.Backoffer, id RegionVerID,
 			cachedRegion.invalidate(Other)
 			logutil.Logger(bo.GetCtx()).Info("invalidate current region, because others failed on same store",
 				zap.Uint64("region", id.GetID()),
-				zap.String("store", store.addr))
+				zap.String("store", store.GetAddr()))
 			// TiFlash will always try to find out a valid peer, avoiding to retry too many times.
 			continue
 		}
@@ -1602,7 +1598,7 @@ func (c *RegionCache) markRegionNeedBeRefill(s *Store, storeIdx int, rs *regionS
 	// invalidate regions in store.
 	epoch := rs.storeEpochs[storeIdx]
 	if atomic.CompareAndSwapUint32(&s.epoch, epoch, epoch+1) {
-		logutil.BgLogger().Info("mark store's regions need be refill", zap.String("store", s.addr))
+		logutil.BgLogger().Info("mark store's regions need be refill", zap.String("store", s.GetAddr()))
 		incEpochStoreIdx = storeIdx
 		metrics.RegionCacheCounterWithInvalidateStoreRegionsOK.Inc()
 	}
@@ -1993,14 +1989,14 @@ func (c *RegionCache) searchCachedRegionByID(regionID uint64) (*Region, bool) {
 // GetStoresByType gets stores by type `typ`
 func (c *RegionCache) GetStoresByType(typ tikvrpc.EndpointType) []*Store {
 	return c.stores.filter(nil, func(s *Store) bool {
-		return s.getResolveState() == resolved && s.storeType == typ
+		return s.getResolveState() == resolved && s.StoreType() == typ
 	})
 }
 
 // GetAllStores gets TiKV and TiFlash stores.
 func (c *RegionCache) GetAllStores() []*Store {
 	return c.stores.filter(nil, func(s *Store) bool {
-		return s.getResolveState() == resolved && (s.storeType == tikvrpc.TiKV || s.storeType == tikvrpc.TiFlash)
+		return s.getResolveState() == resolved && (s.StoreType() == tikvrpc.TiKV || s.StoreType() == tikvrpc.TiFlash)
 	})
 }
 
@@ -2448,14 +2444,11 @@ func (c *RegionCache) getStoreAddr(bo *retry.Backoffer, region *Region, store *S
 	state := store.getResolveState()
 	switch state {
 	case resolved, needCheck:
-		addr = store.addr
+		addr = store.GetAddr()
 		return
 	case unresolved:
 		addr, err = store.initResolve(bo, c.stores)
 		return
-	case deleted:
-		addr = c.changeToActiveStore(region, store.storeID)
-		return
 	case tombstone:
 		return "", nil
 	default:
@@ -2464,7 +2457,7 @@ func (c *RegionCache) getStoreAddr(bo *retry.Backoffer, region *Region, store *S
 }
 
 func (c *RegionCache) getProxyStore(region *Region, store *Store, rs *regionStore, workStoreIdx AccessIndex) (proxyStore *Store, proxyAccessIdx AccessIndex, proxyStoreIdx int) {
-	if !c.enableForwarding || store.storeType != tikvrpc.TiKV || store.getLivenessState() == reachable {
+	if !c.enableForwarding || store.StoreType() != tikvrpc.TiKV || store.getLivenessState() == reachable {
 		return
 	}
 
@@ -2505,29 +2498,6 @@ func (c *RegionCache) getProxyStore(region *Region, store *Store, rs *regionStor
 	return nil, 0, 0
 }
 
-// changeToActiveStore replace the deleted store in the region by an up-to-date store in the stores map.
-// The order is guaranteed by reResolve() which adds the new store before marking old store deleted.
-func (c *RegionCache) changeToActiveStore(region *Region, storeID uint64) (addr string) {
-	store, _ := c.stores.get(storeID)
-	for {
-		oldRegionStore := region.getStore()
-		newRegionStore := oldRegionStore.clone()
-		newRegionStore.stores = make([]*Store, 0, len(oldRegionStore.stores))
-		for _, s := range oldRegionStore.stores {
-			if s.storeID == store.storeID {
-				newRegionStore.stores = append(newRegionStore.stores, store)
-			} else {
-				newRegionStore.stores = append(newRegionStore.stores, s)
-			}
-		}
-		if region.compareAndSwapStore(oldRegionStore, newRegionStore) {
-			break
-		}
-	}
-	addr = store.addr
-	return
-}
-
 // OnBucketVersionNotMatch removes the old buckets meta if the version is stale.
 func (c *RegionCache) OnBucketVersionNotMatch(ctx *RPCContext, version uint64, keys [][]byte) {
 	r := c.GetCachedRegionWithRLock(ctx.Region)
@@ -2586,7 +2556,7 @@ func (c *RegionCache) OnRegionEpochNotMatch(bo *retry.Backoffer, ctx *RPCContext
 			return false, err
 		}
 		var initLeaderStoreID uint64
-		if ctx.Store.storeType == tikvrpc.TiFlash {
+		if ctx.Store.StoreType() == tikvrpc.TiFlash {
 			initLeaderStoreID = region.findElectableStoreID()
 		} else {
 			initLeaderStoreID = ctx.Store.storeID
@@ -2619,7 +2589,7 @@ func (c *RegionCache) PDClient() pd.Client {
 // stores so that users won't be bothered by tombstones. (related issue: https://github.com/pingcap/tidb/issues/46602)
 func (c *RegionCache) GetTiFlashStores(labelFilter LabelFilter) []*Store {
 	return c.stores.filter(nil, func(s *Store) bool {
-		return s.storeType == tikvrpc.TiFlash && labelFilter(s.labels) && s.getResolveState() == resolved
+		return s.StoreType() == tikvrpc.TiFlash && labelFilter(s.GetLabels()) && s.getResolveState() == resolved
 	})
 }
 
@@ -3085,7 +3055,7 @@ func (c *RegionCache) checkAndUpdateStoreHealthStatus(ctx context.Context, now t
 func (c *RegionCache) reportStoreReplicaFlows() {
 	c.stores.forEach(func(store *Store) {
 		for destType := toLeader; destType < numReplicaFlowsType; destType++ {
-			metrics.TiKVPreferLeaderFlowsGauge.WithLabelValues(destType.String(), store.addr).Set(float64(store.getReplicaFlowsStats(destType)))
+			metrics.TiKVPreferLeaderFlowsGauge.WithLabelValues(destType.String(), store.GetAddr()).Set(float64(store.getReplicaFlowsStats(destType)))
 			store.resetReplicaFlowsStats(destType)
 		}
 	})