gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(proxy): handle current/backup swap during retain

Browse Source 
with: #1000910244

Signed-off-by: xiejian <xiejian3@oppo.com>
This commit is contained in:
xiejian 2026-04-23 15:27:10 +08:00 committed by slasher
parent a763740261
commit b09bd2abcb
3 changed files with 65 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
blobstore/proxy/allocator/retain_volume_task.go
View File

@ -81,7 +81,7 @@ func (v *volumeMgr) retain(ctx context.Context, isBackup bool) {
if end > numTokens {
end = numTokens
}
v.retainVolumeFromCm(ctx, retainTokenArgs[start:end], isBackup)
v.retainVolumeFromCm(ctx, retainTokenArgs[start:end])
start = end
if start < numTokens {
time.Sleep(time.Duration(v.RetainBatchIntervalS) * time.Second)
@ -117,12 +117,12 @@ func (v *volumeMgr) handleFullVols(ctx context.Context, isBackup bool) {
// remove retain failed volume and update success expire time
func (v *volumeMgr) handleRetainResult(ctx context.Context, retainTokenArgs []string,
retainRet []clustermgr.RetainVolume, isBackup bool,
retainRet []clustermgr.RetainVolume,
) {
span := trace.SpanFromContextSafe(ctx)
if len(retainRet) == 0 {
for _, token := range retainTokenArgs {
err := v.discardVolume(ctx, token, isBackup)
err := v.discardVolume(ctx, token)
if err != nil {
span.Error(err)
}
@ -136,7 +136,7 @@ func (v *volumeMgr) handleRetainResult(ctx context.Context, retainTokenArgs []st
span.Errorf("decodeToken %v error", infos.Token)
continue
}
err = v.updateExpireTime(vid, infos.ExpireTime, isBackup)
err = v.updateExpireTime(vid, infos.ExpireTime)
if err != nil {
span.Error(err, vid)
}
@ -146,14 +146,14 @@ func (v *volumeMgr) handleRetainResult(ctx context.Context, retainTokenArgs []st
if _, ok := tokenMap[token]; ok {
continue
}
err := v.discardVolume(ctx, token, isBackup)
err := v.discardVolume(ctx, token)
if err != nil {
span.Error(err, token)
}
}
}
func (v *volumeMgr) discardVolume(ctx context.Context, token string, isBackup bool) (err error) {
func (v *volumeMgr) discardVolume(ctx context.Context, token string) (err error) {
span := trace.SpanFromContextSafe(ctx)
_, vid, err := proto.DecodeToken(token)
if err != nil {
@ -161,7 +161,7 @@ func (v *volumeMgr) discardVolume(ctx context.Context, token string, isBackup bo
}
span.Infof("retain failed vid: %v", vid)
for _, info := range v.modeInfos {
if vol, ok := info.Get(vid, isBackup); ok {
if vol, ok := info.GetAny(vid); ok {
vol.mu.Lock()
vol.deleted = true
vol.mu.Unlock()
@ -190,9 +190,9 @@ func (v *volumeMgr) genRetainVolume(ctx context.Context, isBackup bool) (tokens
return
}
func (v *volumeMgr) updateExpireTime(vid proto.Vid, expireTime int64, isBackup bool) (err error) {
func (v *volumeMgr) updateExpireTime(vid proto.Vid, expireTime int64) (err error) {
for _, modeInfo := range v.modeInfos {
if vol, ok := modeInfo.Get(vid, isBackup); ok {
if vol, ok := modeInfo.GetAny(vid); ok {
vol.ExpireTime = expireTime
return nil
}
@ -200,7 +200,7 @@ func (v *volumeMgr) updateExpireTime(vid proto.Vid, expireTime int64, isBackup b
return errors.New("vid does not exist ")
}
func (v *volumeMgr) retainVolumeFromCm(ctx context.Context, tokens []string, isBackup bool) {
func (v *volumeMgr) retainVolumeFromCm(ctx context.Context, tokens []string) {
span := trace.SpanFromContextSafe(ctx)
args := &clustermgr.RetainVolumeArgs{
Tokens: tokens,
@ -211,7 +211,7 @@ func (v *volumeMgr) retainVolumeFromCm(ctx context.Context, tokens []string, isB
return
}
span.Debugf("retain result: %#v, lens: %v\n", retainVolume, len(retainVolume.RetainVolTokens))
v.handleRetainResult(ctx, tokens, retainVolume.RetainVolTokens, isBackup)
v.handleRetainResult(ctx, tokens, retainVolume.RetainVolTokens)
}
// checkAndReplenish inspects the volume cache for every code mode after a retain cycle.

43
blobstore/proxy/allocator/volume_test.go
View File

@ -89,6 +89,49 @@ func TestVolumes_Put(t *testing.T) {
require.False(t, ok)
}
func TestModeInfo_GetAny(t *testing.T) {
newVol := func(vid proto.Vid) *volume {
return &volume{AllocVolumeInfo: cm.AllocVolumeInfo{
VolumeInfo: cm.VolumeInfo{
VolumeInfoBase: cm.VolumeInfoBase{
Free: 10 * 16 * 1024 * 1024 * 1024,
Vid: vid,
},
},
ExpireTime: 100,
}}
}
m := &modeInfo{current: &volumes{}, backup: &volumes{}}
m.Put(newVol(1), false)
m.Put(newVol(2), true)
vol, ok := m.GetAny(proto.Vid(1))
require.True(t, ok)
require.Equal(t, proto.Vid(1), vol.Vid)
vol, ok = m.GetAny(proto.Vid(2))
require.True(t, ok)
require.Equal(t, proto.Vid(2), vol.Vid)
_, ok = m.GetAny(proto.Vid(99))
require.False(t, ok)
// simulate current/backup swap during a retain RPC: the vol that used to
// be in current is now in backup; GetAny should still find it.
m.lock.Lock()
m.current, m.backup = m.backup, m.current
m.lock.Unlock()
vol, ok = m.GetAny(proto.Vid(1))
require.True(t, ok)
require.Equal(t, proto.Vid(1), vol.Vid)
vol, ok = m.GetAny(proto.Vid(2))
require.True(t, ok)
require.Equal(t, proto.Vid(2), vol.Vid)
}
func TestVolumes_Delete(t *testing.T) {
v := volumes{}
for _, i := range []int{1, 3, 2, 4} {

11
blobstore/proxy/allocator/volumemgr.go
View File

@ -141,6 +141,17 @@ func (m *modeInfo) Get(vid proto.Vid, isBackup bool) (res *volume, ok bool) {
return
}
func (m *modeInfo) GetAny(vid proto.Vid) (res *volume, ok bool) {
m.lock.RLock()
if res, ok = m.current.Get(vid); ok {
m.lock.RUnlock()
return
}
res, ok = m.backup.Get(vid)
m.lock.RUnlock()
return
}
func (m *modeInfo) TotalFree() int64 {
m.lock.RLock()
totalFree := m.backup.TotalFree() + m.current.TotalFree()