From a526d5cab63ba733242ef2e6ca9d87a20fc32782 Mon Sep 17 00:00:00 2001 From: lysu Date: Wed, 22 May 2019 19:47:19 +0800 Subject: [PATCH 1/7] tikv: fix notLeader change failure --- store/tikv/region_cache.go | 33 +++++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/store/tikv/region_cache.go b/store/tikv/region_cache.go index 537113696b36d..a481f5e7ec93e 100644 --- a/store/tikv/region_cache.go +++ b/store/tikv/region_cache.go @@ -447,12 +447,19 @@ func (c *RegionCache) UpdateLeader(regionID RegionVerID, leaderStoreID uint64) { zap.Uint64("leaderStoreID", leaderStoreID)) return } - if !c.switchWorkStore(r, leaderStoreID) { + leaderIdx, found := c.getPeerStoreIndex(r, leaderStoreID) + if !found { logutil.Logger(context.Background()).Debug("regionCache: cannot find peer when updating leader", zap.Uint64("regionID", regionID.GetID()), zap.Uint64("leaderStoreID", leaderStoreID)) r.invalidate() + return + } + rs := r.getStore() + if len(rs.stores) > leaderIdx { + rs.stores[leaderIdx].markAccess(nil, true) } + c.switchWorkIdx(r, leaderIdx) } // insertRegionToCache tries to insert the Region to cache. @@ -849,20 +856,27 @@ func (r *Region) EndKey() []byte { // switchWorkStore switches current store to the one on specific store. It returns // false if no peer matches the storeID. -func (c *RegionCache) switchWorkStore(r *Region, targetStoreID uint64) (switchToTarget bool) { +func (c *RegionCache) switchWorkStore(r *Region, targetStoreID uint64) { + leaderIdx, _ := c.getPeerStoreIndex(r, targetStoreID) + c.switchWorkIdx(r, leaderIdx) + return +} + +func (c *RegionCache) getPeerStoreIndex(r *Region, id uint64) (idx int, found bool) { if len(r.meta.Peers) == 0 { return } - - leaderIdx := 0 for i, p := range r.meta.Peers { - if p.GetStoreId() == targetStoreID { - leaderIdx = i - switchToTarget = true - break + if p.GetStoreId() == id { + idx = i + found = true + return } } + return +} +func (c *RegionCache) switchWorkIdx(r *Region, leaderIdx int) { retry: // switch to new leader. oldRegionStore := r.getStore() @@ -1090,6 +1104,9 @@ retry: if !s.compareAndSwapState(oldState, state) { goto retry } + if notifyCheckCh == nil { + return + } if triggerCheck { select { case notifyCheckCh <- struct{}{}: From d4f539e0d452dc0d49b84ebc0d6c100875d835dc Mon Sep 17 00:00:00 2001 From: lysu Date: Thu, 23 May 2019 14:53:32 +0800 Subject: [PATCH 2/7] only change peer after fail --- store/tikv/region_cache.go | 290 +++++++++----------------------- store/tikv/region_cache_test.go | 167 +++--------------- store/tikv/region_request.go | 8 +- 3 files changed, 100 insertions(+), 365 deletions(-) diff --git a/store/tikv/region_cache.go b/store/tikv/region_cache.go index a481f5e7ec93e..6b0a0b9e95264 100644 --- a/store/tikv/region_cache.go +++ b/store/tikv/region_cache.go @@ -23,7 +23,6 @@ import ( "unsafe" "github.com/google/btree" - "github.com/grpc-ecosystem/go-grpc-middleware/util/backoffutils" "github.com/pingcap/errors" "github.com/pingcap/kvproto/pkg/metapb" "github.com/pingcap/pd/client" @@ -210,8 +209,8 @@ func (c *RegionCache) checkAndResolve(needCheckStores []*Store) { c.storeMu.RLock() for _, store := range c.storeMu.stores { - state := store.getState() - if state.resolveState == needCheck { + state := store.getResolveState() + if state == needCheck { needCheckStores = append(needCheckStores, store) } } @@ -224,11 +223,12 @@ func (c *RegionCache) checkAndResolve(needCheckStores []*Store) { // RPCContext contains data that is needed to send RPC to a region. type RPCContext struct { - Region RegionVerID - Meta *metapb.Region - Peer *metapb.Peer - Store *Store - Addr string + Region RegionVerID + Meta *metapb.Region + Peer *metapb.Peer + PeerIdx int + Store *Store + Addr string } // GetStoreID returns StoreID. @@ -258,7 +258,9 @@ func (c *RegionCache) GetRPCContext(bo *Backoffer, id RegionVerID) (*RPCContext, return nil, nil } - store, peer, addr, err := c.routeStoreInRegion(bo, cachedRegion, ts) + regionStore := cachedRegion.getStore() + store, peer, storeIdx := cachedRegion.WorkStorePeer(regionStore) + addr, err := c.getStoreAddr(bo, cachedRegion, store, storeIdx) if err != nil { return nil, err } @@ -269,11 +271,12 @@ func (c *RegionCache) GetRPCContext(bo *Backoffer, id RegionVerID) (*RPCContext, } return &RPCContext{ - Region: id, - Meta: cachedRegion.meta, - Peer: peer, - Store: store, - Addr: addr, + Region: id, + Meta: cachedRegion.meta, + Peer: peer, + PeerIdx: storeIdx, + Store: store, + Addr: addr, }, nil } @@ -439,27 +442,28 @@ func (c *RegionCache) InvalidateCachedRegion(id RegionVerID) { } // UpdateLeader update some region cache with newer leader info. -func (c *RegionCache) UpdateLeader(regionID RegionVerID, leaderStoreID uint64) { +func (c *RegionCache) UpdateLeader(regionID RegionVerID, leader *metapb.Peer, currentPeerIdx int) { r := c.getCachedRegionWithRLock(regionID) if r == nil { logutil.Logger(context.Background()).Debug("regionCache: cannot find region when updating leader", zap.Uint64("regionID", regionID.GetID()), - zap.Uint64("leaderStoreID", leaderStoreID)) + zap.Uint64("leaderStoreID", leader.GetStoreId())) + return + } + + if leader == nil { + c.switchNext(r, currentPeerIdx) return } - leaderIdx, found := c.getPeerStoreIndex(r, leaderStoreID) - if !found { + + leaderStoreID := leader.GetStoreId() + if !c.switchWorkStore(r, leaderStoreID) { logutil.Logger(context.Background()).Debug("regionCache: cannot find peer when updating leader", zap.Uint64("regionID", regionID.GetID()), zap.Uint64("leaderStoreID", leaderStoreID)) r.invalidate() return } - rs := r.getStore() - if len(rs.stores) > leaderIdx { - rs.stores[leaderIdx].markAccess(nil, true) - } - c.switchWorkIdx(r, leaderIdx) } // insertRegionToCache tries to insert the Region to cache. @@ -493,9 +497,6 @@ func (c *RegionCache) searchCachedRegion(key []byte, isEndKey bool) *Region { }) c.mu.RUnlock() if r != nil && (!isEndKey && r.Contains(key) || isEndKey && r.ContainsByEnd(key)) { - if !c.hasAvailableStore(r, ts) { - return nil - } return r } return nil @@ -605,66 +606,9 @@ func (c *RegionCache) getCachedRegionWithRLock(regionID RegionVerID) (r *Region) return } -// routeStoreInRegion ensures region have workable store and return it. -func (c *RegionCache) routeStoreInRegion(bo *Backoffer, region *Region, ts int64) (workStore *Store, workPeer *metapb.Peer, workAddr string, err error) { -retry: - regionStore := region.getStore() - cachedStore, cachedPeer, cachedIdx := region.WorkStorePeer(regionStore) - - // Most of the time, requests are directly routed to stable leader. - // returns if store is stable leader and no need retry other node. - state := cachedStore.getState() - if cachedStore != nil && state.failedAttempt == 0 && state.lastFailedTime == 0 { - workStore = cachedStore - workAddr, err = c.getStoreAddr(bo, region, workStore, cachedIdx, state) - workPeer = cachedPeer - return - } - - // try round-robin find & switch to other peers when old leader meet error. - newIdx := -1 - storeNum := len(regionStore.stores) - i := (cachedIdx + 1) % storeNum - start := i - for { - store := regionStore.stores[i] - state = store.getState() - if state.Available(ts) { - newIdx = i - break - } - i = (i + 1) % storeNum - if i == start { - break - } - } - if newIdx < 0 { - return - } - newRegionStore := regionStore.clone() - newRegionStore.workStoreIdx = int32(newIdx) - newRegionStore.attemptAfterLoad++ - attemptOverThreshold := newRegionStore.attemptAfterLoad == reloadRegionThreshold - if attemptOverThreshold { - newRegionStore.attemptAfterLoad = 0 - } - if !region.compareAndSwapStore(regionStore, newRegionStore) { - goto retry - } - - // reload region info after attempts more than reloadRegionThreshold - if attemptOverThreshold { - region.scheduleReload() - } - - workStore = newRegionStore.stores[newIdx] - workAddr, err = c.getStoreAddr(bo, region, workStore, newIdx, state) - workPeer = region.meta.Peers[newIdx] - return -} - -func (c *RegionCache) getStoreAddr(bo *Backoffer, region *Region, store *Store, storeIdx int, state storeState) (addr string, err error) { - switch state.resolveState { +func (c *RegionCache) getStoreAddr(bo *Backoffer, region *Region, store *Store, storeIdx int) (addr string, err error) { + state := store.getResolveState() + switch state { case resolved, needCheck: addr = store.addr return @@ -702,19 +646,6 @@ func (c *RegionCache) changeToActiveStore(region *Region, store *Store, storeIdx return } -// hasAvailableStore checks whether region has available store. -// different to `routeStoreInRegion` just check and never change work store or peer. -func (c *RegionCache) hasAvailableStore(region *Region, ts int64) bool { - regionStore := region.getStore() - for _, store := range regionStore.stores { - state := store.getState() - if state.Available(ts) { - return true - } - } - return false -} - func (c *RegionCache) getStoreByStoreID(storeID uint64) (store *Store) { var ok bool c.storeMu.Lock() @@ -731,14 +662,10 @@ func (c *RegionCache) getStoreByStoreID(storeID uint64) (store *Store) { // OnSendRequestFail is used for clearing cache when a tikv server does not respond. func (c *RegionCache) OnSendRequestFail(ctx *RPCContext, err error) { - failedStoreID := ctx.Store.storeID - c.storeMu.RLock() - store, exists := c.storeMu.stores[failedStoreID] - c.storeMu.RUnlock() - if !exists { - return + r := c.getCachedRegionWithRLock(ctx.Region) + if r == nil { + c.switchNext(r, ctx.PeerIdx) } - store.markAccess(c.notifyCheckCh, false) } // OnRegionEpochNotMatch removes the old region and inserts new regions into the cache. @@ -856,12 +783,34 @@ func (r *Region) EndKey() []byte { // switchWorkStore switches current store to the one on specific store. It returns // false if no peer matches the storeID. -func (c *RegionCache) switchWorkStore(r *Region, targetStoreID uint64) { - leaderIdx, _ := c.getPeerStoreIndex(r, targetStoreID) +func (c *RegionCache) switchWorkStore(r *Region, targetStoreID uint64) (found bool) { + leaderIdx, found := c.getPeerStoreIndex(r, targetStoreID) c.switchWorkIdx(r, leaderIdx) return } +func (c *RegionCache) switchNext(r *Region, currentPeerIdx int) { + regionStore := r.getStore() + if int(regionStore.workStoreIdx) != currentPeerIdx { + return + } + nextIdx := (currentPeerIdx + 1) % len(regionStore.stores) + newRegionStore := regionStore.clone() + newRegionStore.workStoreIdx = int32(nextIdx) + newRegionStore.attemptAfterLoad++ + attemptOverThreshold := newRegionStore.attemptAfterLoad == reloadRegionThreshold + if attemptOverThreshold { + newRegionStore.attemptAfterLoad = 0 + } + if !r.compareAndSwapStore(regionStore, newRegionStore) { + return + } + // reload region info after attempts more than reloadRegionThreshold + if attemptOverThreshold { + r.scheduleReload() + } +} + func (c *RegionCache) getPeerStoreIndex(r *Region, id uint64) (idx int, found bool) { if len(r.meta.Peers) == 0 { return @@ -914,15 +863,7 @@ type Store struct { resolveMutex sync.Mutex // protect pd from concurrent init requests } -// storeState contains store's access info. -type storeState struct { - lastFailedTime uint32 - failedAttempt uint16 - resolveState resolveState - _Align int8 -} - -type resolveState uint8 +type resolveState uint64 const ( unresolved resolveState = iota @@ -934,9 +875,9 @@ const ( // initResolve resolves addr for store that never resolved. func (s *Store) initResolve(bo *Backoffer, c *RegionCache) (addr string, err error) { s.resolveMutex.Lock() - state := s.getState() + state := s.getResolveState() defer s.resolveMutex.Unlock() - if state.resolveState != unresolved { + if state != unresolved { addr = s.addr return } @@ -965,14 +906,12 @@ func (s *Store) initResolve(bo *Backoffer, c *RegionCache) (addr string, err err addr = store.GetAddress() s.addr = addr retry: - state = s.getState() - if state.resolveState != unresolved { + state = s.getResolveState() + if state != unresolved { addr = s.addr return } - newState := state - newState.resolveState = resolved - if !s.compareAndSwapState(state, newState) { + if !s.compareAndSwapState(state, resolved) { goto retry } return @@ -999,8 +938,7 @@ func (s *Store) reResolve(c *RegionCache) { addr = store.GetAddress() if s.addr != addr { - var state storeState - state.resolveState = resolved + state := resolved newStore := &Store{storeID: s.storeID, addr: addr} newStore.state = *(*uint64)(unsafe.Pointer(&state)) c.storeMu.Lock() @@ -1008,123 +946,53 @@ func (s *Store) reResolve(c *RegionCache) { c.storeMu.Unlock() retryMarkDel: // all region used those - oldState := s.getState() - if oldState.resolveState == deleted { + oldState := s.getResolveState() + if oldState == deleted { return } - newState := oldState - newState.resolveState = deleted + newState := deleted if !s.compareAndSwapState(oldState, newState) { goto retryMarkDel } return } retryMarkResolved: - oldState := s.getState() - if oldState.resolveState != needCheck { + oldState := s.getResolveState() + if oldState != needCheck { return } - newState := oldState - newState.resolveState = resolved + newState := resolved if !s.compareAndSwapState(oldState, newState) { goto retryMarkResolved } return } -const ( - // maxExponentAttempt before this blackout time is exponent increment. - maxExponentAttempt = 10 - // startBlackoutAfterAttempt after continue fail attempts start blackout store. - startBlackoutAfterAttempt = 20 -) - -func (s *Store) getState() storeState { - var state storeState +func (s *Store) getResolveState() resolveState { + var state resolveState if s == nil { return state } - x := atomic.LoadUint64(&s.state) - *(*uint64)(unsafe.Pointer(&state)) = x - return state + return resolveState(atomic.LoadUint64(&s.state)) } -func (s *Store) compareAndSwapState(oldState, newState storeState) bool { - oldValue, newValue := *(*uint64)(unsafe.Pointer(&oldState)), *(*uint64)(unsafe.Pointer(&newState)) - return atomic.CompareAndSwapUint64(&s.state, oldValue, newValue) +func (s *Store) compareAndSwapState(oldState, newState resolveState) bool { + return atomic.CompareAndSwapUint64(&s.state, uint64(oldState), uint64(newState)) } -func (s *Store) storeState(newState storeState) { +func (s *Store) storeState(newState resolveState) { newValue := *(*uint64)(unsafe.Pointer(&newState)) atomic.StoreUint64(&s.state, newValue) } -// Available returns whether store be available for current. -func (state storeState) Available(ts int64) bool { - if state.failedAttempt == 0 || state.lastFailedTime == 0 { - // return quickly if it's continue success. - return true - } - // first `startBlackoutAfterAttempt` can retry immediately. - if state.failedAttempt < startBlackoutAfterAttempt { - return true - } - // continue fail over than `startBlackoutAfterAttempt` start blackout store logic. - // check blackout time window to determine store's reachable. - if state.failedAttempt > startBlackoutAfterAttempt+maxExponentAttempt { - state.failedAttempt = startBlackoutAfterAttempt + maxExponentAttempt - } - blackoutDeadline := int64(state.lastFailedTime) + 1*int64(backoffutils.ExponentBase2(uint(state.failedAttempt-startBlackoutAfterAttempt+1))) - return blackoutDeadline <= ts -} - -// markAccess marks the processing result. -func (s *Store) markAccess(notifyCheckCh chan struct{}, success bool) { -retry: - var triggerCheck bool - oldState := s.getState() - if (oldState.failedAttempt == 0 && success) || (!success && oldState.failedAttempt >= (startBlackoutAfterAttempt+maxExponentAttempt)) { - // return quickly if continue success, and no more mark when attempt meet max bound. - return - } - state := oldState - if !success { - if state.lastFailedTime == 0 { - state.lastFailedTime = uint32(time.Now().Unix()) - } - state.failedAttempt = state.failedAttempt + 1 - if state.resolveState == resolved { - state.resolveState = needCheck - triggerCheck = true - } - } else { - state.lastFailedTime = 0 - state.failedAttempt = 0 - } - if !s.compareAndSwapState(oldState, state) { - goto retry - } - if notifyCheckCh == nil { - return - } - if triggerCheck { - select { - case notifyCheckCh <- struct{}{}: - default: - } - } -} - // markNeedCheck marks resolved store to be async resolve to check store addr change. func (s *Store) markNeedCheck(notifyCheckCh chan struct{}) { retry: - oldState := s.getState() - if oldState.resolveState != resolved { + oldState := s.getResolveState() + if oldState != resolved { return } - state := oldState - state.resolveState = needCheck - if !s.compareAndSwapState(oldState, state) { + if !s.compareAndSwapState(oldState, needCheck) { goto retry } select { diff --git a/store/tikv/region_cache_test.go b/store/tikv/region_cache_test.go index 1542d937f65d4..153ab74866663 100644 --- a/store/tikv/region_cache_test.go +++ b/store/tikv/region_cache_test.go @@ -15,7 +15,6 @@ package tikv import ( "context" - "errors" "fmt" "testing" "time" @@ -63,58 +62,32 @@ func (s *testRegionCacheSuite) storeAddr(id uint64) string { func (s *testRegionCacheSuite) checkCache(c *C, len int) { ts := time.Now().Unix() - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, len) - c.Assert(workableRegionsInBtree(s.cache, s.cache.mu.sorted, ts), Equals, len) - for _, r := range s.cache.mu.regions { - if r.checkRegionCacheTTL(ts) { - bo := NewBackoffer(context.Background(), 100) - if store, _, _, _ := s.cache.routeStoreInRegion(bo, r, ts); store != nil { - c.Assert(r, DeepEquals, s.cache.searchCachedRegion(r.StartKey(), false)) - } - } - } + c.Assert(validRegions(s.cache.mu.regions, ts), Equals, len) + c.Assert(validRegionsInBtree(s.cache.mu.sorted, ts), Equals, len) } -func workableRegions(cache *RegionCache, regions map[RegionVerID]*Region, ts int64) (len int) { +func validRegions(regions map[RegionVerID]*Region, ts int64) (len int) { for _, region := range regions { if !region.checkRegionCacheTTL(ts) { continue } - bo := NewBackoffer(context.Background(), 100) - store, _, _, _ := cache.routeStoreInRegion(bo, region, ts) - if store != nil { - len++ - } + len++ } return } -func workableRegionsInBtree(cache *RegionCache, t *btree.BTree, ts int64) (len int) { +func validRegionsInBtree(t *btree.BTree, ts int64) (len int) { t.Descend(func(item btree.Item) bool { r := item.(*btreeItem).cachedRegion if !r.checkRegionCacheTTL(ts) { return true } - bo := NewBackoffer(context.Background(), 100) - store, _, _, _ := cache.routeStoreInRegion(bo, r, ts) - if store != nil { - len++ - } + len++ return true }) return } -func reachableStore(stores map[uint64]*Store, ts int64) (cnt int) { - for _, store := range stores { - state := store.getState() - if state.Available(ts) { - cnt++ - } - } - return -} - func (s *testRegionCacheSuite) getRegion(c *C, key []byte) *Region { _, err := s.cache.LocateKey(s.bo, key) c.Assert(err, IsNil) @@ -182,7 +155,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader(c *C) { loc, err := s.cache.LocateKey(s.bo, []byte("a")) c.Assert(err, IsNil) // tikv-server reports `NotLeader` - s.cache.UpdateLeader(loc.Region, s.store2) + s.cache.UpdateLeader(loc.Region, &metapb.Peer{StoreId: s.store2}, 0) r := s.getRegion(c, []byte("a")) c.Assert(r, NotNil) @@ -204,7 +177,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader2(c *C) { s.cluster.AddStore(store3, s.storeAddr(store3)) s.cluster.AddPeer(s.region1, store3, peer3) // tikv-server reports `NotLeader` - s.cache.UpdateLeader(loc.Region, store3) + s.cache.UpdateLeader(loc.Region, &metapb.Peer{StoreId: store3}, 0) // Store3 does not exist in cache, causes a reload from PD. r := s.getRegion(c, []byte("a")) @@ -215,7 +188,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader2(c *C) { // tikv-server notifies new leader to pd-server. s.cluster.ChangeLeader(s.region1, peer3) // tikv-server reports `NotLeader` again. - s.cache.UpdateLeader(r.VerID(), store3) + s.cache.UpdateLeader(r.VerID(), &metapb.Peer{StoreId: store3}, 0) r = s.getRegion(c, []byte("a")) c.Assert(r, NotNil) c.Assert(r.GetID(), Equals, s.region1) @@ -236,7 +209,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader3(c *C) { // tikv-server notifies new leader to pd-server. s.cluster.ChangeLeader(s.region1, peer3) // tikv-server reports `NotLeader`(store2 is the leader) - s.cache.UpdateLeader(loc.Region, s.store2) + s.cache.UpdateLeader(loc.Region, &metapb.Peer{StoreId: s.store2}, 0) // Store2 does not exist any more, causes a reload from PD. r := s.getRegion(c, []byte("a")) @@ -250,6 +223,10 @@ func (s *testRegionCacheSuite) TestUpdateLeader3(c *C) { c.Assert(s.getAddr(c, []byte("a")), Equals, s.storeAddr(store3)) } +func (s *testRegionCacheSuite) TestUpdateLeader4(c *C) { + +} + func (s *testRegionCacheSuite) TestSplit(c *C) { r := s.getRegion(c, []byte("x")) c.Assert(r.GetID(), Equals, s.region1) @@ -311,86 +288,6 @@ func (s *testRegionCacheSuite) TestReconnect(c *C) { s.checkCache(c, 1) } -func (s *testRegionCacheSuite) TestSendFailBlackout(c *C) { - ts := time.Now().Unix() - region := s.getRegion(c, []byte("a")) - - // init with 1 region 2 stores - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, 1) - c.Assert(reachableStore(s.cache.storeMu.stores, ts), Equals, 2) - - // for each stores has 20 chance to retry, and still have chance to access store for 21 - for i := 0; i < 38; i++ { - ctx, _ := s.cache.GetRPCContext(s.bo, region.VerID()) - if ctx == nil { - fmt.Println() - } - s.cache.OnSendRequestFail(ctx, errors.New("test error")) - - } - ts = time.Now().Unix() - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, 1) - c.Assert(reachableStore(s.cache.storeMu.stores, ts), Equals, 2) - - // 21 fail attempts will start blackout store in 1 second - for i := 0; i < 2; i++ { - // first fail request make 1st store' failAttempt + 1 - ctx, _ := s.cache.GetRPCContext(s.bo, region.VerID()) - s.cache.OnSendRequestFail(ctx, errors.New("test error")) - } - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, 0) - c.Assert(reachableStore(s.cache.storeMu.stores, ts), Equals, 0) - - // after 1 second blackout, 2 store can be access again. - time.Sleep(1 * time.Second) - ts = time.Now().Unix() - s.getRegion(c, []byte("a")) - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, 1) - c.Assert(reachableStore(s.cache.storeMu.stores, ts), Equals, 2) -} - -func (s *testRegionCacheSuite) TestSendFailBlackTwoRegion(c *C) { - ts := time.Now().Unix() - // key range: ['' - 'm' - 'z'] - region2 := s.cluster.AllocID() - newPeers := s.cluster.AllocIDs(2) - s.cluster.Split(s.region1, region2, []byte("m"), newPeers, newPeers[0]) - - // Check the two regions. - loc1, err := s.cache.LocateKey(s.bo, []byte("a")) - c.Assert(err, IsNil) - c.Assert(loc1.Region.id, Equals, s.region1) - loc2, err := s.cache.LocateKey(s.bo, []byte("x")) - c.Assert(err, IsNil) - c.Assert(loc2.Region.id, Equals, region2) - - c.Assert(reachableStore(s.cache.storeMu.stores, ts), Equals, 2) - s.checkCache(c, 2) - - // send request fail in 2 regions backed by same 2 stores. - for i := 0; i < startBlackoutAfterAttempt; i++ { - ctx, _ := s.cache.GetRPCContext(s.bo, loc1.Region) - s.cache.OnSendRequestFail(ctx, errors.New("test error")) - } - for i := 0; i < startBlackoutAfterAttempt; i++ { - ctx, _ := s.cache.GetRPCContext(s.bo, loc2.Region) - s.cache.OnSendRequestFail(ctx, errors.New("test error")) - } - - // both 2 region are invalidate and both 2 store are available. - c.Assert(reachableStore(s.cache.storeMu.stores, ts), Equals, 0) - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, 0) - - // after sleep 1 second, region recover - time.Sleep(1 * time.Second) - ts = time.Now().Unix() - c.Assert(reachableStore(s.cache.storeMu.stores, ts), Equals, 2) - s.getRegion(c, []byte("a")) - s.getRegion(c, []byte("x")) - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, 2) - c.Assert(workableRegions(s.cache, s.cache.mu.regions, ts), Equals, 2) -} - func (s *testRegionCacheSuite) TestRegionEpochAheadOfTiKV(c *C) { // Create a separated region cache to do this test. pdCli := &codecPDClient{mocktikv.NewPDClient(s.cluster)} @@ -414,31 +311,6 @@ func (s *testRegionCacheSuite) TestRegionEpochAheadOfTiKV(c *C) { c.Assert(len(bo.errors), Equals, 2) } -func (s *testRegionCacheSuite) TestDropStoreOnSendRequestFail(c *C) { - ts := time.Now().Unix() - regionCnt, storeCount := 8, 3 - cluster := createClusterWithStoresAndRegions(regionCnt, storeCount) - - cache := NewRegionCache(mocktikv.NewPDClient(cluster)) - defer cache.Close() - loadRegionsToCache(cache, regionCnt) - c.Assert(workableRegions(cache, cache.mu.regions, ts), Equals, regionCnt) - - bo := NewBackoffer(context.Background(), 1) - loc, err := cache.LocateKey(bo, []byte{}) - c.Assert(err, IsNil) - - // fail on one region make all stores be unavailable. - for j := 0; j < 20; j++ { - for i := 0; i < storeCount; i++ { - rpcCtx, err := cache.GetRPCContext(bo, loc.Region) - c.Assert(err, IsNil) - cache.OnSendRequestFail(rpcCtx, errors.New("test error")) - } - } - c.Assert(workableRegions(cache, cache.mu.regions, ts), Equals, 0) -} - const regionSplitKeyFormat = "t%08d" func createClusterWithStoresAndRegions(regionCnt, storeCount int) *mocktikv.Cluster { @@ -563,14 +435,15 @@ func BenchmarkOnRequestFail(b *testing.B) { region := cache.getRegionByIDFromCache(loc.Region.id) b.ResetTimer() regionStore := region.getStore() - store, peer, _ := region.WorkStorePeer(regionStore) + store, peer, idx := region.WorkStorePeer(regionStore) b.RunParallel(func(pb *testing.PB) { for pb.Next() { rpcCtx := &RPCContext{ - Region: loc.Region, - Meta: region.meta, - Peer: peer, - Store: store, + Region: loc.Region, + Meta: region.meta, + PeerIdx: idx, + Peer: peer, + Store: store, } cache.OnSendRequestFail(rpcCtx, nil) } diff --git a/store/tikv/region_request.go b/store/tikv/region_request.go index d3a55d3a135d1..57b7e4e675cbd 100644 --- a/store/tikv/region_request.go +++ b/store/tikv/region_request.go @@ -149,15 +149,9 @@ func (s *RegionRequestSender) sendReqToRegion(bo *Backoffer, ctx *RPCContext, re } return nil, true, nil } - s.onSendSuccess(ctx) return } -func (s *RegionRequestSender) onSendSuccess(ctx *RPCContext) { - store := s.regionCache.getStoreByStoreID(ctx.Store.storeID) - store.markAccess(s.regionCache.notifyCheckCh, true) -} - func (s *RegionRequestSender) onSendFail(bo *Backoffer, ctx *RPCContext, err error) error { // If it failed because the context is cancelled by ourself, don't retry. if errors.Cause(err) == context.Canceled { @@ -213,7 +207,7 @@ func (s *RegionRequestSender) onRegionError(bo *Backoffer, ctx *RPCContext, regi logutil.Logger(context.Background()).Debug("tikv reports `NotLeader` retry later", zap.String("notLeader", notLeader.String()), zap.String("ctx", ctx.String())) - s.regionCache.UpdateLeader(ctx.Region, notLeader.GetLeader().GetStoreId()) + s.regionCache.UpdateLeader(ctx.Region, notLeader.GetLeader(), ctx.PeerIdx) var boType backoffType if notLeader.GetLeader() != nil { From be708876aab087a2d802a9b4f120f22c4c299472 Mon Sep 17 00:00:00 2001 From: lysu Date: Thu, 23 May 2019 16:25:32 +0800 Subject: [PATCH 3/7] address comments --- store/tikv/region_cache.go | 48 ++++++++++++--------------------- store/tikv/region_cache_test.go | 13 +++++---- store/tikv/region_request.go | 24 ++++++++++++----- 3 files changed, 43 insertions(+), 42 deletions(-) diff --git a/store/tikv/region_cache.go b/store/tikv/region_cache.go index 6b0a0b9e95264..66281175901a8 100644 --- a/store/tikv/region_cache.go +++ b/store/tikv/region_cache.go @@ -64,17 +64,15 @@ type Region struct { // RegionStore represents region stores info // it will be store as unsafe.Pointer and be load at once type RegionStore struct { - workStoreIdx int32 // point to current work peer in meta.Peers and work store in stores(same idx) - stores []*Store // stores in this region - attemptAfterLoad uint8 // indicate switch peer attempts after load region info + workStoreIdx int32 // point to current work peer in meta.Peers and work store in stores(same idx) + stores []*Store // stores in this region } // clone clones region store struct. func (r *RegionStore) clone() *RegionStore { return &RegionStore{ - workStoreIdx: r.workStoreIdx, - stores: r.stores, - attemptAfterLoad: r.attemptAfterLoad, + workStoreIdx: r.workStoreIdx, + stores: r.stores, } } @@ -442,22 +440,21 @@ func (c *RegionCache) InvalidateCachedRegion(id RegionVerID) { } // UpdateLeader update some region cache with newer leader info. -func (c *RegionCache) UpdateLeader(regionID RegionVerID, leader *metapb.Peer, currentPeerIdx int) { +func (c *RegionCache) UpdateLeader(regionID RegionVerID, leaderStoreID uint64, currentPeerIdx int) { r := c.getCachedRegionWithRLock(regionID) if r == nil { logutil.Logger(context.Background()).Debug("regionCache: cannot find region when updating leader", zap.Uint64("regionID", regionID.GetID()), - zap.Uint64("leaderStoreID", leader.GetStoreId())) + zap.Uint64("leaderStoreID", leaderStoreID)) return } - if leader == nil { - c.switchNext(r, currentPeerIdx) + if leaderStoreID == 0 { + c.switchNextPeer(r, currentPeerIdx) return } - leaderStoreID := leader.GetStoreId() - if !c.switchWorkStore(r, leaderStoreID) { + if !c.switchToPeer(r, leaderStoreID) { logutil.Logger(context.Background()).Debug("regionCache: cannot find peer when updating leader", zap.Uint64("regionID", regionID.GetID()), zap.Uint64("leaderStoreID", leaderStoreID)) @@ -557,7 +554,7 @@ func (c *RegionCache) loadRegion(bo *Backoffer, key []byte, isEndKey bool) (*Reg region := &Region{meta: meta} region.init(c) if leader != nil { - c.switchWorkStore(region, leader.StoreId) + c.switchToPeer(region, leader.StoreId) } return region, nil } @@ -593,7 +590,7 @@ func (c *RegionCache) loadRegionByID(bo *Backoffer, regionID uint64) (*Region, e region := &Region{meta: meta} region.init(c) if leader != nil { - c.switchWorkStore(region, leader.GetStoreId()) + c.switchToPeer(region, leader.GetStoreId()) } return region, nil } @@ -664,7 +661,7 @@ func (c *RegionCache) getStoreByStoreID(storeID uint64) (store *Store) { func (c *RegionCache) OnSendRequestFail(ctx *RPCContext, err error) { r := c.getCachedRegionWithRLock(ctx.Region) if r == nil { - c.switchNext(r, ctx.PeerIdx) + c.switchNextPeer(r, ctx.PeerIdx) } } @@ -693,7 +690,7 @@ func (c *RegionCache) OnRegionEpochNotMatch(bo *Backoffer, ctx *RPCContext, curr } region := &Region{meta: meta} region.init(c) - c.switchWorkStore(region, ctx.Store.storeID) + c.switchToPeer(region, ctx.Store.storeID) c.insertRegionToCache(region) if ctx.Region == region.VerID() { needInvalidateOld = false @@ -781,15 +778,15 @@ func (r *Region) EndKey() []byte { return r.meta.EndKey } -// switchWorkStore switches current store to the one on specific store. It returns +// switchToPeer switches current store to the one on specific store. It returns // false if no peer matches the storeID. -func (c *RegionCache) switchWorkStore(r *Region, targetStoreID uint64) (found bool) { +func (c *RegionCache) switchToPeer(r *Region, targetStoreID uint64) (found bool) { leaderIdx, found := c.getPeerStoreIndex(r, targetStoreID) c.switchWorkIdx(r, leaderIdx) return } -func (c *RegionCache) switchNext(r *Region, currentPeerIdx int) { +func (c *RegionCache) switchNextPeer(r *Region, currentPeerIdx int) { regionStore := r.getStore() if int(regionStore.workStoreIdx) != currentPeerIdx { return @@ -797,18 +794,7 @@ func (c *RegionCache) switchNext(r *Region, currentPeerIdx int) { nextIdx := (currentPeerIdx + 1) % len(regionStore.stores) newRegionStore := regionStore.clone() newRegionStore.workStoreIdx = int32(nextIdx) - newRegionStore.attemptAfterLoad++ - attemptOverThreshold := newRegionStore.attemptAfterLoad == reloadRegionThreshold - if attemptOverThreshold { - newRegionStore.attemptAfterLoad = 0 - } - if !r.compareAndSwapStore(regionStore, newRegionStore) { - return - } - // reload region info after attempts more than reloadRegionThreshold - if attemptOverThreshold { - r.scheduleReload() - } + r.compareAndSwapStore(regionStore, newRegionStore) } func (c *RegionCache) getPeerStoreIndex(r *Region, id uint64) (idx int, found bool) { diff --git a/store/tikv/region_cache_test.go b/store/tikv/region_cache_test.go index 153ab74866663..5cba40eb912dd 100644 --- a/store/tikv/region_cache_test.go +++ b/store/tikv/region_cache_test.go @@ -155,7 +155,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader(c *C) { loc, err := s.cache.LocateKey(s.bo, []byte("a")) c.Assert(err, IsNil) // tikv-server reports `NotLeader` - s.cache.UpdateLeader(loc.Region, &metapb.Peer{StoreId: s.store2}, 0) + s.cache.UpdateLeader(loc.Region, s.store2, 0) r := s.getRegion(c, []byte("a")) c.Assert(r, NotNil) @@ -177,7 +177,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader2(c *C) { s.cluster.AddStore(store3, s.storeAddr(store3)) s.cluster.AddPeer(s.region1, store3, peer3) // tikv-server reports `NotLeader` - s.cache.UpdateLeader(loc.Region, &metapb.Peer{StoreId: store3}, 0) + s.cache.UpdateLeader(loc.Region, store3, 0) // Store3 does not exist in cache, causes a reload from PD. r := s.getRegion(c, []byte("a")) @@ -188,7 +188,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader2(c *C) { // tikv-server notifies new leader to pd-server. s.cluster.ChangeLeader(s.region1, peer3) // tikv-server reports `NotLeader` again. - s.cache.UpdateLeader(r.VerID(), &metapb.Peer{StoreId: store3}, 0) + s.cache.UpdateLeader(r.VerID(), store3, 0) r = s.getRegion(c, []byte("a")) c.Assert(r, NotNil) c.Assert(r.GetID(), Equals, s.region1) @@ -209,7 +209,7 @@ func (s *testRegionCacheSuite) TestUpdateLeader3(c *C) { // tikv-server notifies new leader to pd-server. s.cluster.ChangeLeader(s.region1, peer3) // tikv-server reports `NotLeader`(store2 is the leader) - s.cache.UpdateLeader(loc.Region, &metapb.Peer{StoreId: s.store2}, 0) + s.cache.UpdateLeader(loc.Region, s.store2, 0) // Store2 does not exist any more, causes a reload from PD. r := s.getRegion(c, []byte("a")) @@ -445,7 +445,10 @@ func BenchmarkOnRequestFail(b *testing.B) { Peer: peer, Store: store, } - cache.OnSendRequestFail(rpcCtx, nil) + r := cache.getCachedRegionWithRLock(rpcCtx.Region) + if r == nil { + cache.switchNextPeer(r, rpcCtx.PeerIdx) + } } }) if len(cache.mu.regions) != regionCnt*2/3 { diff --git a/store/tikv/region_request.go b/store/tikv/region_request.go index 57b7e4e675cbd..2a7631071c33a 100644 --- a/store/tikv/region_request.go +++ b/store/tikv/region_request.go @@ -52,10 +52,11 @@ var ShuttingDown uint32 // errors, since region range have changed, the request may need to split, so we // simply return the error to caller. type RegionRequestSender struct { - regionCache *RegionCache - client Client - storeAddr string - rpcError error + regionCache *RegionCache + client Client + storeAddr string + rpcError error + failStoreIDs map[uint64]struct{} } // NewRegionRequestSender creates a new sender. @@ -171,7 +172,18 @@ func (s *RegionRequestSender) onSendFail(bo *Backoffer, ctx *RPCContext, err err } } - s.regionCache.OnSendRequestFail(ctx, err) + r := s.regionCache.getCachedRegionWithRLock(ctx.Region) + if r != nil { + s.regionCache.switchNextPeer(r, ctx.PeerIdx) + + if s.failStoreIDs == nil { + s.failStoreIDs = make(map[uint64]struct{}) + } + s.failStoreIDs[ctx.Store.storeID] = struct{}{} + if len(s.failStoreIDs) == len(ctx.Meta.Peers) { + r.scheduleReload() + } + } // Retry on send request failure when it's not canceled. // When a store is not available, the leader of related region should be elected quickly. @@ -207,7 +219,7 @@ func (s *RegionRequestSender) onRegionError(bo *Backoffer, ctx *RPCContext, regi logutil.Logger(context.Background()).Debug("tikv reports `NotLeader` retry later", zap.String("notLeader", notLeader.String()), zap.String("ctx", ctx.String())) - s.regionCache.UpdateLeader(ctx.Region, notLeader.GetLeader(), ctx.PeerIdx) + s.regionCache.UpdateLeader(ctx.Region, notLeader.GetLeader().GetStoreId(), ctx.PeerIdx) var boType backoffType if notLeader.GetLeader() != nil { From bc43184c0e8ff0e4f9bcb48d37736e37ca85459f Mon Sep 17 00:00:00 2001 From: lysu Date: Thu, 23 May 2019 17:00:47 +0800 Subject: [PATCH 4/7] remove noused code --- store/tikv/region_cache.go | 8 -------- 1 file changed, 8 deletions(-) diff --git a/store/tikv/region_cache.go b/store/tikv/region_cache.go index 66281175901a8..7e52382f118e5 100644 --- a/store/tikv/region_cache.go +++ b/store/tikv/region_cache.go @@ -657,14 +657,6 @@ func (c *RegionCache) getStoreByStoreID(storeID uint64) (store *Store) { return } -// OnSendRequestFail is used for clearing cache when a tikv server does not respond. -func (c *RegionCache) OnSendRequestFail(ctx *RPCContext, err error) { - r := c.getCachedRegionWithRLock(ctx.Region) - if r == nil { - c.switchNextPeer(r, ctx.PeerIdx) - } -} - // OnRegionEpochNotMatch removes the old region and inserts new regions into the cache. func (c *RegionCache) OnRegionEpochNotMatch(bo *Backoffer, ctx *RPCContext, currentRegions []*metapb.Region) error { // Find whether the region epoch in `ctx` is ahead of TiKV's. If so, backoff. From c4b3494ab727afd7b6b50adc0269e7575fc5779d Mon Sep 17 00:00:00 2001 From: lysu Date: Thu, 23 May 2019 18:51:17 +0800 Subject: [PATCH 5/7] add test & check needReload for getRegionByID --- store/tikv/region_cache.go | 28 +++++++-- store/tikv/region_cache_test.go | 96 ++++++++++++++++++++++++++++++- store/tikv/region_request.go | 26 +++++---- store/tikv/region_request_test.go | 39 +++++++++++++ 4 files changed, 173 insertions(+), 16 deletions(-) diff --git a/store/tikv/region_cache.go b/store/tikv/region_cache.go index 7e52382f118e5..baf67ec4f19e3 100644 --- a/store/tikv/region_cache.go +++ b/store/tikv/region_cache.go @@ -35,7 +35,6 @@ const ( btreeDegree = 32 rcDefaultRegionCacheTTLSec = 600 invalidatedLastAccessTime = -1 - reloadRegionThreshold = 5 ) var ( @@ -359,20 +358,42 @@ func (c *RegionCache) findRegionByKey(bo *Backoffer, key []byte, isEndKey bool) return r, nil } +func (c *RegionCache) OnSendFail(bo *Backoffer, ctx *RPCContext, scheduleReload bool) { + r := c.getCachedRegionWithRLock(ctx.Region) + if r != nil { + c.switchNextPeer(r, ctx.PeerIdx) + if scheduleReload { + r.scheduleReload() + } + } +} + // LocateRegionByID searches for the region with ID. func (c *RegionCache) LocateRegionByID(bo *Backoffer, regionID uint64) (*KeyLocation, error) { c.mu.RLock() r := c.getRegionByIDFromCache(regionID) + c.mu.RUnlock() if r != nil { + if r.needReload() { + lr, err := c.loadRegionByID(bo, regionID) + if err != nil { + // ignore error and use old region info. + logutil.Logger(bo.ctx).Error("load region failure", + zap.Uint64("regionID", regionID), zap.Error(err)) + } else { + r = lr + c.mu.Lock() + c.insertRegionToCache(r) + c.mu.Unlock() + } + } loc := &KeyLocation{ Region: r.VerID(), StartKey: r.StartKey(), EndKey: r.EndKey(), } - c.mu.RUnlock() return loc, nil } - c.mu.RUnlock() r, err := c.loadRegionByID(bo, regionID) if err != nil { @@ -459,7 +480,6 @@ func (c *RegionCache) UpdateLeader(regionID RegionVerID, leaderStoreID uint64, c zap.Uint64("regionID", regionID.GetID()), zap.Uint64("leaderStoreID", leaderStoreID)) r.invalidate() - return } } diff --git a/store/tikv/region_cache_test.go b/store/tikv/region_cache_test.go index 5cba40eb912dd..dc78d01413eb9 100644 --- a/store/tikv/region_cache_test.go +++ b/store/tikv/region_cache_test.go @@ -223,8 +223,102 @@ func (s *testRegionCacheSuite) TestUpdateLeader3(c *C) { c.Assert(s.getAddr(c, []byte("a")), Equals, s.storeAddr(store3)) } -func (s *testRegionCacheSuite) TestUpdateLeader4(c *C) { +func (s *testRegionCacheSuite) TestSendFailedButLeaderNotChange(c *C) { + // 3 nodes and no.1 is leader. + store3 := s.cluster.AllocID() + peer3 := s.cluster.AllocID() + s.cluster.AddStore(store3, s.storeAddr(store3)) + s.cluster.AddPeer(s.region1, store3, peer3) + s.cluster.ChangeLeader(s.region1, s.peer1) + + loc, err := s.cache.LocateKey(s.bo, []byte("a")) + c.Assert(err, IsNil) + ctx, err := s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer1) + c.Assert(len(ctx.Meta.Peers), Equals, 3) + + // send fail leader switch to 2 + s.cache.OnSendFail(s.bo, ctx, false) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer2) + + // access 1 it will return NotLeader, leader back to 2 again + s.cache.UpdateLeader(loc.Region, s.store2, ctx.PeerIdx) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer2) +} + +func (s *testRegionCacheSuite) TestSendFailedInHibernateRegion(c *C) { + // 3 nodes and no.1 is leader. + store3 := s.cluster.AllocID() + peer3 := s.cluster.AllocID() + s.cluster.AddStore(store3, s.storeAddr(store3)) + s.cluster.AddPeer(s.region1, store3, peer3) + s.cluster.ChangeLeader(s.region1, s.peer1) + + loc, err := s.cache.LocateKey(s.bo, []byte("a")) + c.Assert(err, IsNil) + ctx, err := s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer1) + c.Assert(len(ctx.Meta.Peers), Equals, 3) + // send fail leader switch to 2 + s.cache.OnSendFail(s.bo, ctx, false) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer2) + + // access 2, it's in hibernate and return 0 leader, so switch to 3 + s.cache.UpdateLeader(loc.Region, 0, ctx.PeerIdx) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, peer3) + + // again peer back to 1 + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + s.cache.UpdateLeader(loc.Region, 0, ctx.PeerIdx) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer1) +} + +func (s *testRegionCacheSuite) TestSendFailedInMultipleNode(c *C) { + // 3 nodes and no.1 is leader. + store3 := s.cluster.AllocID() + peer3 := s.cluster.AllocID() + s.cluster.AddStore(store3, s.storeAddr(store3)) + s.cluster.AddPeer(s.region1, store3, peer3) + s.cluster.ChangeLeader(s.region1, s.peer1) + + loc, err := s.cache.LocateKey(s.bo, []byte("a")) + c.Assert(err, IsNil) + ctx, err := s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer1) + c.Assert(len(ctx.Meta.Peers), Equals, 3) + + // send fail leader switch to 2 + s.cache.OnSendFail(s.bo, ctx, false) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer2) + + // send 2 fail leader switch to 3 + s.cache.OnSendFail(s.bo, ctx, false) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, peer3) + + // 3 can be access, so switch to 1 + s.cache.UpdateLeader(loc.Region, s.store1, ctx.PeerIdx) + ctx, err = s.cache.GetRPCContext(s.bo, loc.Region) + c.Assert(err, IsNil) + c.Assert(ctx.Peer.Id, Equals, s.peer1) } func (s *testRegionCacheSuite) TestSplit(c *C) { diff --git a/store/tikv/region_request.go b/store/tikv/region_request.go index 2a7631071c33a..64cbfb71227cc 100644 --- a/store/tikv/region_request.go +++ b/store/tikv/region_request.go @@ -172,18 +172,9 @@ func (s *RegionRequestSender) onSendFail(bo *Backoffer, ctx *RPCContext, err err } } - r := s.regionCache.getCachedRegionWithRLock(ctx.Region) - if r != nil { - s.regionCache.switchNextPeer(r, ctx.PeerIdx) + needReloadRegion := s.needReloadRegion(ctx) - if s.failStoreIDs == nil { - s.failStoreIDs = make(map[uint64]struct{}) - } - s.failStoreIDs[ctx.Store.storeID] = struct{}{} - if len(s.failStoreIDs) == len(ctx.Meta.Peers) { - r.scheduleReload() - } - } + s.regionCache.OnSendFail(bo, ctx, needReloadRegion) // Retry on send request failure when it's not canceled. // When a store is not available, the leader of related region should be elected quickly. @@ -193,6 +184,19 @@ func (s *RegionRequestSender) onSendFail(bo *Backoffer, ctx *RPCContext, err err return errors.Trace(err) } +// needReloadRegion checks is all peers has sent failed, if so need reload. +func (s *RegionRequestSender) needReloadRegion(ctx *RPCContext) (need bool) { + if s.failStoreIDs == nil { + s.failStoreIDs = make(map[uint64]struct{}) + } + s.failStoreIDs[ctx.Store.storeID] = struct{}{} + need = len(s.failStoreIDs) == len(ctx.Meta.Peers) + if need { + s.failStoreIDs = nil + } + return +} + func regionErrorToLabel(e *errorpb.Error) string { if e.GetNotLeader() != nil { return "not_leader" diff --git a/store/tikv/region_request_test.go b/store/tikv/region_request_test.go index 949af3f9eef7f..52cc1636bef7e 100644 --- a/store/tikv/region_request_test.go +++ b/store/tikv/region_request_test.go @@ -15,6 +15,7 @@ package tikv import ( "context" + "fmt" "net" "sync" "time" @@ -91,6 +92,44 @@ func (s *testRegionRequestSuite) TestOnSendFailedWithStoreRestart(c *C) { c.Assert(resp.RawPut, NotNil) } +func (s *testRegionRequestSuite) TestOnSendFailedWithCloseKnownStoreThenUseNewOne(c *C) { + req := &tikvrpc.Request{ + Type: tikvrpc.CmdRawPut, + RawPut: &kvrpcpb.RawPutRequest{ + Key: []byte("key"), + Value: []byte("value"), + }, + } + region, err := s.cache.LocateRegionByID(s.bo, s.region) + c.Assert(err, IsNil) + c.Assert(region, NotNil) + resp, err := s.regionRequestSender.SendReq(s.bo, req, region.Region, time.Second) + c.Assert(err, IsNil) + c.Assert(resp.RawPut, NotNil) + + // add new unknown region + store2 := s.cluster.AllocID() + peer2 := s.cluster.AllocID() + s.cluster.AddStore(store2, fmt.Sprintf("store%d", store2)) + s.cluster.AddPeer(region.Region.id, store2, peer2) + + // stop known region + s.cluster.StopStore(s.store) + + // send to failed store + resp, err = s.regionRequestSender.SendReq(NewBackoffer(context.Background(), 100), req, region.Region, time.Second) + c.Assert(err, NotNil) + + // retry to send store by old region info + region, err = s.cache.LocateRegionByID(s.bo, s.region) + c.Assert(region, NotNil) + c.Assert(err, IsNil) + + // retry again, reload region info and send to new store. + resp, err = s.regionRequestSender.SendReq(NewBackoffer(context.Background(), 100), req, region.Region, time.Second) + c.Assert(err, NotNil) +} + func (s *testRegionRequestSuite) TestSendReqCtx(c *C) { req := &tikvrpc.Request{ Type: tikvrpc.CmdRawPut, From 04e559b93c38045f30a188b2eb23acdd06ea7fe9 Mon Sep 17 00:00:00 2001 From: lysu Date: Thu, 23 May 2019 18:56:31 +0800 Subject: [PATCH 6/7] fix ci --- store/tikv/region_cache.go | 5 +++-- store/tikv/region_request.go | 4 +--- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/store/tikv/region_cache.go b/store/tikv/region_cache.go index baf67ec4f19e3..a21ac4f5b2056 100644 --- a/store/tikv/region_cache.go +++ b/store/tikv/region_cache.go @@ -237,8 +237,8 @@ func (c *RPCContext) GetStoreID() uint64 { } func (c *RPCContext) String() string { - return fmt.Sprintf("region ID: %d, meta: %s, peer: %s, addr: %s", - c.Region.GetID(), c.Meta, c.Peer, c.Addr) + return fmt.Sprintf("region ID: %d, meta: %s, peer: %s, addr: %s, idx: %d, store: %v", + c.Region.GetID(), c.Meta, c.Peer, c.Addr, c.PeerIdx, c.Store) } // GetRPCContext returns RPCContext for a region. If it returns nil, the region @@ -358,6 +358,7 @@ func (c *RegionCache) findRegionByKey(bo *Backoffer, key []byte, isEndKey bool) return r, nil } +// OnSendFail handles send request fail logic. func (c *RegionCache) OnSendFail(bo *Backoffer, ctx *RPCContext, scheduleReload bool) { r := c.getCachedRegionWithRLock(ctx.Region) if r != nil { diff --git a/store/tikv/region_request.go b/store/tikv/region_request.go index 64cbfb71227cc..78fde3854f853 100644 --- a/store/tikv/region_request.go +++ b/store/tikv/region_request.go @@ -172,9 +172,7 @@ func (s *RegionRequestSender) onSendFail(bo *Backoffer, ctx *RPCContext, err err } } - needReloadRegion := s.needReloadRegion(ctx) - - s.regionCache.OnSendFail(bo, ctx, needReloadRegion) + s.regionCache.OnSendFail(bo, ctx, s.needReloadRegion(ctx)) // Retry on send request failure when it's not canceled. // When a store is not available, the leader of related region should be elected quickly. From 0d09d392d2e4bf3c18761f54addd9b0152ac84a5 Mon Sep 17 00:00:00 2001 From: lysu Date: Thu, 23 May 2019 19:18:45 +0800 Subject: [PATCH 7/7] fix ci --- store/tikv/region_cache.go | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/store/tikv/region_cache.go b/store/tikv/region_cache.go index a21ac4f5b2056..64005740fbc8a 100644 --- a/store/tikv/region_cache.go +++ b/store/tikv/region_cache.go @@ -237,8 +237,8 @@ func (c *RPCContext) GetStoreID() uint64 { } func (c *RPCContext) String() string { - return fmt.Sprintf("region ID: %d, meta: %s, peer: %s, addr: %s, idx: %d, store: %v", - c.Region.GetID(), c.Meta, c.Peer, c.Addr, c.PeerIdx, c.Store) + return fmt.Sprintf("region ID: %d, meta: %s, peer: %s, addr: %s, idx: %d", + c.Region.GetID(), c.Meta, c.Peer, c.Addr, c.PeerIdx) } // GetRPCContext returns RPCContext for a region. If it returns nil, the region @@ -979,11 +979,6 @@ func (s *Store) compareAndSwapState(oldState, newState resolveState) bool { return atomic.CompareAndSwapUint64(&s.state, uint64(oldState), uint64(newState)) } -func (s *Store) storeState(newState resolveState) { - newValue := *(*uint64)(unsafe.Pointer(&newState)) - atomic.StoreUint64(&s.state, newValue) -} - // markNeedCheck marks resolved store to be async resolve to check store addr change. func (s *Store) markNeedCheck(notifyCheckCh chan struct{}) { retry: