From bdf1f207bddd853a6357aa6311c07eee8aeb85df Mon Sep 17 00:00:00 2001 From: clinx Date: Tue, 19 Aug 2025 14:55:06 +0800 Subject: [PATCH] fix(flash): store fg slots when slots changed with: #1000292546 Signed-off-by: clinx --- master/cluster.go | 1 + master/cluster_test.go | 1 + master/master_manager.go | 1 + remotecache/flashgroupmanager/cluster.go | 3 +- remotecache/flashgroupmanager/flash_group.go | 21 +++++-- .../flashgroupmanager/flash_group_test.go | 16 ++++-- .../flashgroupmanager/flash_node_topology.go | 57 +++++++++++-------- .../flashgroupmanager/master_manager.go | 1 + sdk/remotecache/client.go | 2 +- 9 files changed, 65 insertions(+), 38 deletions(-) diff --git a/master/cluster.go b/master/cluster.go index b548835c8..ee6b7f7e6 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -482,6 +482,7 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition c.AutoDecommissionInterval.Store(int64(defaultAutoDecommissionDiskInterval)) c.server = server c.flashNodeTopo = flashgroupmanager.NewFlashNodeTopology() + c.flashNodeTopo.SyncFlashGroupFunc = c.syncUpdateFlashGroup c.cleanTask = make(map[string]*CleanTask) c.PlanRun = false c.flashManMgr = newFlashManualTaskManager(c) diff --git a/master/cluster_test.go b/master/cluster_test.go index 279529397..af98ac7f0 100644 --- a/master/cluster_test.go +++ b/master/cluster_test.go @@ -375,6 +375,7 @@ func TestMasterClientLeaderChange(t *testing.T) { flashNodeTopo: flashgroupmanager.NewFlashNodeTopology(), leaderInfo: server.leaderInfo, } + cluster.flashNodeTopo.SyncFlashGroupFunc = cluster.syncUpdateFlashGroup server.cluster = cluster cluster.t = newTopology() diff --git a/master/master_manager.go b/master/master_manager.go index 431e66406..549085671 100644 --- a/master/master_manager.go +++ b/master/master_manager.go @@ -339,6 +339,7 @@ func (m *Server) clearMetadata() { m.cluster.flashNodeTopo.Clear() m.cluster.flashNodeTopo = flashgroupmanager.NewFlashNodeTopology() + m.cluster.flashNodeTopo.SyncFlashGroupFunc = m.cluster.syncUpdateFlashGroup } func (m *Server) refreshUser() (err error) { diff --git a/remotecache/flashgroupmanager/cluster.go b/remotecache/flashgroupmanager/cluster.go index a6a656ab1..edfc6279a 100644 --- a/remotecache/flashgroupmanager/cluster.go +++ b/remotecache/flashgroupmanager/cluster.go @@ -33,6 +33,7 @@ func newCluster(name string, cfg *clusterConfig, leaderInfo *LeaderInfo, fsm *Me c = new(Cluster) c.Name = name c.flashNodeTopo = NewFlashNodeTopology() + c.flashNodeTopo.SyncFlashGroupFunc = c.syncUpdateFlashGroup c.stopc = make(chan bool) c.fsm = fsm c.partition = partition @@ -89,8 +90,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 9470f79b5..8dca6f516 100644 --- a/remotecache/flashgroupmanager/flash_group.go +++ b/remotecache/flashgroupmanager/flash_group.go @@ -14,7 +14,7 @@ import ( const ( UnusedFlashNodeFlashGroupID = 0 DefaultWaitClientUpdateFgTimeSec = 65 - WaitForRecoverCount = 10 + WaitForRecoverCount = 20 ) type FlashGroupValue struct { @@ -65,6 +65,7 @@ func newFlashGroup(id uint64, slots []uint32, slotStatus proto.SlotStatus, pendi fg.Slots = slots fg.SlotStatus = slotStatus fg.PendingSlots = pendingSlots + fg.ReservedSlots = make([]uint32, 0) fg.Step = step fg.Weight = weight fg.Status = status @@ -91,7 +92,7 @@ func (fg *FlashGroup) IsLostAllFlashNode() bool { return atomic.LoadInt32(&fg.LostAllFlashNode) != 0 } -func (fg *FlashGroup) ReduceSlot() { +func (fg *FlashGroup) ReduceSlot(syncFlashGroupFunc SyncUpdateFlashGroupFunc) { if atomic.CompareAndSwapInt32(&fg.ReducingSlots, 0, 1) { return } @@ -115,13 +116,12 @@ func (fg *FlashGroup) ReduceSlot() { if i <= WaitForRecoverCount { continue } - fg.executeReduceSlot(numToSelect) - atomic.StoreInt32(&fg.SlotChanged, 1) + fg.executeReduceSlot(numToSelect, syncFlashGroupFunc) } }() } -func (fg *FlashGroup) executeReduceSlot(numToReduce int) { +func (fg *FlashGroup) executeReduceSlot(numToReduce int, syncFlashGroupFunc SyncUpdateFlashGroupFunc) { if len(fg.Slots) == 0 { return } @@ -132,8 +132,18 @@ func (fg *FlashGroup) executeReduceSlot(numToReduce int) { if numToReduce == 0 { return } + var oldSlots, oldReservedSlots []uint32 + oldSlots = append(oldSlots, fg.Slots...) + oldReservedSlots = append(oldReservedSlots, fg.ReservedSlots...) + if log.EnableInfo() { + log.LogInfof("executeReduceSlot fg(%v) oldSlots(%v) oldReservedSlots(%v)", fg.ID, oldSlots, oldReservedSlots) + } fg.ReservedSlots = append(fg.ReservedSlots, fg.Slots[:numToReduce]...) fg.Slots = fg.Slots[numToReduce:] + if err := syncFlashGroupFunc(fg); err != nil { + fg.Slots = oldSlots + fg.ReservedSlots = oldReservedSlots + } } func (fg *FlashGroup) GetStatus() (st proto.FlashGroupStatus) { @@ -219,6 +229,7 @@ func NewFlashGroupFromFgv(fgv FlashGroupValue) *FlashGroup { fg.Slots = fgv.Slots fg.SlotStatus = fgv.SlotStatus fg.PendingSlots = fgv.PendingSlots + fg.ReservedSlots = fgv.ReservedSlots fg.Step = fgv.Step fg.Weight = fgv.Weight fg.Status = fgv.Status diff --git a/remotecache/flashgroupmanager/flash_group_test.go b/remotecache/flashgroupmanager/flash_group_test.go index 9055e86ec..35c8c1ce7 100644 --- a/remotecache/flashgroupmanager/flash_group_test.go +++ b/remotecache/flashgroupmanager/flash_group_test.go @@ -11,6 +11,10 @@ import ( "github.com/stretchr/testify/assert" ) +func tSyncUpdateFlashGroup(flashGroup *FlashGroup) (err error) { + return nil +} + // TestNewFlashGroup tests the creation of a new FlashGroup instance func TestNewFlashGroup(t *testing.T) { id := uint64(123) @@ -64,7 +68,7 @@ func TestFlashGroup_ReduceSlot_AlreadyReducing(t *testing.T) { atomic.StoreInt32(&fg.ReducingSlots, 1) // This should return immediately without starting a new goroutine - fg.ReduceSlot() + fg.ReduceSlot(tSyncUpdateFlashGroup) // Verify the flag is still set assert.Equal(t, int32(1), atomic.LoadInt32(&fg.ReducingSlots)) @@ -78,7 +82,7 @@ func TestFlashGroup_ReduceSlot_NoSlots(t *testing.T) { atomic.StoreInt32(&fg.LostAllFlashNode, 1) // Start reduction - fg.ReduceSlot() + fg.ReduceSlot(tSyncUpdateFlashGroup) // Wait a bit for the goroutine to process time.Sleep(100 * time.Millisecond) @@ -98,7 +102,7 @@ func TestFlashGroup_ReduceSlot_WithSlots(t *testing.T) { atomic.StoreInt32(&fg.LostAllFlashNode, 1) // Start reduction - fg.ReduceSlot() + fg.ReduceSlot(tSyncUpdateFlashGroup) // Wait for the first reduction cycle (20 seconds, but we'll wait less for testing) // In a real scenario, this would take 20 seconds @@ -122,7 +126,7 @@ func TestFlashGroup_executeReduceSlot(t *testing.T) { initialSlotCount := len(fg.Slots) // Execute reduction - fg.executeReduceSlot((len(fg.Slots) + 4 - 1) / 4) + fg.executeReduceSlot((len(fg.Slots)+4-1)/4, tSyncUpdateFlashGroup) // Verify that approximately 25% of slots were moved (8 slots -> 6 slots, 2 to reserved) expectedRemaining := (initialSlotCount * 3) / 4 // 75% remaining @@ -140,7 +144,7 @@ func TestFlashGroup_executeReduceSlot_EmptySlots(t *testing.T) { fg := newFlashGroup(1, []uint32{}, proto.SlotStatus_Completed, []uint32{}, 0, proto.FlashGroupStatus_Active, 100) // This should not panic or cause issues - fg.executeReduceSlot((len(fg.Slots) + 4 - 1) / 4) + fg.executeReduceSlot((len(fg.Slots)+4-1)/4, tSyncUpdateFlashGroup) assert.Equal(t, 0, len(fg.Slots)) assert.Equal(t, 0, len(fg.ReservedSlots)) @@ -150,7 +154,7 @@ func TestFlashGroup_executeReduceSlot_EmptySlots(t *testing.T) { func TestFlashGroup_executeReduceSlot_SingleSlot(t *testing.T) { fg := newFlashGroup(1, []uint32{1}, proto.SlotStatus_Completed, []uint32{}, 1, proto.FlashGroupStatus_Active, 100) - fg.executeReduceSlot((len(fg.Slots) + 4 - 1) / 4) + fg.executeReduceSlot((len(fg.Slots)+4-1)/4, tSyncUpdateFlashGroup) // With 1 slot, 25% rounded up is 1, so all slots should be moved to reserved assert.Equal(t, 0, len(fg.Slots)) diff --git a/remotecache/flashgroupmanager/flash_node_topology.go b/remotecache/flashgroupmanager/flash_node_topology.go index 0b059e29f..298fe7881 100644 --- a/remotecache/flashgroupmanager/flash_node_topology.go +++ b/remotecache/flashgroupmanager/flash_node_topology.go @@ -78,10 +78,11 @@ type FlashNodeTopology struct { flashNodeMap sync.Map // key: FlashNodeAddr, value: *FlashNode zoneMap sync.Map // key: zoneName, value: *FlashNodeZone - clientEmpty []byte // empty response cache - clientOff atomic.Value // []byte, default nil (on) - clientCache atomic.Value // []byte - clientUpdateCh chan struct{} // update client response cache + clientEmpty []byte // empty response cache + clientOff atomic.Value // []byte, default nil (on) + clientCache atomic.Value // []byte + clientUpdateCh chan struct{} // update client response cache + SyncFlashGroupFunc SyncUpdateFlashGroupFunc } func NewFlashNodeTopology() (t *FlashNodeTopology) { @@ -201,30 +202,49 @@ func (t *FlashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) { t.flashGroupMap.Range(func(_, value interface{}) bool { fg := value.(*FlashGroup) if fg.GetStatus().IsActive() { + var oldSlots, oldReservedSlots []uint32 hosts := fg.getFlashNodeHostsEnableAndActive() if len(hosts) == 0 { if log.EnableInfo() { log.LogInfof("fg(%v) lost all flashnodes", fg) } atomic.StoreInt32(&fg.LostAllFlashNode, 1) - fg.ReduceSlot() - } else if len(fg.ReservedSlots) > 0 { - fg.lock.Lock() - fg.Slots = append(fg.Slots, fg.ReservedSlots...) - fg.ReservedSlots = make([]uint32, 0) - fg.lock.Unlock() + fg.ReduceSlot(t.SyncFlashGroupFunc) + } else { atomic.StoreInt32(&fg.LostAllFlashNode, 0) - atomic.StoreInt32(&fg.SlotChanged, 1) + 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) + } + 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() + } } if len(fg.Slots) == 0 { disableFlashGroupNum++ if disableFlashGroupNum >= maxDisableFlashGroupCount && 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) + } 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() - atomic.StoreInt32(&fg.SlotChanged, 1) } else { return true } @@ -234,6 +254,7 @@ func (t *FlashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) { Slot: fg.Slots, Hosts: hosts, }) + } return true }) @@ -781,18 +802,6 @@ func (t *FlashNodeTopology) GetAllFlashNodes() (flashNodes []proto.NodeView) { return } -func (t *FlashNodeTopology) UpdateFlashGroup(syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc) { - t.flashGroupMap.Range(func(_, value interface{}) bool { - fg := value.(*FlashGroup) - slotChanged := atomic.LoadInt32(&fg.SlotChanged) != 0 - if slotChanged { - syncUpdateFlashGroupFunc(fg) - atomic.StoreInt32(&fg.SlotChanged, 0) - } - return true - }) -} - func (t *FlashNodeTopology) ClientUpdateChannel() <-chan struct{} { return t.clientUpdateCh } diff --git a/remotecache/flashgroupmanager/master_manager.go b/remotecache/flashgroupmanager/master_manager.go index 392910231..bde866610 100644 --- a/remotecache/flashgroupmanager/master_manager.go +++ b/remotecache/flashgroupmanager/master_manager.go @@ -47,6 +47,7 @@ func (m *FlashGroupManager) handleLeaderChange(leader uint64) { func (m *FlashGroupManager) clearMetadata() { m.cluster.flashNodeTopo.Clear() m.cluster.flashNodeTopo = NewFlashNodeTopology() + m.cluster.flashNodeTopo.SyncFlashGroupFunc = m.cluster.syncUpdateFlashGroup } func (m *FlashGroupManager) loadMetadata() { diff --git a/sdk/remotecache/client.go b/sdk/remotecache/client.go index d77e2ee9f..b6aece4ac 100755 --- a/sdk/remotecache/client.go +++ b/sdk/remotecache/client.go @@ -594,7 +594,7 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re return } replyPacket := NewFlashCacheReply() - if err = replyPacket.ReadFromConnExt(conn, int(rc.firstPacketTimeout)); err != nil { + if err = replyPacket.ReadFromConn(conn, proto.ReadDeadlineTime); err != nil { log.LogWarnf("FlashGroup put: reqId(%v) failed to ReadFromConn, replyPacket(%v), fg host(%v) , err(%v)", reqId, replyPacket, addr, err) return }