Repository: tikv/pd
Pull request #4219 (2021-10-19 05:57:12)

core: allow cancel load region (#4175)
This is an automated cherry-pick of #4175

### What problem does this PR solve?

Fix https://github.com/tikv/pd/issues/3936

### What is changed and how it works?

- add cancel support for core.LoadRegion functions
- replace `chan bool` with `contextes` in region syncer client
- add test case

### Check List

<!-- Remove the items that are not applicable. -->

Tests

<!-- At least one of them must be included. -->

- Unit test

### Release note

```release-note
Fix the issue that PD may not elect leader as soon as leader step down
```


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/tikv/pd/pulls/4219/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/tikv/pd/pulls/4219/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found [here](https://prow.tidb.io/command-help?repo=tikv%2Fpd).

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
ti-chi-bot: @ti-chi-bot: This cherry pick PR is for a release branch and has not yet been approved by release team.
Adding the `do-not-merge/cherry-pick-not-approved` label.

To merge this cherry pick, it must first be approved by the collaborators.

**AFTER** it has been approved by collaborators, please ping the release team in a comment to request a cherry pick review.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

Please provide the associated issue number in the commit messages, for example: `close #12345`, or `ref #12345`.


<!--Checker Notification Identifier-->

codecov[bot]: # [Codecov](https://codecov.io/gh/tikv/pd/pull/4219?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv) Report
> Merging [#4219](https://codecov.io/gh/tikv/pd/pull/4219?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv) (27f1c6f) into [release-4.0](https://codecov.io/gh/tikv/pd/commit/605dc238b9a9e54491a5e6b0843f30de33421402?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv) (605dc23) will **increase** coverage by `0.08%`.
> The diff coverage is `84.09%`.

[![Impacted file tree graph](https://codecov.io/gh/tikv/pd/pull/4219/graphs/tree.svg?width=650&height=150&src=pr&token=J1yRL471OI&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv)](https://codecov.io/gh/tikv/pd/pull/4219?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv)

```diff
@@               Coverage Diff               @@
##           release-4.0    #4219      +/-   ##
===============================================
+ Coverage        74.87%   74.95%   +0.08%     
===============================================
  Files              211      211              
  Lines            20061    20061              
===============================================
+ Hits             15020    15037      +17     
+ Misses            3703     3699       -4     
+ Partials          1338     1325      -13     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `74.95% <84.09%> (+0.08%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/tikv/pd/pull/4219?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv) | Coverage Δ | |
|---|---|---|
| [server/cluster/cluster.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-c2VydmVyL2NsdXN0ZXIvY2x1c3Rlci5nbw==) | `77.14% <0.00%> (+0.25%)` | :arrow_up: |
| [server/core/storage.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-c2VydmVyL2NvcmUvc3RvcmFnZS5nbw==) | `68.39% <75.00%> (+0.86%)` | :arrow_up: |
| [server/region\_syncer/server.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-c2VydmVyL3JlZ2lvbl9zeW5jZXIvc2VydmVyLmdv) | `79.41% <75.00%> (+0.15%)` | :arrow_up: |
| [server/region\_syncer/client.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-c2VydmVyL3JlZ2lvbl9zeW5jZXIvY2xpZW50Lmdv) | `78.90% <90.00%> (+0.96%)` | :arrow_up: |
| [server/core/region\_storage.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-c2VydmVyL2NvcmUvcmVnaW9uX3N0b3JhZ2UuZ28=) | `80.23% <100.00%> (+1.75%)` | :arrow_up: |
| [pkg/mock/mockhbstream/mockhbstream.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-cGtnL21vY2svbW9ja2hic3RyZWFtL21vY2toYnN0cmVhbS5nbw==) | `87.03% <0.00%> (-1.86%)` | :arrow_down: |
| [server/grpc\_service.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-c2VydmVyL2dycGNfc2VydmljZS5nbw==) | `58.87% <0.00%> (-1.42%)` | :arrow_down: |
| [client/client.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-Y2xpZW50L2NsaWVudC5nbw==) | `68.52% <0.00%> (-1.02%)` | :arrow_down: |
| [server/member/leader.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-c2VydmVyL21lbWJlci9sZWFkZXIuZ28=) | `67.88% <0.00%> (-0.46%)` | :arrow_down: |
| [pkg/btree/btree.go](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv#diff-cGtnL2J0cmVlL2J0cmVlLmdv) | `86.02% <0.00%> (-0.25%)` | :arrow_down: |
| ... and [7 more](https://codecov.io/gh/tikv/pd/pull/4219/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/tikv/pd/pull/4219?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/tikv/pd/pull/4219?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv). Last update [605dc23...27f1c6f](https://codecov.io/gh/tikv/pd/pull/4219?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=tikv).

nolouch: /merge
ti-chi-bot: @nolouch: It seems you want to merge this PR, I will help you trigger all the tests:

/run-all-tests


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: 7ed160febbd1d5023993a64025e41bc66418dade</details>
ti-chi-bot: @ti-chi-bot: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>

Patch:
diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go
index ec0c9fea9..a449c5985 100644
--- a/server/cluster/cluster.go
+++ b/server/cluster/cluster.go
@@ -284,7 +284,7 @@ func (c *RaftCluster) LoadClusterInfo() (*RaftCluster, error) {
 	start = time.Now()
 
 	// used to load region from kv storage to cache storage.
-	if err := c.storage.LoadRegionsOnce(c.core.CheckAndPutRegion); err != nil {
+	if err := c.storage.LoadRegionsOnce(c.ctx, c.core.CheckAndPutRegion); err != nil {
 		return nil, err
 	}
 	log.Info("load regions",
diff --git a/server/core/region_storage.go b/server/core/region_storage.go
index be6de0d03..2eef4a389 100644
--- a/server/core/region_storage.go
+++ b/server/core/region_storage.go
@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/errs"
@@ -118,7 +119,7 @@ func deleteRegion(kv kv.Base, region *metapb.Region) error {
 	return kv.Remove(regionPath(region.GetId()))
 }
 
-func loadRegions(kv kv.Base, f func(region *RegionInfo) []*RegionInfo) error {
+func loadRegions(ctx context.Context, kv kv.Base, f func(region *RegionInfo) []*RegionInfo) error {
 	nextID := uint64(0)
 	endKey := regionPath(math.MaxUint64)
 
@@ -127,6 +128,10 @@ func loadRegions(kv kv.Base, f func(region *RegionInfo) []*RegionInfo) error {
 	// a variable rangeLimit to work around.
 	rangeLimit := maxKVRangeLimit
 	for {
+		failpoint.Inject("slowLoadRegion", func() {
+			rangeLimit = 1
+			time.Sleep(time.Second)
+		})
 		startKey := regionPath(nextID)
 		_, res, err := kv.LoadRange(startKey, endKey, rangeLimit)
 		if err != nil {
@@ -135,6 +140,11 @@ func loadRegions(kv kv.Base, f func(region *RegionInfo) []*RegionInfo) error {
 			}
 			return err
 		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
 
 		for _, s := range res {
 			region := &metapb.Region{}
diff --git a/server/core/storage.go b/server/core/storage.go
index 9ad32b421..66a9e6099 100644
--- a/server/core/storage.go
+++ b/server/core/storage.go
@@ -14,6 +14,7 @@
 package core
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"math"
@@ -160,22 +161,22 @@ func (s *Storage) LoadRegion(regionID uint64, region *metapb.Region) (bool, erro
 }
 
 // LoadRegions loads all regions from storage to RegionsInfo.
-func (s *Storage) LoadRegions(f func(region *RegionInfo) []*RegionInfo) error {
+func (s *Storage) LoadRegions(ctx context.Context, f func(region *RegionInfo) []*RegionInfo) error {
 	if atomic.LoadInt32(&s.useRegionStorage) > 0 {
-		return loadRegions(s.regionStorage, f)
+		return loadRegions(ctx, s.regionStorage, f)
 	}
-	return loadRegions(s.Base, f)
+	return loadRegions(ctx, s.Base, f)
 }
 
 // LoadRegionsOnce loads all regions from storage to RegionsInfo.Only load one time from regionStorage.
-func (s *Storage) LoadRegionsOnce(f func(region *RegionInfo) []*RegionInfo) error {
+func (s *Storage) LoadRegionsOnce(ctx context.Context, f func(region *RegionInfo) []*RegionInfo) error {
 	if atomic.LoadInt32(&s.useRegionStorage) == 0 {
-		return loadRegions(s.Base, f)
+		return loadRegions(ctx, s.Base, f)
 	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.regionLoaded == 0 {
-		if err := loadRegions(s.regionStorage, f); err != nil {
+		if err := loadRegions(ctx, s.regionStorage, f); err != nil {
 			return err
 		}
 		s.regionLoaded = 1
diff --git a/server/region_syncer/client.go b/server/region_syncer/client.go
index 007665739..9eb037c26 100644
--- a/server/region_syncer/client.go
+++ b/server/region_syncer/client.go
@@ -8,6 +8,7 @@
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
@@ -40,30 +41,22 @@ const (
 // StopSyncWithLeader stop to sync the region with leader.
 func (s *RegionSyncer) StopSyncWithLeader() {
 	s.reset()
-	s.Lock()
-	close(s.closed)
-	s.closed = make(chan struct{})
-	s.Unlock()
 	s.wg.Wait()
 }
 
 func (s *RegionSyncer) reset() {
-	s.Lock()
-	defer s.Unlock()
+	s.mu.Lock()
+	defer s.mu.Unlock()
 
-	if s.regionSyncerCancel == nil {
-		return
+	if s.mu.clientCancel != nil {
+		s.mu.clientCancel()
 	}
-	s.regionSyncerCancel()
-	s.regionSyncerCancel, s.regionSyncerCtx = nil, nil
+	s.mu.clientCancel, s.mu.clientCtx = nil, nil
 }
 
-func (s *RegionSyncer) establish(addr string) (*grpc.ClientConn, error) {
-	s.reset()
-	ctx, cancel := context.WithCancel(s.server.LoopContext())
+func (s *RegionSyncer) establish(ctx context.Context, addr string) (*grpc.ClientConn, error) {
 	tlsCfg, err := s.securityConfig.ToTLSConfig()
 	if err != nil {
-		cancel()
 		return nil, err
 	}
 	cc, err := grpcutil.GetClientConn(
@@ -88,21 +81,16 @@ func (s *RegionSyncer) establish(addr string) (*grpc.ClientConn, error) {
 		grpc.WithBlock(),
 	)
 	if err != nil {
-		cancel()
 		return nil, errors.WithStack(err)
 	}
-
-	s.Lock()
-	s.regionSyncerCtx, s.regionSyncerCancel = ctx, cancel
-	s.Unlock()
 	return cc, nil
 }
 
-func (s *RegionSyncer) syncRegion(conn *grpc.ClientConn) (ClientStream, error) {
+func (s *RegionSyncer) syncRegion(ctx context.Context, conn *grpc.ClientConn) (ClientStream, error) {
 	cli := pdpb.NewPDClient(conn)
-	syncStream, err := cli.SyncRegions(s.regionSyncerCtx)
+	syncStream, err := cli.SyncRegions(ctx)
 	if err != nil {
-		return syncStream, errs.ErrGRPCCreateStream.Wrap(err).FastGenWithCause()
+		return nil, errs.ErrGRPCCreateStream.Wrap(err).FastGenWithCause()
 	}
 	err = syncStream.Send(&pdpb.SyncRegionRequest{
 		Header:     &pdpb.RequestHeader{ClusterId: s.server.ClusterID()},
@@ -110,7 +98,7 @@ func (s *RegionSyncer) syncRegion(conn *grpc.ClientConn) (ClientStream, error) {
 		StartIndex: s.history.GetNextIndex(),
 	})
 	if err != nil {
-		return syncStream, errs.ErrGRPCSend.Wrap(err).FastGenWithCause()
+		return nil, errs.ErrGRPCSend.Wrap(err).FastGenWithCause()
 	}
 
 	return syncStream, nil
@@ -121,15 +109,21 @@ var regionGuide = core.GenerateRegionGuideFunc(false)
 // StartSyncWithLeader starts to sync with leader.
 func (s *RegionSyncer) StartSyncWithLeader(addr string) {
 	s.wg.Add(1)
-	s.RLock()
-	closed := s.closed
-	s.RUnlock()
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.mu.clientCtx, s.mu.clientCancel = context.WithCancel(s.server.LoopContext())
+	ctx := s.mu.clientCtx
+
 	go func() {
 		defer s.wg.Done()
 		// used to load region from kv storage to cache storage.
 		bc := s.server.GetBasicCluster()
 		storage := s.server.GetStorage()
-		err := storage.LoadRegionsOnce(bc.CheckAndPutRegion)
+		log.Info("region syncer start load region")
+		start := time.Now()
+		err := storage.LoadRegionsOnce(ctx, bc.CheckAndPutRegion)
+		log.Info("region syncer finished load region", zap.Duration("time-cost", time.Since(start)))
 		if err != nil {
 			log.Warn("failed to load regions.", errs.ZapError(err))
 		}
@@ -137,11 +131,11 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
 		var conn *grpc.ClientConn
 		for {
 			select {
-			case <-closed:
+			case <-ctx.Done():
 				return
 			default:
 			}
-			conn, err = s.establish(addr)
+			conn, err = s.establish(ctx, addr)
 			if err != nil {
 				log.Error("cannot establish connection with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName()), errs.ZapError(err))
 				continue
@@ -153,12 +147,12 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
 		// Start syncing data.
 		for {
 			select {
-			case <-closed:
+			case <-ctx.Done():
 				return
 			default:
 			}
 
-			stream, err := s.syncRegion(conn)
+			stream, err := s.syncRegion(ctx, conn)
 			if err != nil {
 				if ev, ok := status.FromError(err); ok {
 					if ev.Code() == codes.Canceled {
diff --git a/server/region_syncer/server.go b/server/region_syncer/server.go
index ba163866d..34609e54c 100644
--- a/server/region_syncer/server.go
+++ b/server/region_syncer/server.go
@@ -8,6 +8,7 @@
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
@@ -67,16 +68,17 @@ type Server interface {
 
 // RegionSyncer is used to sync the region information without raft.
 type RegionSyncer struct {
-	sync.RWMutex
-	streams            map[string]ServerStream
-	regionSyncerCtx    context.Context
-	regionSyncerCancel context.CancelFunc
-	server             Server
-	closed             chan struct{}
-	wg                 sync.WaitGroup
-	history            *historyBuffer
-	limit              *ratelimit.Bucket
-	securityConfig     *grpcutil.SecurityConfig
+	mu struct {
+		sync.RWMutex
+		streams      map[string]ServerStream
+		clientCtx    context.Context
+		clientCancel context.CancelFunc
+	}
+	server         Server
+	wg             sync.WaitGroup
+	history        *historyBuffer
+	limit          *ratelimit.Bucket
+	securityConfig *grpcutil.SecurityConfig
 }
 
 // NewRegionSyncer returns a region syncer.
@@ -85,14 +87,14 @@ type RegionSyncer struct {
 // Usually open the region syncer in huge cluster and the server
 // no longer etcd but go-leveldb.
 func NewRegionSyncer(s Server) *RegionSyncer {
-	return &RegionSyncer{
-		streams:        make(map[string]ServerStream),
+	syncer := &RegionSyncer{
 		server:         s,
-		closed:         make(chan struct{}),
 		history:        newHistoryBuffer(defaultHistoryBufferSize, s.GetStorage().GetRegionStorage()),
 		limit:          ratelimit.NewBucketWithRate(defaultBucketRate, defaultBucketCapacity),
 		securityConfig: s.GetSecurityConfig(),
 	}
+	syncer.mu.streams = make(map[string]ServerStream)
+	return syncer
 }
 
 // RunServer runs the server of the region syncer.
@@ -102,6 +104,7 @@ func (s *RegionSyncer) RunServer(regionNotifier <-chan *core.RegionInfo, quit ch
 	var stats []*pdpb.RegionStat
 	var leaders []*metapb.Peer
 	ticker := time.NewTicker(syncerKeepAliveInterval)
+	defer ticker.Stop()
 	for {
 		select {
 		case <-quit:
@@ -251,28 +254,28 @@ func (s *RegionSyncer) syncHistoryRegion(request *pdpb.SyncRegionRequest, stream
 
 // bindStream binds the established server stream.
 func (s *RegionSyncer) bindStream(name string, stream ServerStream) {
-	s.Lock()
-	defer s.Unlock()
-	s.streams[name] = stream
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.mu.streams[name] = stream
 }
 
 func (s *RegionSyncer) broadcast(regions *pdpb.SyncRegionResponse) {
 	var failed []string
-	s.RLock()
-	for name, sender := range s.streams {
+	s.mu.RLock()
+	for name, sender := range s.mu.streams {
 		err := sender.Send(regions)
 		if err != nil {
 			log.Error("region syncer send data meet error", errs.ZapError(errs.ErrGRPCSend, err))
 			failed = append(failed, name)
 		}
 	}
-	s.RUnlock()
+	s.mu.RUnlock()
 	if len(failed) > 0 {
-		s.Lock()
+		s.mu.Lock()
 		for _, name := range failed {
-			delete(s.streams, name)
+			delete(s.mu.streams, name)
 			log.Info("region syncer delete the stream", zap.String("stream", name))
 		}
-		s.Unlock()
+		s.mu.Unlock()
 	}
 }


Test patch:
diff --git a/server/core/storage_test.go b/server/core/storage_test.go
index 95f8b8329..6696f7e13 100644
--- a/server/core/storage_test.go
+++ b/server/core/storage_test.go
@@ -14,6 +14,7 @@
 package core
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"math"
@@ -142,7 +143,7 @@ func (s *testKVSuite) TestLoadRegions(c *C) {
 
 	n := 10
 	regions := mustSaveRegions(c, storage, n)
-	c.Assert(storage.LoadRegions(cache.SetRegion), IsNil)
+	c.Assert(storage.LoadRegions(context.Background(), cache.SetRegion), IsNil)
 
 	c.Assert(cache.GetRegionCount(), Equals, n)
 	for _, region := range cache.GetMetaRegions() {
@@ -156,7 +157,7 @@ func (s *testKVSuite) TestLoadRegionsToCache(c *C) {
 
 	n := 10
 	regions := mustSaveRegions(c, storage, n)
-	c.Assert(storage.LoadRegionsOnce(cache.SetRegion), IsNil)
+	c.Assert(storage.LoadRegionsOnce(context.Background(), cache.SetRegion), IsNil)
 
 	c.Assert(cache.GetRegionCount(), Equals, n)
 	for _, region := range cache.GetMetaRegions() {
@@ -165,7 +166,7 @@ func (s *testKVSuite) TestLoadRegionsToCache(c *C) {
 
 	n = 20
 	mustSaveRegions(c, storage, n)
-	c.Assert(storage.LoadRegionsOnce(cache.SetRegion), IsNil)
+	c.Assert(storage.LoadRegionsOnce(context.Background(), cache.SetRegion), IsNil)
 	c.Assert(cache.GetRegionCount(), Equals, n)
 }
 
@@ -175,7 +176,7 @@ func (s *testKVSuite) TestLoadRegionsExceedRangeLimit(c *C) {
 
 	n := 1000
 	regions := mustSaveRegions(c, storage, n)
-	c.Assert(storage.LoadRegions(cache.SetRegion), IsNil)
+	c.Assert(storage.LoadRegions(context.Background(), cache.SetRegion), IsNil)
 	c.Assert(cache.GetRegionCount(), Equals, n)
 	for _, region := range cache.GetMetaRegions() {
 		c.Assert(region, DeepEquals, regions[region.GetId()])
diff --git a/server/region_syncer/client_test.go b/server/region_syncer/client_test.go
new file mode 100644
index 000000000..48f9d4137
--- /dev/null
+++ b/server/region_syncer/client_test.go
@@ -0,0 +1,108 @@
+// Copyright 2021 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package syncer
+
+import (
+	"context"
+	"io/ioutil"
+	"os"
+	"time"
+
+	. "github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
+	"github.com/pingcap/kvproto/pkg/metapb"
+	"github.com/pingcap/kvproto/pkg/pdpb"
+	"github.com/tikv/pd/pkg/grpcutil"
+	"github.com/tikv/pd/server/core"
+	"github.com/tikv/pd/server/kv"
+)
+
+var _ = Suite(&testClientSuite{})
+
+type testClientSuite struct{}
+
+// For issue https://github.com/tikv/pd/issues/3936
+func (t *testClientSuite) TestLoadRegion(c *C) {
+	tempDir, err := ioutil.TempDir(os.TempDir(), "region_syncer_load_region")
+	c.Assert(err, IsNil)
+	defer os.RemoveAll(tempDir)
+	rs, err := core.NewRegionStorage(context.Background(), tempDir)
+	c.Assert(err, IsNil)
+
+	storage := core.NewStorage(kv.NewMemoryKV())
+	storage.SetRegionStorage(rs)
+
+	server := &mockServer{
+		ctx:     context.Background(),
+		storage: storage,
+		bc:      core.NewBasicCluster(),
+	}
+	for i := 0; i < 30; i++ {
+		rs.SaveRegion(&metapb.Region{Id: uint64(i) + 1})
+	}
+	c.Assert(failpoint.Enable("github.com/tikv/pd/server/core/slowLoadRegion", "return(true)"), IsNil)
+	defer func() { c.Assert(failpoint.Disable("github.com/tikv/pd/server/core/slowLoadRegion"), IsNil) }()
+
+	rc := NewRegionSyncer(server)
+	start := time.Now()
+	rc.StartSyncWithLeader("")
+	time.Sleep(time.Second)
+	rc.StopSyncWithLeader()
+	c.Assert(time.Since(start), Greater, time.Second) // make sure failpoint is injected
+	c.Assert(time.Since(start), Less, time.Second*2)
+}
+
+type mockServer struct {
+	ctx            context.Context
+	member, leader *pdpb.Member
+	storage        *core.Storage
+	bc             *core.BasicCluster
+}
+
+func (s *mockServer) LoopContext() context.Context {
+	return s.ctx
+}
+
+func (s *mockServer) ClusterID() uint64 {
+	return 1
+}
+
+func (s *mockServer) GetMemberInfo() *pdpb.Member {
+	return s.member
+}
+
+func (s *mockServer) GetLeader() *pdpb.Member {
+	return s.leader
+}
+
+func (s *mockServer) GetStorage() *core.Storage {
+	return s.storage
+}
+
+func (s *mockServer) Name() string {
+	return "mock-server"
+}
+
+func (s *mockServer) GetRegions() []*core.RegionInfo {
+	return s.bc.GetRegions()
+}
+
+func (s *mockServer) GetSecurityConfig() *grpcutil.SecurityConfig {
+	return &grpcutil.SecurityConfig{}
+}
+
+func (s *mockServer) GetBasicCluster() *core.BasicCluster {
+	return s.bc
+}
diff --git a/tests/server/cluster/cluster_test.go b/tests/server/cluster/cluster_test.go
index 7f0fa1b0d..9b4a595ad 100644
--- a/tests/server/cluster/cluster_test.go
+++ b/tests/server/cluster/cluster_test.go
@@ -718,7 +718,7 @@ func (s *clusterTestSuite) TestLoadClusterInfo(c *C) {
 	for _, region := range regions {
 		c.Assert(storage.SaveRegion(region), IsNil)
 	}
-	raftCluster.GetStorage().LoadRegionsOnce(raftCluster.GetCacheCluster().PutRegion)
+	raftCluster.GetStorage().LoadRegionsOnce(s.ctx, raftCluster.GetCacheCluster().PutRegion)
 	c.Assert(raftCluster.GetRegionCount(), Equals, n)
 }