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(flashnode): do not close channel twice

Browse Source 
close:#1000016077

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-03-24 14:32:08 +08:00 committed by zhumingze1108
parent 2b80a78b76
commit baa7412321
5 changed files with 41 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
flashnode/cachengine/block.go
View File

@ -563,7 +563,7 @@ func (cb *CacheBlock) GetRootPath() string {
func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto.DataSource, done chan struct{}) {
cb.initOnce.Do(func() {
cb.InitForCacheRead(sources, engine.readDataNodeTimeout, done)
cb.InitForCacheRead(sources, engine.readDataNodeTimeout)
select {
case <-cb.closeCh:
engine.deleteCacheBlock(cb.blockKey)
@ -571,9 +571,10 @@ func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto
default:
}
})
close(done)
}
func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNodeTimeout int, done chan struct{}) {
func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNodeTimeout int) {
var err error
var file *os.File
bgTime := stat.BeginStat()
@ -592,7 +593,6 @@ func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNode
}
stat.EndStat("MissCacheRead:InitForCacheRead", err, bgTime, 1)
metric.SetWithLabels(err, map[string]string{exporter.Vol: cb.volume})
close(done)
}()
sb := strings.Builder{}
for _, s := range sources {

3
flashnode/flashnode.go
View File

@ -104,6 +104,9 @@ const (
cfgHandlerFileRoutineNumPerTask = "loadHandlerRoutineNumPerTask"
cfgManualScanLimitPerSecond = "manualScanLimitPerSecond"
cfgPrepareLimitPerSecond = "prepareLimitPerSecond"
paramIocc = "iocc"
paramFlow = "flow"
paramFactor = "factor"
)
// The FlashNode manages the inode block cache to speed the file reading.

38
flashnode/flashnode_op.go
View File

@ -179,25 +179,24 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
hitRateMap := f.cacheEngine.GetHitRate()
for dataPath, hitRate := range hitRateMap {
if hitRate < f.lowerHitRate {
log.LogWarnf("opCacheRead: flashnode %v dataPath(%v) is lower hitrate %v", f.localAddr, dataPath, hitRate)
log.LogDebugf("opCacheRead: flashnode %v dataPath(%v) is lower hitrate %v", f.localAddr, dataPath, hitRate)
errMetric := exporter.NewCounter("lowerHitRate")
errMetric.AddWithLabels(1, map[string]string{exporter.FlashNode: f.localAddr, exporter.Disk: dataPath, exporter.Err: "LowerHitRate"})
}
}
bgTime2 := stat.BeginStat()
taskDone := make(chan struct{})
missTaskDone := make(chan struct{})
missCacheMetric := exporter.NewTPCnt("MissCacheRead")
// try to cache more miss data, but reply to client more quickly
if writable := f.limitWrite.TryRunWithContext(ctx, int(req.Size_), func() {
if err = f.limitWrite.TryRunAsync(int(req.Size_), func() {
if block2, err := f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String(), false); err != nil {
log.LogWarnf("opCacheRead: CreateBlock failed, req(%v) err(%v)", req, err)
close(taskDone)
close(missTaskDone)
return
} else {
block2.InitOnceForCacheRead(f.cacheEngine, cr.Sources, taskDone)
block2.InitOnceForCacheRead(f.cacheEngine, cr.Sources, missTaskDone)
}
}); !writable {
err = fmt.Errorf("create block cache limited")
}); err != nil {
stat.EndStat("MissCacheReadLimit", err, bgTime2, 1)
missCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return
@ -207,7 +206,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
stat.EndStat("MissCacheReadCancel", ctx.Err(), bgTime2, 1)
missCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return ctx.Err()
case <-taskDone:
case <-missTaskDone:
block, err = f.cacheEngine.GetCacheBlockForRead(volume, cr.Inode, cr.FixedFileOffset, cr.Version, req.Size_)
}
stat.EndStat("MissCacheRead", err, bgTime2, 1)
@ -219,17 +218,24 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
bgTime2 := stat.BeginStat()
hitCacheMetric := exporter.NewTPCnt("HitCacheRead")
// reply to client as quick as possible if hit cache
err2 := f.limitRead.RunNoWait(int(req.Size_), false, func() {
err = f.doStreamReadRequest(ctx, conn, req, p, block)
hitTaskDone := make(chan struct{})
err = f.limitRead.TryRunAsync(int(req.Size_), func() {
err = f.doStreamReadRequest(ctx, conn, req, p, block, hitTaskDone)
})
if err2 != nil {
err = err2
if err != nil {
stat.EndStat("HitCacheRead", err, bgTime2, 1)
hitCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return
}
stat.EndStat("HitCacheRead", err, bgTime2, 1)
hitCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
select {
case <-ctx.Done():
stat.EndStat("HitCacheReadCancel", ctx.Err(), bgTime2, 1)
hitCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return ctx.Err()
case <-hitTaskDone:
stat.EndStat("HitCacheRead", err, bgTime2, 1)
hitCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
}
return
}
@ -386,7 +392,8 @@ func (f *FlashNode) getValidViewInfo(req *proto.FlashNodeManualTaskRequest) (met
return
}
func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req *proto.CacheReadRequest, p *proto.Packet, block *cachengine.CacheBlock) (err error) {
func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req *proto.CacheReadRequest, p *proto.Packet,
block *cachengine.CacheBlock, done chan struct{}) (err error) {
const action = "action[doStreamReadRequest]"
needReplySize := uint32(req.Size_)
offset := int64(req.Offset)
@ -397,6 +404,7 @@ func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req
f.metrics.updateReadCountMetric(block.GetRootPath())
f.metrics.updateReadBytesMetric(req.Size_, block.GetRootPath())
}
close(done)
}()
for needReplySize > 0 {
err = nil

12
flashnode/serve_http.go
View File

@ -113,9 +113,9 @@ func (f *FlashNode) handleSetWriteDiskQos(w http.ResponseWriter, r *http.Request
updated := false
for key, pVal := range map[string]*int{
cfgDiskWriteFlow: &f.diskWriteFlow,
cfgDiskWriteIocc: &f.diskWriteIocc,
cfgDiskWriteIoFactor: &f.diskWriteIoFactorFlow,
paramFlow: &f.diskWriteFlow,
paramIocc: &f.diskWriteIocc,
paramFactor: &f.diskWriteIoFactorFlow,
} {
val, err, has := parser(key)
if err != nil {
@ -154,9 +154,9 @@ func (f *FlashNode) handleSetReadDiskQos(w http.ResponseWriter, r *http.Request)
updated := false
for key, pVal := range map[string]*int{
cfgDiskReadFlow: &f.diskReadFlow,
cfgDiskReadIocc: &f.diskReadIocc,
cfgDiskReadIoFactor: &f.diskReadIoFactorFlow,
paramFlow: &f.diskReadFlow,
paramIocc: &f.diskReadIocc,
paramFactor: &f.diskReadIoFactorFlow,
} {
val, err, has := parser(key)
if err != nil {

13
util/limit_io.go
View File

@ -128,17 +128,16 @@ func (l *IoLimiter) TryRun(size int, taskFn func()) bool {
return true
}
func (l *IoLimiter) TryRunWithContext(ctx context.Context, size int, taskFn func()) bool {
if size > 0 {
if err := l.flow.WaitN(ctx, size); err != nil {
log.LogWarnf("action[limitio] tryrun wait flow with %d %s", size, err.Error())
return false
func (l *IoLimiter) TryRunAsync(size int, taskFn func()) error {
if size > 0 && l.limit > 0 {
if !l.flow.AllowN(time.Now(), size) {
return fmt.Errorf("flow limited")
}
}
if ok := l.getIO().TryRun(taskFn, true); !ok {
return false
return fmt.Errorf("run limited")
}
return true
return nil
}
func (l *IoLimiter) Status(ignoreUsed bool) (st LimiterStatus) {