From 82d52a297d95b3c6a39e058210eb01e8c5cd0fef Mon Sep 17 00:00:00 2001 From: clinx Date: Tue, 19 Aug 2025 16:34:17 +0800 Subject: [PATCH] fix(master): long-term unused slot activation employs a gradual addition approach. with: #1000292905 Signed-off-by: clinx (cherry picked from commit bdf08f93a5f2fb4a1a98c5d879facc5797229f3d) --- master/flash_node_topology.go | 2 - remotecache/flashgroupmanager/flash_group.go | 66 ++++- .../flashgroupmanager/flash_group_test.go | 258 ++++++++++++++++++ .../flashgroupmanager/flash_node_topology.go | 32 ++- 4 files changed, 342 insertions(+), 16 deletions(-) diff --git a/master/flash_node_topology.go b/master/flash_node_topology.go index ec02d9045..356e47bb2 100644 --- a/master/flash_node_topology.go +++ b/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: diff --git a/remotecache/flashgroupmanager/flash_group.go b/remotecache/flashgroupmanager/flash_group.go index 8dca6f516..4b52ec75d 100644 --- a/remotecache/flashgroupmanager/flash_group.go +++ b/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 + } } } diff --git a/remotecache/flashgroupmanager/flash_group_test.go b/remotecache/flashgroupmanager/flash_group_test.go index 35c8c1ce7..02f87e58c 100644 --- a/remotecache/flashgroupmanager/flash_group_test.go +++ b/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) + }) + } +} diff --git a/remotecache/flashgroupmanager/flash_node_topology.go b/remotecache/flashgroupmanager/flash_node_topology.go index 298fe7881..727905169 100644 --- a/remotecache/flashgroupmanager/flash_node_topology.go +++ b/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() - 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) + if len(fg.Slots) == 0 && fg.ReduceAllTime != 0 && (time.Now().Unix()-fg.ReduceAllTime >= twoDaysInSeconds) { + fg.IncreaseSlot(t.SyncFlashGroupFunc) + } else if atomic.LoadInt32(&fg.IncreasingSlots) == 0 { + fg.lock.Lock() + 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 {