solve ABA

Signed-off-by: bufferflies <1045931706@qq.com>

server/cluster/cluster.go

@@ -613,13 +613,22 @@ func (c *RaftCluster) processBucketHeartbeat(buckets *metapb.Buckets) error {
 		bucketEventCounter.WithLabelValues("region_cache_miss").Inc()
 		return errors.Errorf("region %v not found", buckets.GetRegionId())
 	}
-
-	// region should not update if the version of the buckets is less than the old one.
-	if old := region.GetBuckets(); old != nil && old.Version >= buckets.Version {
-		bucketEventCounter.WithLabelValues("version_not_match").Inc()
-		return nil
+	// use CAS to update the bucket information.
+	// the two request(A:3,B:2) get the same region and need to update the buckets.
+	// the A will pass the check and set the version to 3, the B will fail because the region.bucket has changed.
+	// the retry should keep the old version and the new version will be set to the region.bucket, like two requests (A:2,B:3).
+	for retry := 0; retry < 3; retry++ {
+		old := region.GetBuckets()
+		// region should not update if the version of the buckets is less than the old one.
+		if old != nil && buckets.GetVersion() <= old.GetVersion() {
+			bucketEventCounter.WithLabelValues("version_not_match").Inc()
+			return nil
+		}
+		if ok := region.UpdateBuckets(buckets, old); ok {
+			return nil
+		}
 	}
-	region.UpdateBuckets(buckets)
+	bucketEventCounter.WithLabelValues("bucket_update_failed").Inc()
 	return nil
 }
 

server/core/region.go

@@ -414,18 +414,18 @@ func (r *RegionInfo) GetStat() *pdpb.RegionStat {
 }
 
 // UpdateBuckets sets the buckets of the region.
-func (r *RegionInfo) UpdateBuckets(buckets *metapb.Buckets) {
+func (r *RegionInfo) UpdateBuckets(buckets, old *metapb.Buckets) bool {
 	// the bucket can't be nil except in the test cases.
 	if buckets == nil {
-		return
+		return true
 	}
-	// only need to update bucket keys,versions.
+	// only need to update bucket keys, versions.
 	newBuckets := &metapb.Buckets{
 		RegionId: buckets.GetRegionId(),
 		Version:  buckets.GetVersion(),
 		Keys:     buckets.GetKeys(),
 	}
-	atomic.StorePointer(&r.buckets, unsafe.Pointer(newBuckets))
+	return atomic.CompareAndSwapPointer(&r.buckets, unsafe.Pointer(old), unsafe.Pointer(newBuckets))
 }
 
 // GetBuckets returns the buckets of the region.

server/core/region_test.go

@@ -186,24 +186,23 @@ func (s *testRegionInfoSuite) TestInherit(c *C) {
 
 	// bucket
 	data := []struct {
-		originExist   bool
 		originBuckets *metapb.Buckets
 		buckets       *metapb.Buckets
 		same          bool
 	}{
-		{false, nil, nil, true},
-		{false, nil, &metapb.Buckets{RegionId: 1, Version: 2}, false},
-		{true, &metapb.Buckets{RegionId: 1, Version: 2}, &metapb.Buckets{RegionId: 1, Version: 3}, false},
-		{true, &metapb.Buckets{RegionId: 1, Version: 2}, nil, true},
+		{nil, nil, true},
+		{nil, &metapb.Buckets{RegionId: 1, Version: 2}, false},
+		{&metapb.Buckets{RegionId: 1, Version: 2}, &metapb.Buckets{RegionId: 1, Version: 3}, false},
+		{&metapb.Buckets{RegionId: 1, Version: 2}, nil, true},
 	}
 	for _, d := range data {
 		var origin *RegionInfo
-		if d.originExist {
+		if d.originBuckets != nil {
 			origin = NewRegionInfo(&metapb.Region{Id: 100}, nil)
-			origin.UpdateBuckets(d.originBuckets)
+			origin.UpdateBuckets(d.originBuckets, origin.GetBuckets())
 		}
 		r := NewRegionInfo(&metapb.Region{Id: 100}, nil)
-		r.UpdateBuckets(d.buckets)
+		r.UpdateBuckets(d.buckets, r.GetBuckets())
 		r.Inherit(origin)
 		if d.same {
 			c.Assert(r.GetBuckets(), DeepEquals, d.originBuckets)