fix(master): long-term unused slot activation employs a gradual addition approach.

with: #1000292905

Signed-off-by: clinx <chenlin1@oppo.com>
(cherry picked from commit bdf08f93a5)

master/flash_node_topology.go

@@ -84,8 +84,6 @@ func (c *Cluster) scheduleToUpdateFlashGroupRespCache() {
 		for {
 			if c.partition != nil && c.partition.IsRaftLeader() {
 				c.flashNodeTopo.UpdateClientResponse()
-				// sync fg if slots changed
-				c.flashNodeTopo.UpdateFlashGroup(c.syncUpdateFlashGroup)
 			}
 			select {
 			case <-c.stopc:

remotecache/flashgroupmanager/flash_group.go

@@ -1,7 +1,6 @@
 package flashgroupmanager
 
 import (
-	"math/rand"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -28,6 +27,8 @@ type FlashGroupValue struct {
 	Status           proto.FlashGroupStatus
 	LostAllFlashNode int32
 	ReducingSlots    int32
+	IncreasingSlots  int32
+	ReduceAllTime    int64 // unix second
 	SlotChanged      int32
 }
 
@@ -121,11 +122,38 @@ func (fg *FlashGroup) ReduceSlot(syncFlashGroupFunc SyncUpdateFlashGroupFunc) {
 	}()
 }
 
+func (fg *FlashGroup) IncreaseSlot(syncFlashGroupFunc SyncUpdateFlashGroupFunc) {
+	if atomic.CompareAndSwapInt32(&fg.IncreasingSlots, 0, 1) {
+		return
+	}
+	if log.EnableDebug() {
+		log.LogDebugf("flashgroup %v is increasing slots", fg)
+	}
+	totalSlots := len(fg.Slots) + len(fg.ReservedSlots)
+	numToSelect := (totalSlots + 8 - 1) / 8
+	fg.executeIncreaseSlot(numToSelect, syncFlashGroupFunc)
+	go func() {
+		ticker := time.NewTicker(20 * time.Second)
+		defer func() {
+			atomic.StoreInt32(&fg.IncreasingSlots, 0)
+			ticker.Stop()
+		}()
+		var i int
+		for {
+			if len(fg.ReservedSlots) == 0 {
+				return
+			}
+			<-ticker.C
+			i++
+			fg.executeIncreaseSlot(numToSelect, syncFlashGroupFunc)
+		}
+	}()
+}
+
 func (fg *FlashGroup) executeReduceSlot(numToReduce int, syncFlashGroupFunc SyncUpdateFlashGroupFunc) {
 	if len(fg.Slots) == 0 {
 		return
 	}
-	rand.Seed(time.Now().UnixNano())
 	fg.lock.Lock()
 	defer fg.lock.Unlock()
 	numToReduce = util.Min(numToReduce, len(fg.Slots))
@@ -140,9 +168,43 @@ func (fg *FlashGroup) executeReduceSlot(numToReduce int, syncFlashGroupFunc Sync
 	}
 	fg.ReservedSlots = append(fg.ReservedSlots, fg.Slots[:numToReduce]...)
 	fg.Slots = fg.Slots[numToReduce:]
+	if len(fg.Slots) == 0 {
+		fg.ReduceAllTime = time.Now().Unix()
+	}
 	if err := syncFlashGroupFunc(fg); err != nil {
 		fg.Slots = oldSlots
 		fg.ReservedSlots = oldReservedSlots
+		fg.ReduceAllTime = 0
+	}
+}
+
+func (fg *FlashGroup) executeIncreaseSlot(numToIncrease int, syncFlashGroupFunc SyncUpdateFlashGroupFunc) {
+	if len(fg.ReservedSlots) == 0 {
+		return
+	}
+	fg.lock.Lock()
+	defer fg.lock.Unlock()
+	numToIncrease = util.Min(numToIncrease, len(fg.ReservedSlots))
+	if numToIncrease == 0 {
+		return
+	}
+	var oldSlots, oldReservedSlots []uint32
+	oldSlots = append(oldSlots, fg.Slots...)
+	oldReservedSlots = append(oldReservedSlots, fg.ReservedSlots...)
+	if log.EnableInfo() {
+		log.LogInfof("executeIncreaseSlot fg(%v) oldSlots(%v) oldReservedSlots(%v)", fg.ID, oldSlots, oldReservedSlots)
+	}
+	fg.Slots = append(fg.Slots, fg.ReservedSlots[:numToIncrease]...)
+	fg.ReservedSlots = fg.ReservedSlots[numToIncrease:]
+	if len(fg.ReservedSlots) == 0 {
+		fg.ReduceAllTime = 0
+	}
+	if err := syncFlashGroupFunc(fg); err != nil {
+		fg.Slots = oldSlots
+		fg.ReservedSlots = oldReservedSlots
+		if len(fg.ReservedSlots) == 0 {
+			fg.ReduceAllTime = 0
+		}
 	}
 }
 

remotecache/flashgroupmanager/flash_group_test.go

@@ -336,3 +336,261 @@ func TestFlashGroup_ConcurrentAccess(t *testing.T) {
 	assert.Equal(t, 8, len(fg.Slots))
 	assert.Equal(t, numGoroutines, len(fg.flashNodes))
 }
+
+// TestFlashGroup_IncreaseSlot_AlreadyIncreasing tests that IncreaseSlot doesn't start multiple goroutines
+func TestFlashGroup_IncreaseSlot_AlreadyIncreasing(t *testing.T) {
+	fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+
+	// Add some reserved slots
+	fg.ReservedSlots = []uint32{4, 5, 6, 7, 8}
+
+	// Set the flag to indicate increase is already in progress
+	atomic.StoreInt32(&fg.IncreasingSlots, 1)
+
+	// This should return immediately without starting a new goroutine
+	fg.IncreaseSlot(tSyncUpdateFlashGroup)
+
+	// Verify the flag is still set
+	assert.Equal(t, int32(1), atomic.LoadInt32(&fg.IncreasingSlots))
+}
+
+// TestFlashGroup_IncreaseSlot_NoReservedSlots tests IncreaseSlot behavior when there are no reserved slots
+func TestFlashGroup_IncreaseSlot_NoReservedSlots(t *testing.T) {
+	fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+
+	// No reserved slots
+	assert.Equal(t, 0, len(fg.ReservedSlots))
+
+	// Start increase
+	fg.IncreaseSlot(tSyncUpdateFlashGroup)
+
+	// Wait a bit for the goroutine to process
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify no slots were processed since there were no reserved slots
+	assert.Equal(t, 3, len(fg.Slots))
+	assert.Equal(t, 0, len(fg.ReservedSlots))
+}
+
+// TestFlashGroup_IncreaseSlot_WithReservedSlots tests the complete IncreaseSlot functionality
+func TestFlashGroup_IncreaseSlot_WithReservedSlots(t *testing.T) {
+	// Create a flash group with 3 slots and 8 reserved slots
+	initialSlots := []uint32{1, 2, 3}
+	reservedSlots := []uint32{4, 5, 6, 7, 8, 9, 10, 11}
+	fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+	fg.ReservedSlots = reservedSlots
+
+	initialSlotCount := len(fg.Slots)
+	initialReservedCount := len(fg.ReservedSlots)
+
+	// Start increase
+	fg.IncreaseSlot(tSyncUpdateFlashGroup)
+
+	// Wait for the first increase cycle
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify that slots were moved from reserved to active
+	// Total slots should remain the same
+	assert.Equal(t, initialSlotCount+initialReservedCount, len(fg.Slots)+len(fg.ReservedSlots))
+
+	// Verify the increasing flag is set
+	assert.Equal(t, int32(1), atomic.LoadInt32(&fg.IncreasingSlots))
+}
+
+// TestFlashGroup_executeIncreaseSlot tests the executeIncreaseSlot method directly
+func TestFlashGroup_executeIncreaseSlot(t *testing.T) {
+	// Create a flash group with 3 slots and 8 reserved slots
+	initialSlots := []uint32{1, 2, 3}
+	reservedSlots := []uint32{4, 5, 6, 7, 8, 9, 10, 11}
+	fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+	fg.ReservedSlots = reservedSlots
+
+	initialSlotCount := len(fg.Slots)
+	initialReservedCount := len(fg.ReservedSlots)
+
+	// Calculate expected number to increase: (total + 8 - 1) / 8 = (11 + 7) / 8 = 2
+	totalSlots := initialSlotCount + initialReservedCount
+	expectedToIncrease := (totalSlots + 8 - 1) / 8
+
+	// Execute increase
+	fg.executeIncreaseSlot(expectedToIncrease, tSyncUpdateFlashGroup)
+
+	// Verify that slots were moved from reserved to active
+	expectedRemainingSlots := initialSlotCount + expectedToIncrease
+	expectedRemainingReserved := initialReservedCount - expectedToIncrease
+
+	assert.Equal(t, expectedRemainingSlots, len(fg.Slots))
+	assert.Equal(t, expectedRemainingReserved, len(fg.ReservedSlots))
+
+	// Verify total count remains the same
+	assert.Equal(t, totalSlots, len(fg.Slots)+len(fg.ReservedSlots))
+}
+
+// TestFlashGroup_executeIncreaseSlot_EmptyReservedSlots tests executeIncreaseSlot with no reserved slots
+func TestFlashGroup_executeIncreaseSlot_EmptyReservedSlots(t *testing.T) {
+	fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+
+	// No reserved slots
+	assert.Equal(t, 0, len(fg.ReservedSlots))
+
+	// This should not panic or cause issues
+	fg.executeIncreaseSlot(5, tSyncUpdateFlashGroup)
+
+	assert.Equal(t, 3, len(fg.Slots))
+	assert.Equal(t, 0, len(fg.ReservedSlots))
+}
+
+// TestFlashGroup_executeIncreaseSlot_SingleReservedSlot tests executeIncreaseSlot with only one reserved slot
+func TestFlashGroup_executeIncreaseSlot_SingleReservedSlot(t *testing.T) {
+	fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+	fg.ReservedSlots = []uint32{4}
+
+	initialSlotCount := len(fg.Slots)
+	initialReservedCount := len(fg.ReservedSlots)
+
+	fg.executeIncreaseSlot(2, tSyncUpdateFlashGroup)
+
+	// With 1 reserved slot, trying to increase by 2 should only move 1
+	assert.Equal(t, initialSlotCount+1, len(fg.Slots))
+	assert.Equal(t, 0, len(fg.ReservedSlots))
+
+	// Verify we had the expected number of reserved slots initially
+	assert.Equal(t, 1, initialReservedCount)
+}
+
+// TestFlashGroup_executeIncreaseSlot_AllReservedSlots tests executeIncreaseSlot moving all reserved slots
+func TestFlashGroup_executeIncreaseSlot_AllReservedSlots(t *testing.T) {
+	initialSlots := []uint32{1, 2, 3}
+	reservedSlots := []uint32{4, 5, 6}
+	fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+	fg.ReservedSlots = reservedSlots
+
+	initialSlotCount := len(fg.Slots)
+	initialReservedCount := len(fg.ReservedSlots)
+	totalSlots := initialSlotCount + initialReservedCount
+
+	// Try to increase by more than available reserved slots
+	fg.executeIncreaseSlot(10, tSyncUpdateFlashGroup)
+
+	// Should move all reserved slots
+	assert.Equal(t, totalSlots, len(fg.Slots))
+	assert.Equal(t, 0, len(fg.ReservedSlots))
+
+	// Verify the total count remains the same and initialReservedCount was used
+	assert.Equal(t, totalSlots, len(fg.Slots)+len(fg.ReservedSlots))
+	assert.Equal(t, initialReservedCount, 3) // Verify we had 3 reserved slots initially
+}
+
+// TestFlashGroup_executeIncreaseSlot_ReduceAllTimeReset tests that ReduceAllTime is reset when all reserved slots are consumed
+func TestFlashGroup_executeIncreaseSlot_ReduceAllTimeReset(t *testing.T) {
+	initialSlots := []uint32{1, 2, 3}
+	reservedSlots := []uint32{4, 5}
+	fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+	fg.ReservedSlots = reservedSlots
+
+	// Set a non-zero ReduceAllTime
+	fg.ReduceAllTime = time.Now().Unix()
+	assert.NotEqual(t, int64(0), fg.ReduceAllTime)
+
+	// Execute increase that will consume all reserved slots
+	fg.executeIncreaseSlot(10, tSyncUpdateFlashGroup)
+
+	// Verify ReduceAllTime is reset to 0
+	assert.Equal(t, int64(0), fg.ReduceAllTime)
+}
+
+// TestFlashGroup_executeIncreaseSlot_SyncError tests executeIncreaseSlot behavior when sync function fails
+func TestFlashGroup_executeIncreaseSlot_SyncError(t *testing.T) {
+	initialSlots := []uint32{1, 2, 3}
+	reservedSlots := []uint32{4, 5, 6}
+	fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+	fg.ReservedSlots = reservedSlots
+
+	initialSlotCount := len(fg.Slots)
+	initialReservedCount := len(fg.ReservedSlots)
+
+	// Create a sync function that always fails
+	failingSyncFunc := func(flashGroup *FlashGroup) error {
+		return fmt.Errorf("sync failed")
+	}
+
+	// Execute increase with failing sync
+	fg.executeIncreaseSlot(2, failingSyncFunc)
+
+	// Should rollback to original state due to sync failure
+	assert.Equal(t, initialSlotCount, len(fg.Slots))
+	assert.Equal(t, initialReservedCount, len(fg.ReservedSlots))
+}
+
+// TestFlashGroup_IncreaseSlot_ConcurrentExecution tests that IncreaseSlot handles concurrent execution properly
+func TestFlashGroup_IncreaseSlot_ConcurrentExecution(t *testing.T) {
+	initialSlots := []uint32{1, 2, 3}
+	reservedSlots := []uint32{4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}
+	fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100)
+	fg.ReservedSlots = reservedSlots
+
+	var wg sync.WaitGroup
+	numGoroutines := 5
+
+	// Try to start multiple IncreaseSlot operations concurrently
+	for i := 0; i < numGoroutines; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			fg.IncreaseSlot(tSyncUpdateFlashGroup)
+		}()
+	}
+
+	wg.Wait()
+
+	// Only one should succeed in setting the flag
+	// The others should return immediately
+	assert.Equal(t, int32(1), atomic.LoadInt32(&fg.IncreasingSlots))
+}
+
+// TestFlashGroup_IncreaseSlot_StepCalculation tests the step calculation logic
+func TestFlashGroup_IncreaseSlot_StepCalculation(t *testing.T) {
+	testCases := []struct {
+		name          string
+		slots         []uint32
+		reservedSlots []uint32
+		expectedStep  int
+	}{
+		{
+			name:          "5 total slots",
+			slots:         []uint32{1, 2, 3},
+			reservedSlots: []uint32{4, 5},
+			expectedStep:  1, // (5 + 7) / 8 = 12 / 8 = 1
+		},
+		{
+			name:          "11 total slots",
+			slots:         []uint32{1, 2, 3, 4, 5},
+			reservedSlots: []uint32{6, 7, 8, 9, 10, 11},
+			expectedStep:  2, // (11 + 7) / 8 = 18 / 8 = 2
+		},
+		{
+			name:          "16 total slots",
+			slots:         []uint32{1, 2, 3, 4, 5, 6, 7, 8},
+			reservedSlots: []uint32{9, 10, 11, 12, 13, 14, 15, 16},
+			expectedStep:  2, // (16 + 7) / 8 = 23 / 8 = 2
+		},
+		{
+			name:          "20 total slots",
+			slots:         []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
+			reservedSlots: []uint32{11, 12, 13, 14, 15, 16, 17, 18, 19, 20},
+			expectedStep:  3, // (20 + 7) / 8 = 27 / 8 = 3
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fg := newFlashGroup(1, tc.slots, proto.SlotStatus_Completed, []uint32{}, uint32(len(tc.slots)), proto.FlashGroupStatus_Active, 100)
+			fg.ReservedSlots = tc.reservedSlots
+
+			totalSlots := len(tc.slots) + len(tc.reservedSlots)
+			expectedStep := (totalSlots + 8 - 1) / 8
+
+			assert.Equal(t, tc.expectedStep, expectedStep)
+		})
+	}
+}

remotecache/flashgroupmanager/flash_node_topology.go

@@ -20,6 +20,7 @@ import (
 
 const (
 	defaultFlashGroupSlotsCount = 32
+	twoDaysInSeconds            = 2 * 24 * 60 * 60
 )
 
 type (
@@ -213,19 +214,24 @@ func (t *FlashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) {
 			} else {
 				atomic.StoreInt32(&fg.LostAllFlashNode, 0)
 				if len(fg.ReservedSlots) > 0 {
-					fg.lock.Lock()
+					if len(fg.Slots) == 0 && fg.ReduceAllTime != 0 && (time.Now().Unix()-fg.ReduceAllTime >= twoDaysInSeconds) {
-					oldSlots = append(oldSlots, fg.Slots...)
+						fg.IncreaseSlot(t.SyncFlashGroupFunc)
-					oldReservedSlots = append(oldReservedSlots, fg.ReservedSlots...)
+					} else if atomic.LoadInt32(&fg.IncreasingSlots) == 0 {
-					if log.EnableInfo() {
+						fg.lock.Lock()
-						log.LogInfof("recover fg(%v) oldSlots(%v) oldReservedSlots(%v)", fg.ID, oldSlots, oldReservedSlots)
+						oldSlots = append(oldSlots, fg.Slots...)
+						oldReservedSlots = append(oldReservedSlots, fg.ReservedSlots...)
+						if log.EnableInfo() {
+							log.LogInfof("recover fg(%v) oldSlots(%v) oldReservedSlots(%v)", fg.ID, oldSlots, oldReservedSlots)
+						}
+						fg.Slots = append(fg.Slots, fg.ReservedSlots...)
+						fg.ReservedSlots = make([]uint32, 0)
+						fg.ReduceAllTime = 0
+						if err := t.SyncFlashGroupFunc(fg); err != nil {
+							fg.Slots = oldSlots
+							fg.ReservedSlots = oldReservedSlots
+						}
+						fg.lock.Unlock()
 					}
-					fg.Slots = append(fg.Slots, fg.ReservedSlots...)
-					fg.ReservedSlots = make([]uint32, 0)
-					if err := t.SyncFlashGroupFunc(fg); err != nil {
-						fg.Slots = oldSlots
-						fg.ReservedSlots = oldReservedSlots
-					}
-					fg.lock.Unlock()
 				}
 			}
 
@@ -240,9 +246,11 @@ func (t *FlashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) {
 					}
 					fg.Slots = append(fg.Slots, fg.ReservedSlots...)
 					fg.ReservedSlots = make([]uint32, 0)
+					fg.ReduceAllTime = 0
 					if err := t.SyncFlashGroupFunc(fg); err != nil {
 						fg.Slots = oldSlots
 						fg.ReservedSlots = oldReservedSlots
+						fg.ReduceAllTime = 0
 					}
 					fg.lock.Unlock()
 				} else {