mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
chore(blobnode): adjust code of compact process , structural adjustment
with: #1000223631 Signed-off-by: mawei029 <mawei2@oppo.com>
This commit is contained in:
mawei029
slasher
parent
2de4f09154
commit
80d64c053b
@ -123,6 +123,91 @@ ErrCompact:
|
||||
return ncs, nil
|
||||
}
|
||||
|
||||
func (cs *chunk) CommitCompact(ctx context.Context, ncs core.ChunkAPI) (err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
newcs := ncs.(*chunk)
|
||||
|
||||
// replStg -> normal background stg
|
||||
repStg, backgroundStg := cs.getStg(), newcs.getStg()
|
||||
|
||||
span.Infof("compact commit. Change <%s> to <%s>", repStg.ID(), backgroundStg.ID())
|
||||
|
||||
cs.lock.Lock()
|
||||
{
|
||||
// exchange handler
|
||||
cs.setStg(backgroundStg)
|
||||
newcs.setStg(repStg)
|
||||
}
|
||||
cs.lock.Unlock()
|
||||
|
||||
cs.lastCompactTime = time.Now().UnixNano()
|
||||
|
||||
// finally init stats
|
||||
if err = cs.refreshFstat(ctx); err != nil {
|
||||
span.Errorf("Failed refresh fstat. <%s>", cs.ID())
|
||||
return err
|
||||
}
|
||||
|
||||
// wait for all requests before switching handles to complete
|
||||
timestamp := cs.consistent.Synchronize()
|
||||
span.Infof("All requests before the switch handle are completed, timestamp:%v", timestamp)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cs *chunk) StopCompact(ctx context.Context, ncs core.ChunkAPI) (err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
cs.lock.Lock()
|
||||
defer cs.lock.Unlock()
|
||||
|
||||
cs.compacting = false
|
||||
cs.resetCompactTask()
|
||||
|
||||
// note: In an abnormal situation, compact fails, but stg is double-written stg
|
||||
curStg := cs.getStg()
|
||||
rawStg := curStg.RawStorage()
|
||||
if rawStg != nil {
|
||||
// If a commit is made, then it will definitely not enter this branch;
|
||||
span.Warnf("cur:%s/raw:%s compact fails, stg is double-written stg", curStg.ID(), rawStg.ID())
|
||||
// restore stg
|
||||
cs.setStg(rawStg)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cs *chunk) NeedCompact(ctx context.Context) bool {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
stg := cs.getStg()
|
||||
|
||||
stat, err := stg.Stat(context.TODO())
|
||||
if err != nil {
|
||||
span.Errorf("get chunk data space info failed: %v", err)
|
||||
return false
|
||||
}
|
||||
|
||||
size, phySize := stat.FileSize, stat.PhySize
|
||||
|
||||
// file size is too large
|
||||
if size >= cs.conf.CompactTriggerThreshold {
|
||||
span.Debugf("phySize:%v/fsize:%v, threshold:%v",
|
||||
phySize, size, cs.conf.CompactTriggerThreshold)
|
||||
return true
|
||||
}
|
||||
|
||||
// void rate exceeds threshold
|
||||
if size > cs.conf.CompactMinSizeThreshold && (1-float64(phySize)/float64(size)) >= cs.conf.CompactEmptyRateThreshold {
|
||||
span.Debugf("phySize:%v/fsize:%v, minSize:%v threshold:%v",
|
||||
phySize, size, cs.conf.CompactMinSizeThreshold, cs.conf.CompactEmptyRateThreshold)
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (cs *chunk) handleErrCompact(ctx context.Context, ncs core.ChunkAPI) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
@ -257,91 +342,6 @@ COMPACT:
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cs *chunk) CommitCompact(ctx context.Context, ncs core.ChunkAPI) (err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
newcs := ncs.(*chunk)
|
||||
|
||||
// replStg -> normal background stg
|
||||
repStg, backgroundStg := cs.getStg(), newcs.getStg()
|
||||
|
||||
span.Infof("compact commit. Change <%s> to <%s>", repStg.ID(), backgroundStg.ID())
|
||||
|
||||
cs.lock.Lock()
|
||||
{
|
||||
// exchange handler
|
||||
cs.setStg(backgroundStg)
|
||||
newcs.setStg(repStg)
|
||||
}
|
||||
cs.lock.Unlock()
|
||||
|
||||
cs.lastCompactTime = time.Now().UnixNano()
|
||||
|
||||
// finally init stats
|
||||
if err = cs.refreshFstat(ctx); err != nil {
|
||||
span.Errorf("Failed refresh fstat. <%s>", cs.ID())
|
||||
return err
|
||||
}
|
||||
|
||||
// wait for all requests before switching handles to complete
|
||||
timestamp := cs.consistent.Synchronize()
|
||||
span.Infof("All requests before the switch handle are completed, timestamp:%v", timestamp)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cs *chunk) StopCompact(ctx context.Context, ncs core.ChunkAPI) (err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
cs.lock.Lock()
|
||||
defer cs.lock.Unlock()
|
||||
|
||||
cs.compacting = false
|
||||
cs.resetCompactTask()
|
||||
|
||||
// note: In an abnormal situation, compact fails, but stg is double-written stg
|
||||
curStg := cs.getStg()
|
||||
rawStg := curStg.RawStorage()
|
||||
if rawStg != nil {
|
||||
// If a commit is made, then it will definitely not enter this branch;
|
||||
span.Warnf("cur:%s/raw:%s compact fails, stg is double-written stg", curStg.ID(), rawStg.ID())
|
||||
// restore stg
|
||||
cs.setStg(rawStg)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cs *chunk) NeedCompact(ctx context.Context) bool {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
stg := cs.getStg()
|
||||
|
||||
stat, err := stg.Stat(context.TODO())
|
||||
if err != nil {
|
||||
span.Errorf("get chunk data space info failed: %v", err)
|
||||
return false
|
||||
}
|
||||
|
||||
size, phySize := stat.FileSize, stat.PhySize
|
||||
|
||||
// file size is too large
|
||||
if size >= cs.conf.CompactTriggerThreshold {
|
||||
span.Debugf("phySize:%v/fsize:%v, threshold:%v",
|
||||
phySize, size, cs.conf.CompactTriggerThreshold)
|
||||
return true
|
||||
}
|
||||
|
||||
// void rate exceeds threshold
|
||||
if size > cs.conf.CompactMinSizeThreshold && (1-float64(phySize)/float64(size)) >= cs.conf.CompactEmptyRateThreshold {
|
||||
span.Debugf("phySize:%v/fsize:%v, minSize:%v threshold:%v",
|
||||
phySize, size, cs.conf.CompactMinSizeThreshold, cs.conf.CompactEmptyRateThreshold)
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (cs *chunk) compactCheck(ctx context.Context, ncs *chunk) (err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
|
||||
@ -31,69 +31,6 @@ const (
|
||||
setCompactInterval = 100 * time.Millisecond
|
||||
)
|
||||
|
||||
func (ds *DiskStorage) loopCompactFile() {
|
||||
span, _ := trace.StartSpanFromContextWithTraceID(context.Background(), "", "Compact"+ds.Conf.Path)
|
||||
|
||||
timer := initTimer(ds.Conf.ChunkCompactIntervalSec)
|
||||
defer timer.Stop()
|
||||
|
||||
span.Infof("start compact executor.")
|
||||
|
||||
// consumer
|
||||
ds.loopAttach(func() {
|
||||
for {
|
||||
select {
|
||||
case <-ds.closeCh:
|
||||
span.Warnf("loopCompact done...")
|
||||
return
|
||||
case vuid := <-ds.compactCh:
|
||||
span.Debugf("recv compact message. vuid:[%d]", vuid)
|
||||
if err := ds.ExecCompactChunk(vuid); err != nil {
|
||||
span.Errorf("compact vuid: %d err:%v", vuid, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
span.Infof("start compact checker.")
|
||||
|
||||
// producer
|
||||
for {
|
||||
select {
|
||||
case <-ds.closeCh:
|
||||
span.Warnf("loopCompact done...")
|
||||
return
|
||||
case <-timer.C:
|
||||
ds.runCompactFiles()
|
||||
resetTimer(ds.Conf.ChunkCompactIntervalSec, timer)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) runCompactFiles() {
|
||||
span, ctx := trace.StartSpanFromContextWithTraceID(context.Background(), "", base.BackgroudReqID("Compact"+ds.Conf.Path))
|
||||
|
||||
span.Debugf("find chunks that meet the conditions ===")
|
||||
defer span.Debugf("check compact files done. ===")
|
||||
|
||||
chunks := make([]core.ChunkAPI, 0)
|
||||
ds.Lock.RLock()
|
||||
for _, chunk := range ds.Chunks {
|
||||
chunks = append(chunks, chunk)
|
||||
}
|
||||
ds.Lock.RUnlock()
|
||||
|
||||
for _, chunk := range chunks {
|
||||
if !chunk.NeedCompact(ctx) {
|
||||
continue
|
||||
}
|
||||
span.Infof("will compact vuid:<%d>", chunk.Vuid())
|
||||
ds.EnqueueCompact(ctx, chunk.Vuid())
|
||||
// Once in a round
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) EnqueueCompact(ctx context.Context, vuid proto.Vuid) {
|
||||
ds.compactCh <- vuid
|
||||
}
|
||||
@ -167,34 +104,6 @@ STOPCOMPACT:
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) destroyRedundant(ctx context.Context, ncs core.ChunkAPI) (err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
// keep new chunkstorage meta
|
||||
ncsMeta := ncs.VuidMeta()
|
||||
|
||||
// wait old stg all request done；
|
||||
for {
|
||||
time.Sleep(10 * time.Second)
|
||||
if !ncs.HasPendingRequest() {
|
||||
break
|
||||
}
|
||||
span.Debugf("=== wait chunk(%s) all request done ===", ncsMeta.ChunkID)
|
||||
}
|
||||
|
||||
span.Infof("safe here. id:%s request all done.", ncsMeta.ChunkID)
|
||||
|
||||
// isolated. safe
|
||||
ncs.Close(ctx)
|
||||
|
||||
// update chunk status, mark destroy. destroy async
|
||||
ncsMeta.Status = cmapi.ChunkStatusRelease
|
||||
ncsMeta.Reason = cmapi.ReleaseForCompact
|
||||
ncsMeta.Mtime = time.Now().UnixNano()
|
||||
|
||||
return ds.SuperBlock.UpsertChunk(ctx, ncsMeta.ChunkID, *ncsMeta)
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) ExecCompactChunk(vuid proto.Vuid) (err error) {
|
||||
span, ctx := trace.StartSpanFromContextWithTraceID(context.Background(), "", base.BackgroudReqID("Compact"+ds.Conf.Path))
|
||||
|
||||
@ -235,6 +144,97 @@ func (ds *DiskStorage) ExecCompactChunk(vuid proto.Vuid) (err error) {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) loopCompactFile() {
|
||||
span, _ := trace.StartSpanFromContextWithTraceID(context.Background(), "", "Compact"+ds.Conf.Path)
|
||||
|
||||
timer := initTimer(ds.Conf.ChunkCompactIntervalSec)
|
||||
defer timer.Stop()
|
||||
|
||||
span.Infof("start compact executor.")
|
||||
|
||||
// consumer
|
||||
ds.loopAttach(func() {
|
||||
for {
|
||||
select {
|
||||
case <-ds.closeCh:
|
||||
span.Warnf("loopCompact done...")
|
||||
return
|
||||
case vuid := <-ds.compactCh:
|
||||
span.Debugf("recv compact message. vuid:[%d]", vuid)
|
||||
if err := ds.ExecCompactChunk(vuid); err != nil {
|
||||
span.Errorf("compact vuid: %d err:%v", vuid, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
span.Infof("start compact checker.")
|
||||
|
||||
// producer
|
||||
for {
|
||||
select {
|
||||
case <-ds.closeCh:
|
||||
span.Warnf("loopCompact done...")
|
||||
return
|
||||
case <-timer.C:
|
||||
ds.runCompactFiles()
|
||||
resetTimer(ds.Conf.ChunkCompactIntervalSec, timer)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) runCompactFiles() {
|
||||
span, ctx := trace.StartSpanFromContextWithTraceID(context.Background(), "", base.BackgroudReqID("Compact"+ds.Conf.Path))
|
||||
|
||||
span.Debugf("find chunks that meet the conditions ===")
|
||||
defer span.Debugf("check compact files done. ===")
|
||||
|
||||
chunks := make([]core.ChunkAPI, 0)
|
||||
ds.Lock.RLock()
|
||||
for _, chunk := range ds.Chunks {
|
||||
chunks = append(chunks, chunk)
|
||||
}
|
||||
ds.Lock.RUnlock()
|
||||
|
||||
for _, chunk := range chunks {
|
||||
if !chunk.NeedCompact(ctx) {
|
||||
continue
|
||||
}
|
||||
span.Infof("will compact vuid:<%d>", chunk.Vuid())
|
||||
ds.EnqueueCompact(ctx, chunk.Vuid())
|
||||
// Once in a round
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) destroyRedundant(ctx context.Context, ncs core.ChunkAPI) (err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
|
||||
// keep new chunkstorage meta
|
||||
ncsMeta := ncs.VuidMeta()
|
||||
|
||||
// wait old stg all request done；
|
||||
for {
|
||||
time.Sleep(10 * time.Second)
|
||||
if !ncs.HasPendingRequest() {
|
||||
break
|
||||
}
|
||||
span.Debugf("=== wait chunk(%s) all request done ===", ncsMeta.ChunkID)
|
||||
}
|
||||
|
||||
span.Infof("safe here. id:%s request all done.", ncsMeta.ChunkID)
|
||||
|
||||
// isolated. safe
|
||||
ncs.Close(ctx)
|
||||
|
||||
// update chunk status, mark destroy. destroy async
|
||||
ncsMeta.Status = cmapi.ChunkStatusRelease
|
||||
ncsMeta.Reason = cmapi.ReleaseForCompact
|
||||
ncsMeta.Mtime = time.Now().UnixNano()
|
||||
|
||||
return ds.SuperBlock.UpsertChunk(ctx, ncsMeta.ChunkID, *ncsMeta)
|
||||
}
|
||||
|
||||
func (ds *DiskStorage) notifyCompacting(ctx context.Context, vuid proto.Vuid, compacting bool) (
|
||||
err error,
|
||||
) {
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -106,6 +106,8 @@ func addCmdDiskDrop(diskCommand *grumble.Command) {
|
||||
}
|
||||
|
||||
func dropStatCheck(c *grumble.Context) error {
|
||||
log.SetOutputLevel(log.Level(c.Flags.Uint("log_level")))
|
||||
|
||||
diskInfos, err := checkDiskDropConf(c)
|
||||
if err != nil {
|
||||
return err
|
||||
@ -117,7 +119,6 @@ func dropStatCheck(c *grumble.Context) error {
|
||||
cmCli := newCmClient(c)
|
||||
fmt.Printf("start time: " + time.Now().Format("2006-01-02 15:04:05") + "\n")
|
||||
printDiskID(diskInfos)
|
||||
log.SetOutputLevel(log.Level(c.Flags.Uint("log_level")))
|
||||
|
||||
// check cm
|
||||
if vuidCmCnt, err = getVuidFromCm(ctx, cmCli, diskInfos); err != nil {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
