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(client):fixing the prefetching sliding window logic

Browse Source 
close:#1000354285

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-09-22 16:11:53 +08:00 committed by 常亮
parent 9d45588a13
commit 5657874b3f
4 changed files with 43 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
sdk/data/stream/stream_aheadread.go
View File

@ -87,12 +87,9 @@ type AheadReadTask struct {
}
type AheadReadWindow struct {
taskC chan *AheadReadTask
curTaskMap map[string]interface{}
curTaskMutex sync.RWMutex
cache *AheadReadCache
streamer *Streamer
canAheadRead bool
taskC chan *AheadReadTask
cache *AheadReadCache
streamer *Streamer
}
type AheadReadCache struct {
@ -203,10 +200,9 @@ func (arc *AheadReadCache) Stop() {
func NewAheadReadWindow(arc *AheadReadCache, s *Streamer) *AheadReadWindow {
arw := &AheadReadWindow{
taskC: make(chan *AheadReadTask, arc.winCnt),
curTaskMap: make(map[string]interface{}),
cache: arc,
streamer: s,
taskC: make(chan *AheadReadTask, arc.winCnt),
cache: arc,
streamer: s,
}
go arw.backgroundAheadReadTask()
return arw
@ -234,19 +230,6 @@ func (arw *AheadReadWindow) doTask(task *AheadReadTask) {
realHost = "invaild"
)
key := fmt.Sprintf("%v-%v-%v-%v", task.p.inode, task.p.PartitionID, task.p.ExtentID, task.p.ExtentOffset)
if arw.putTaskIfNotExist(key) {
return
}
defer func() {
arw.deleteTask(key)
if log.EnableDebug() {
log.LogDebugf("aheadRead done:key(%v) size(%v) err(%v) %v reqID(%v)",
task.req, task.cacheSize, err, time.Since(task.time), task.reqID)
}
stat.EndStat("PrepareAheadData", err, task.logTime, 1)
stat.EndStat(fmt.Sprintf("PrepareAheadData[%v]", task.cacheType), err, task.logTime, 1)
stat.EndStat(fmt.Sprintf("PrepareAheadData[%v][%v]", task.cacheType, realHost), err, task.logTime, 1)
}()
if _, ok := arw.cache.blockCache.Load(key); ok {
return
}
@ -263,6 +246,15 @@ func (arw *AheadReadWindow) doTask(task *AheadReadTask) {
log.LogWarnf("aheadRead done:key(%v) no more block", key)
return
}
defer func() {
if log.EnableDebug() {
log.LogDebugf("doTask done:key(%v) size(%v) err(%v) %v reqID(%v)",
key, task.cacheSize, err, time.Since(task.time), task.reqID)
}
stat.EndStat("PrepareAheadData", err, task.logTime, 1)
stat.EndStat(fmt.Sprintf("PrepareAheadData[%v]", realHost), err, task.logTime, 1)
}()
log.LogDebugf("doTask key(%v) start", key)
cacheBlock.inode = task.p.inode
cacheBlock.partitionId = task.p.PartitionID
cacheBlock.extentId = task.p.ExtentID
@ -300,10 +292,10 @@ func (arw *AheadReadWindow) doTask(task *AheadReadTask) {
readBytes += int(rp.Size)
atomic.AddUint64(&cacheBlock.readBytes, uint64(rp.Size))
curSize := atomic.LoadUint64(&cacheBlock.readBytes)
log.LogDebugf("aheadRead doTask key(%v) curSize(%v) readBytes(%v) rp.Size(%v) task.cacheSize(%v) addr(%p)",
log.LogDebugf("doTask key(%v) curSize(%v) readBytes(%v) rp.Size(%v) task.cacheSize(%v) addr(%p)",
key, curSize, readBytes, rp.Size, task.cacheSize, cacheBlock)
if curSize > util.CacheReadBlockSize {
log.LogErrorf("aheadRead doTask out of range key(%v) curSize(%v) readBytes(%v) rp.Size(%v) task.cacheSize(%v) addr(%p)",
log.LogErrorf("doTask out of range key(%v) curSize(%v) readBytes(%v) rp.Size(%v) task.cacheSize(%v) addr(%p)",
key, curSize, readBytes, rp.Size, task.cacheSize, cacheBlock)
}
}
@ -313,35 +305,18 @@ func (arw *AheadReadWindow) doTask(task *AheadReadTask) {
if err == nil {
atomic.StoreUint32(&cacheBlock.state, AheadReadBlockStateInit)
arw.canAheadRead = true
log.LogDebugf("aheadRead ready: key(%v) offset(%v) size(%v) err(%v) %v reqID(%v)",
task.req, task.p.ExtentOffset, task.cacheSize, err, time.Since(task.time), task.reqID)
log.LogDebugf("doTask ready: key(%v) offset(%v) size(%v) err(%v) %v reqID(%v)",
key, task.p.ExtentOffset, task.cacheSize, err, time.Since(task.time), task.reqID)
return
}
atomic.StoreUint32(&cacheBlock.state, AheadReadBlockStateInit)
log.LogErrorf("aheadRead doTask inode(%v) offset(%v) size(%v) err(%v) - read failed,"+
log.LogErrorf("doTask inode(%v) offset(%v) size(%v) err(%v) - read failed,"+
" marking as recycled and deleting cache block",
cacheBlock.inode, task.p.ExtentOffset, task.cacheSize, err)
arw.cache.blockCache.Delete(key)
arw.cache.putAheadReadBlock(key, cacheBlock)
}
func (arw *AheadReadWindow) putTaskIfNotExist(key string) (exist bool) {
arw.curTaskMutex.Lock()
defer arw.curTaskMutex.Unlock()
if _, exist = arw.curTaskMap[key]; exist {
return
}
arw.curTaskMap[key] = struct{}{}
return
}
func (arw *AheadReadWindow) deleteTask(key string) {
arw.curTaskMutex.Lock()
defer arw.curTaskMutex.Unlock()
delete(arw.curTaskMap, key)
}
func (arw *AheadReadWindow) addNextTask(offset int, dp *wrapper.DataPartition, req *ExtentRequest, stTime time.Time,
reqID string, storageClass uint32) {
id := offset/util.CacheReadBlockSize + arw.cache.winCnt
@ -364,25 +339,29 @@ func (arw *AheadReadWindow) addNextTask(offset int, dp *wrapper.DataPartition, r
func (arw *AheadReadWindow) doMultiAheadRead(offset, remainSize int, req *ExtentRequest, dp *wrapper.DataPartition,
startTime time.Time, reqID string, storageClass uint32) {
if len(arw.curTaskMap) > 0 && !arw.canAheadRead {
return
}
cacheOffset := offset / util.CacheReadBlockSize * util.CacheReadBlockSize
key := fmt.Sprintf("%v-%v-%v-%v", arw.streamer.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, cacheOffset)
log.LogDebugf("doMultiAheadRead send: key(%v) reqID(%v) remainSize(%v) index(%v) req（%v）",
key, reqID, remainSize, offset/util.CacheReadBlockSize, req)
winCnt := arw.cache.winCnt
curReq := &ExtentRequest{
FileOffset: req.FileOffset,
Size: req.Size,
ExtentKey: req.ExtentKey,
}
id := 0
for i := offset / util.CacheReadBlockSize; i < winCnt; i++ {
id := offset / util.CacheReadBlockSize
for w := 0; w < winCnt; w++ {
if remainSize <= 0 {
id = 0
var err error
curReq.ExtentKey = arw.streamer.getNextExtent(int(curReq.ExtentKey.FileOffset))
if curReq.ExtentKey == nil {
log.LogWarnf("doMultiAheadRead send: key(%v) next ExtentKey is nil", key)
return
}
key = fmt.Sprintf("%v-%v-%v-%v", arw.streamer.inode, curReq.ExtentKey.PartitionId, curReq.ExtentKey.ExtentId, cacheOffset)
if dp, err = arw.streamer.client.dataWrapper.GetDataPartition(curReq.ExtentKey.PartitionId); err != nil {
log.LogWarnf("doMultiAheadRead send: key(%v) dp nil,err(%v)", key, err)
continue
}
remainSize = int(curReq.ExtentKey.Size)
@ -395,8 +374,8 @@ func (arw *AheadReadWindow) doMultiAheadRead(offset, remainSize int, req *Extent
task.cacheType = "pass"
task.logTime = stat.BeginStat()
task.reqID = reqID
log.LogDebugf("aheadRead send: key(%v) offset(%v) size(%v) reqID(%v)",
task.req, task.p.ExtentOffset, task.cacheSize, reqID)
log.LogDebugf("doMultiAheadRead send: key(%v) offset(%v) size(%v) reqID(%v)",
key, task.p.ExtentOffset, task.cacheSize, reqID)
arw.taskC <- task
}
remainSize -= size

10
sdk/data/stream/stream_aheadread_test.go
View File

@ -2,11 +2,10 @@ package stream
import (
"fmt"
"sync/atomic"
"testing"
"time"
"sync/atomic"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/data/wrapper"
"github.com/cubefs/cubefs/util"
@ -39,10 +38,9 @@ func newTestStreamerWithAheadRead(t *testing.T, partitionID uint64) (*Streamer,
s.extents = NewExtentCache(s.inode)
// Construct AheadReadWindow (no background goroutine needed)
s.aheadReadWindow = &AheadReadWindow{
cache: arc,
streamer: s,
taskC: make(chan *AheadReadTask, arc.winCnt),
curTaskMap: make(map[string]interface{}),
cache: arc,
streamer: s,
taskC: make(chan *AheadReadTask, arc.winCnt),
}
return s, arc
}

2
sdk/data/stream/stream_reader.go
View File

@ -258,7 +258,9 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
} else {
log.LogDebugf("Stream read: ino(%v) req(%v) s.needBCache(%v) s.client.bcacheEnable(%v) aheadReadEnable(%v)", s.inode, req, s.needBCache, s.client.bcacheEnable, s.aheadReadEnable)
if s.aheadReadEnable && req.ExtentKey.Size > util.CacheReadBlockSize {
bgTime := stat.BeginStat()
readBytes, err = s.aheadRead(req, storageClass)
stat.EndStat("ReadFromMem", err, bgTime, 1)
if err == nil && readBytes == req.Size {
total += readBytes
continue

6
util/stat/statistic.go
View File

@ -257,6 +257,12 @@ func WriteStat() error {
if PrintModuleStat != nil {
PrintModuleStat(ioStream)
}
mem, ok1 := gSt.typeInfoMap["ReadFromMem"]
read, ok2 := gSt.typeInfoMap["Read"]
if ok1 && ok2 {
rate := float64(mem.allCount) / float64(read.allCount)
fmt.Fprintf(ioStream, "AheadReadHitRate %.2f%%\n", rate*100)
}
fmt.Fprintf(ioStream, "%-42s|%10s|%8s|%8s|%8s|%8s|%8s|%8s|%8s|\n",
"", "TOTAL", "FAILED", "AVG(ms)", "MAX(ms)", "MIN(ms)",
">"+strconv.Itoa(int(gSt.timeOutUs[0])/1000)+"ms",