add ConcurrentReportBucket

Signed-off-by: bufferflies <1045931706@qq.com>
tikv · Mar 31, 2022 · f2d0db6 · f2d0db6
1 parent d34547c
commit f2d0db6
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 11 deletions.
diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go
@@ -606,8 +606,8 @@ func (c *RaftCluster) HandleStoreHeartbeat(stats *pdpb.StoreStats) error {
 	return nil
 }
 
-// processBucketHeartbeat update the bucket information.
-func (c *RaftCluster) processBucketHeartbeat(buckets *metapb.Buckets) error {
+// processReportBuckets update the bucket information.
+func (c *RaftCluster) processReportBuckets(buckets *metapb.Buckets) error {
 	region := c.core.GetRegion(buckets.GetRegionId())
 	if region == nil {
 		bucketEventCounter.WithLabelValues("region_cache_miss").Inc()
@@ -624,11 +624,14 @@ func (c *RaftCluster) processBucketHeartbeat(buckets *metapb.Buckets) error {
 			bucketEventCounter.WithLabelValues("version_not_match").Inc()
 			return nil
 		}
+		failpoint.Inject("concurrentBucketHeartbeat", func() {
+			time.Sleep(500 * time.Millisecond)
+		})
 		if ok := region.UpdateBuckets(buckets, old); ok {
 			return nil
 		}
 	}
-	bucketEventCounter.WithLabelValues("bucket_update_failed").Inc()
+	bucketEventCounter.WithLabelValues("update_failed").Inc()
 	return nil
 }
 

diff --git a/server/cluster/cluster_test.go b/server/cluster/cluster_test.go
@@ -439,7 +439,7 @@ func (s *testClusterInfoSuite) TestBucketHeartbeat(c *C) {
 		Version:  1,
 		Keys:     [][]byte{{'1'}, {'2'}},
 	}
-	c.Assert(cluster.processBucketHeartbeat(buckets), NotNil)
+	c.Assert(cluster.processReportBuckets(buckets), NotNil)
 
 	// case2: bucket can be processed after the region update.
 	stores := newTestStores(3, "2.0.0")
@@ -451,14 +451,14 @@ func (s *testClusterInfoSuite) TestBucketHeartbeat(c *C) {
 
 	c.Assert(cluster.processRegionHeartbeat(regions[0]), IsNil)
 	c.Assert(cluster.GetRegion(uint64(0)).GetBuckets(), IsNil)
-	c.Assert(cluster.processBucketHeartbeat(buckets), IsNil)
+	c.Assert(cluster.processReportBuckets(buckets), IsNil)
 	c.Assert(cluster.GetRegion(uint64(0)).GetBuckets(), DeepEquals, buckets)
 
 	// case3: the bucket version is same.
-	c.Assert(cluster.processBucketHeartbeat(buckets), IsNil)
+	c.Assert(cluster.processReportBuckets(buckets), IsNil)
 	// case4: the bucket version is changed.
 	buckets.Version = 3
-	c.Assert(cluster.processBucketHeartbeat(buckets), IsNil)
+	c.Assert(cluster.processReportBuckets(buckets), IsNil)
 	c.Assert(cluster.GetRegion(uint64(0)).GetBuckets(), DeepEquals, buckets)
 
 	//case5: region update should inherit buckets.
@@ -702,6 +702,32 @@ func (s *testClusterInfoSuite) TestRegionFlowChanged(c *C) {
 	c.Assert(newRegion.GetBytesRead(), Equals, uint64(1000))
 }
 
+func (s *testClusterInfoSuite) TestConcurrentReportBucket(c *C) {
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
+	cluster := newTestRaftCluster(s.ctx, mockid.NewIDAllocator(), opt, storage.NewStorageWithMemoryBackend(), core.NewBasicCluster())
+	cluster.coordinator = newCoordinator(s.ctx, cluster, nil)
+
+	regions := []*core.RegionInfo{core.NewTestRegionInfo([]byte{}, []byte{})}
+	heartbeatRegions(c, cluster, regions)
+	c.Assert(cluster.GetRegion(0), NotNil)
+
+	bucket1 := &metapb.Buckets{RegionId: 0, Version: 3}
+	bucket2 := &metapb.Buckets{RegionId: 0, Version: 2}
+	var wg sync.WaitGroup
+	wg.Add(1)
+	c.Assert(failpoint.Enable("github.com/tikv/pd/server/cluster/concurrentBucketHeartbeat", "return(true)"), IsNil)
+	go func() {
+		defer wg.Done()
+		cluster.processReportBuckets(bucket1)
+	}()
+	time.Sleep(100 * time.Millisecond)
+	c.Assert(failpoint.Disable("github.com/tikv/pd/server/cluster/concurrentBucketHeartbeat"), IsNil)
+	c.Assert(cluster.processReportBuckets(bucket2), IsNil)
+	wg.Wait()
+	c.Assert(cluster.GetRegion(0).GetBuckets(), DeepEquals, bucket1)
+}
+
 func (s *testClusterInfoSuite) TestConcurrentRegionHeartbeat(c *C) {
 	_, opt, err := newTestScheduleConfig()
 	c.Assert(err, IsNil)

diff --git a/server/cluster/cluster_worker.go b/server/cluster/cluster_worker.go
@@ -224,7 +224,7 @@ func (c *RaftCluster) HandleBatchReportSplit(request *pdpb.ReportBatchSplitReque
 	return &pdpb.ReportBatchSplitResponse{}, nil
 }
 
-// HandleBucketHeartbeat processes RegionInfo reports from client
-func (c *RaftCluster) HandleBucketHeartbeat(buckets *metapb.Buckets) error {
-	return c.processBucketHeartbeat(buckets)
+// HandleReportBuckets processes RegionInfo reports from client
+func (c *RaftCluster) HandleReportBuckets(buckets *metapb.Buckets) error {
+	return c.processReportBuckets(buckets)
 }
diff --git a/server/grpc_service.go b/server/grpc_service.go
@@ -801,7 +801,7 @@ func (s *GrpcServer) ReportBuckets(stream pdpb.PD_ReportBucketsServer) error {
 		bucketReportCounter.WithLabelValues(storeAddress, storeLabel, "report", "recv").Inc()
 
 		start := time.Now()
-		err = rc.HandleBucketHeartbeat(buckets)
+		err = rc.HandleReportBuckets(buckets)
 		if err != nil {
 			bucketReportCounter.WithLabelValues(storeAddress, storeLabel, "report", "err").Inc()
 			continue