diff --git a/client/fs/super.go b/client/fs/super.go index b6470d694..8377d7647 100644 --- a/client/fs/super.go +++ b/client/fs/super.go @@ -250,6 +250,11 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) { OnGetInodeInfo: s.InodeGet, BcacheOnlyForNotSSD: opt.BcacheOnlyForNotSSD, + + AheadReadEnable: opt.AheadReadEnable, + AheadReadTotalMem: opt.AheadReadTotalMem, + AheadReadBlockTimeOut: opt.AheadReadBlockTimeOut, + AheadReadWindowCnt: opt.AheadReadWindowCnt, } s.ec, err = stream.NewExtentClient(extentConfig) diff --git a/client/fuse.go b/client/fuse.go index 0c2980053..ea052dcb8 100644 --- a/client/fuse.go +++ b/client/fuse.go @@ -970,6 +970,26 @@ func parseMountOption(cfg *config.Config) (*proto.MountOptions, error) { opt.DisableMountSubtype = GlobalMountOptions[proto.DisableMountSubtype].GetBool() opt.StreamRetryTimeout = int(GlobalMountOptions[proto.StreamRetryTimeOut].GetInt64()) + opt.AheadReadEnable = GlobalMountOptions[proto.AheadReadEnable].GetBool() + if opt.AheadReadEnable { + var ( + total uint64 + used uint64 + available int64 + ) + opt.AheadReadBlockTimeOut = int(GlobalMountOptions[proto.AheadReadBlockTimeOut].GetInt64()) + opt.AheadReadWindowCnt = int(GlobalMountOptions[proto.AheadReadWindowCnt].GetInt64()) + opt.AheadReadTotalMem = GlobalMountOptions[proto.AheadReadTotalMemGB].GetInt64() * util.GB + total, used, err = util.GetMemInfo() + if err != nil { + return nil, err + } + available = int64((total - used) / 2) + if available < opt.AheadReadTotalMem { + opt.AheadReadTotalMem = available + fmt.Printf("available ahead read mem: %v\n", available) + } + } if opt.MountPoint == "" || opt.Volname == "" || opt.Owner == "" || opt.Master == "" { return nil, errors.New(fmt.Sprintf("invalid config file: lack of mandatory fields, mountPoint(%v), volName(%v), owner(%v), masterAddr(%v)", opt.MountPoint, opt.Volname, opt.Owner, opt.Master)) } diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go index 36de713a8..56f900fa4 100644 --- a/datanode/data_partition_repair.go +++ b/datanode/data_partition_repair.go @@ -588,24 +588,17 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect err = nil reply := makeRspPacket(p.GetReqID(), p.GetPartitionID(), p.GetExtentID()) reply.SetStartT(p.GetStartT()) - currReadSize := uint32(util.Min(int(needReplySize), int(dp.GetRepairBlockSize()))) - if currReadSize == util.RepairReadBlockSize { + var currReadSize uint32 + if currReadSize == util.RepairReadBlockSize || currReadSize == util.BlockSize || currReadSize == util.CacheReadBlockSize { var data []byte - data, err = proto.Buffers.Get(util.RepairReadBlockSize) - if err != nil { - log.LogErrorf("[NormalExtentRepairRead] dp(%v) failed to get repair data, err(%v)", dp.partitionID, err) - return - } - reply.SetData(data) - } else if currReadSize == util.BlockSize { - var data []byte - data, err = proto.Buffers.Get(util.BlockSize) + data, err = proto.Buffers.Get(int(currReadSize)) if err != nil { log.LogErrorf("[NormalExtentRepairRead] dp(%v) failed to get repair data, err(%v)", dp.partitionID, err) return } reply.SetData(data) } else { + currReadSize = uint32(util.Min(int(needReplySize), int(dp.GetRepairBlockSize()))) reply.SetData(bytespool.Alloc(int(currReadSize))) } if !shallDegrade { @@ -644,7 +637,7 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect } needReplySize -= currReadSize offset += int64(currReadSize) - if currReadSize == util.ReadBlockSize || currReadSize == util.RepairReadBlockSize { + if currReadSize == util.ReadBlockSize || currReadSize == util.RepairReadBlockSize || currReadSize == util.CacheReadBlockSize { proto.Buffers.Put(reply.GetData()) } else { bytespool.Free(reply.GetData()) diff --git a/docs-zh/source/feature/aheadread.md b/docs-zh/source/feature/aheadread.md new file mode 100644 index 000000000..3cb5c514e --- /dev/null +++ b/docs-zh/source/feature/aheadread.md @@ -0,0 +1,33 @@ +# 预读（支持副本模式，不支持EC） + +修改内核预读参数(适用于内核未做优化情况) +``` bash +vi /etc/udev/rules.d/99-bdi.rules +ACTION=="add", SUBSYSTEM=="bdi", RUN+="/usr/local/bin/bdi_add_script %k" + +sudo udevadm control --reload-rules + +vi /usr/local/bin/bdi_add_script +#!/bin/bash +device=$1 +echo "`date '+%F %T'` FUSE device added: $device" >> /tmp/fuse.log +if [[ "$device" =~ "0:" ]];then + echo 2048 > /sys/devices/virtual/bdi/$device/read_ahead_kb +fi + +chmod +x /usr/local/bin/bdi_add_script +``` +配置文件参数含义如下表示： + + 参数 | 类型 | 含义 | 必需 | +|--------------|-------|-------------------------------------|----| +| aheadReadEnable | bool | 是否开启预读 | 是 | +| aheadReadTotalMemGB | int64 | 预读占用内存(默认10:GB)，若不够10G则占用当前可用内存的50% | 否 | +| aheadReadBlockTimeOut | int64 | 缓存块未命中的回收时间(默认3s) | 否 | +| aheadReadWindowCnt | int64 | 缓存滑动窗口的大小(默认：8) | 否 | + +![img.png](img.png) +::: tip 提示： +1、开启预读会占用一定的客户端内存，对于客户端内存限制高的场景，可以调节aheadReadTotalMemGB的大小，但是性能会有一定的衰减 +2、预读只对大于4M的文件有效，如果和bcache同时开启，会优先执行预读 +::: \ No newline at end of file diff --git a/docs-zh/source/feature/img.png b/docs-zh/source/feature/img.png new file mode 100644 index 000000000..68200db32 Binary files /dev/null and b/docs-zh/source/feature/img.png differ diff --git a/proto/mount_options.go b/proto/mount_options.go index 488761c1d..d3f957c82 100644 --- a/proto/mount_options.go +++ b/proto/mount_options.go @@ -79,6 +79,11 @@ const ( StreamRetryTimeOut BufferChanSize BcacheOnlyForNotSSD + // aheadread + AheadReadEnable + AheadReadTotalMemGB + AheadReadBlockTimeOut + AheadReadWindowCnt MaxMountOption ) @@ -177,6 +182,11 @@ func InitMountOptions(opts []MountOption) { opts[StreamRetryTimeOut] = MountOption{"streamRetryTimeout", "max stream retry timeout, s", "", int64(0)} opts[BcacheOnlyForNotSSD] = MountOption{"enableBcacheOnlyForNotSSD", "Enable block cache only for not ssd", "", false} + opts[AheadReadEnable] = MountOption{"aheadReadEnable", "enable ahead read", "", false} + opts[AheadReadTotalMemGB] = MountOption{"aheadReadTotalMemGB", "ahead read total mem(GB)", "", int64(10)} + opts[AheadReadBlockTimeOut] = MountOption{"aheadReadBlockTimeOut", "ahead read block expiration time", "", int64(3)} + opts[AheadReadWindowCnt] = MountOption{"aheadReadWindowCnt", "ahead read window block count", "", int64(8)} + for i := 0; i < MaxMountOption; i++ { flag.StringVar(&opts[i].cmdlineValue, opts[i].keyword, "", opts[i].description) } @@ -350,4 +360,9 @@ type MountOptions struct { VolStorageClass uint32 VolAllowedStorageClass []uint32 VolCacheDpStorageClass uint32 + + AheadReadEnable bool + AheadReadTotalMem int64 + AheadReadBlockTimeOut int + AheadReadWindowCnt int } diff --git a/sdk/data/stream/extent_client.go b/sdk/data/stream/extent_client.go index d73f33845..b88e55331 100644 --- a/sdk/data/stream/extent_client.go +++ b/sdk/data/stream/extent_client.go @@ -146,6 +146,11 @@ type ExtentConfig struct { OnGetInodeInfo GetInodeInfoFunc BcacheOnlyForNotSSD bool + + AheadReadEnable bool + AheadReadTotalMem int64 + AheadReadBlockTimeOut int + AheadReadWindowCnt int } type MultiVerMgr struct { @@ -191,6 +196,7 @@ type ExtentClient struct { getInodeInfo GetInodeInfoFunc bcacheOnlyForNotSSD bool InnerReq bool + AheadRead *AheadReadCache } func (client *ExtentClient) UidIsLimited(uid uint32) bool { @@ -344,6 +350,7 @@ retry: } client.readLimiter = rate.NewLimiter(readLimit, defaultReadLimitBurst) client.writeLimiter = rate.NewLimiter(writeLimit, defaultWriteLimitBurst) + client.AheadRead = NewAheadReadCache(config.AheadReadEnable, config.AheadReadTotalMem, config.AheadReadBlockTimeOut, config.AheadReadWindowCnt) if config.MaxStreamerLimit <= 0 { client.disableMetaCache = true diff --git a/sdk/data/stream/stream_aheadread.go b/sdk/data/stream/stream_aheadread.go new file mode 100644 index 000000000..e4eead585 --- /dev/null +++ b/sdk/data/stream/stream_aheadread.go @@ -0,0 +1,437 @@ +// Copyright 2025 The CubeFS Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +// implied. See the License for the specific language governing +// permissions and limitations under the License. + +package stream + +import ( + "fmt" + "net" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/sdk/data/wrapper" + "github.com/cubefs/cubefs/util" + "github.com/cubefs/cubefs/util/btree" + "github.com/cubefs/cubefs/util/log" +) + +type AheadReadCache struct { + enable bool + blockTimeOut int + winCnt int + availableBlockC chan *AheadReadBlock + availableBlockCnt int64 + totalBlockCnt int64 + stopC chan interface{} + blockCache *sync.Map +} + +type AheadReadBlock struct { + inode uint64 + partitionId uint64 + extentId uint64 + offset uint64 + size uint64 + data []byte + time int64 +} + +type AheadReadTask struct { + p *Packet + dnHosts []string + time time.Time + req *ExtentRequest + cacheSize int +} + +type AheadReadWindow struct { + taskC chan *AheadReadTask + curTaskMap map[string]interface{} + curTaskMutex sync.RWMutex + cache *AheadReadCache + streamer *Streamer + canAheadRead bool +} + +func NewAheadReadCache(enable bool, totalMem int64, blockTimeOut, winCnt int) *AheadReadCache { + if !enable { + return nil + } + arc := &AheadReadCache{ + enable: enable, + blockTimeOut: blockTimeOut, + winCnt: winCnt, + stopC: make(chan interface{}), + blockCache: new(sync.Map), + totalBlockCnt: totalMem / util.CacheReadBlockSize, + } + atomic.StoreInt64(&arc.availableBlockCnt, arc.totalBlockCnt) + arc.availableBlockC = make(chan *AheadReadBlock, arc.availableBlockCnt) + for i := int64(0); i < arc.totalBlockCnt; i++ { + block := &AheadReadBlock{ + data: make([]byte, util.CacheReadBlockSize), + } + arc.availableBlockC <- block + } + log.LogInfof("aheadRead enable(%v) totalMem(%v) availableBlockCnt(%v) winCnt(%v)", enable, totalMem, arc.availableBlockCnt, winCnt) + go arc.checkBlockTimeOut() + return arc +} + +func (arc *AheadReadCache) getAheadReadBlock() (block *AheadReadBlock, err error) { + select { + case block = <-arc.availableBlockC: + atomic.AddInt64(&arc.availableBlockCnt, -1) + default: + err = fmt.Errorf("availableBlockC is nil") + } + return +} + +func (arc *AheadReadCache) putAheadReadBlock(block *AheadReadBlock) { + arc.availableBlockC <- block + atomic.AddInt64(&arc.availableBlockCnt, 1) +} + +func (arc *AheadReadCache) checkBlockTimeOut() { + blockTimer := time.NewTimer(time.Second) + printTicker := time.NewTicker(time.Second * 10) + for { + select { + case <-arc.stopC: + blockTimer.Stop() + printTicker.Stop() + return + case <-blockTimer.C: + blockTimer.Stop() + arc.doCheckBlockTimeOut() + blockTimer.Reset(time.Second) + case <-printTicker.C: + log.LogInfof("totalAheadReadBlockCnt(%v) curAvailableCnt(%v)", arc.totalBlockCnt, atomic.LoadInt64(&arc.availableBlockCnt)) + } + } +} + +func (arc *AheadReadCache) doCheckBlockTimeOut() { + curTime := time.Now().Unix() + arc.blockCache.Range(func(key, value interface{}) bool { + bv := value.(*AheadReadBlock) + if curTime-bv.time > int64(arc.blockTimeOut) { + arc.blockCache.Delete(key) + arc.putAheadReadBlock(bv) + } + return true + }) +} + +func (arc *AheadReadCache) Stop() { + close(arc.stopC) + close(arc.availableBlockC) + arc.blockCache.Range(func(key, value interface{}) bool { + arc.blockCache.Delete(key) + return true + }) +} + +func NewAheadReadWindow(arc *AheadReadCache, s *Streamer) *AheadReadWindow { + arw := &AheadReadWindow{ + taskC: make(chan *AheadReadTask, arc.winCnt), + curTaskMap: make(map[string]interface{}), + cache: arc, + streamer: s, + } + go arw.backgroundAheadReadTask() + return arw +} + +func (arw *AheadReadWindow) backgroundAheadReadTask() { + for task := range arw.taskC { + go arw.doTask(task) + } +} + +func (arw *AheadReadWindow) doTask(task *AheadReadTask) { + var ( + err error + hosts = task.dnHosts + readBytes int + ) + 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) + log.LogDebugf("aheadRead step: fetch, key(%v) size(%v) err(%v) %v", key, task.cacheSize, err, time.Since(task.time)) + }() + if _, ok := arw.cache.blockCache.Load(key); ok { + return + } + cacheBlock, err := arw.cache.getAheadReadBlock() + if err != nil { + return + } + cacheBlock.inode = task.p.inode + cacheBlock.partitionId = task.p.PartitionID + cacheBlock.extentId = task.p.ExtentID + cacheBlock.offset = uint64(task.p.ExtentOffset) + cacheBlock.size = uint64(task.cacheSize) + + log.LogDebugf("aheadRead doTask key(%v) size(%v)", key, task.p.Size) + + for _, host := range hosts { + err = sendToNode(host, task.p, func(conn *net.TCPConn) (error, bool) { + readBytes = 0 + for readBytes < task.cacheSize { + rp := NewReply(task.p.ReqID, task.p.PartitionID, task.p.ExtentID) + bufSize := util.Min(util.CacheReadBlockSize, task.cacheSize-readBytes) + rp.Data = cacheBlock.data[readBytes : readBytes+bufSize] + if e := rp.readFromConn(conn, proto.ReadDeadlineTime); e != nil { + return e, false + } + if rp.ResultCode != proto.OpOk { + return fmt.Errorf("resultCode(%x) not ok", rp.ResultCode), false + } + readBytes += int(rp.Size) + } + return nil, false + }) + + if err != nil && strings.Contains(err.Error(), "timeout") { + arw.canAheadRead = false + arw.streamer.aheadReadEnable = false + break + } + if err == nil { + cacheBlock.time = time.Now().Unix() + arw.cache.blockCache.Store(key, cacheBlock) + arw.canAheadRead = true + return + } + } + + arw.cache.putAheadReadBlock(cacheBlock) + log.LogErrorf("aheadRead doTask inode(%v) offset(%v) size(%v) err(%v)", cacheBlock.inode, task.p.ExtentOffset, task.cacheSize, err) +} + +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, dnHosts []string, req *ExtentRequest, stTime time.Time) { + id := offset/util.CacheReadBlockSize + arw.cache.winCnt + remainSize := int(req.ExtentKey.Size) - id*util.CacheReadBlockSize + if remainSize <= 0 { + arw.doMultiAheadRead(0, 0, req, dnHosts, stTime) + return + } + // log.LogDebugf("addNextTask ek(%v) remainSize(%v) %v-%v-%v-%v", req.ExtentKey, arw.streamer.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, id*util.CacheReadBlockSize) + task := arw.getAheadReadTask(dnHosts, req, id, util.Min(remainSize, util.CacheReadBlockSize)) + if task == nil { + return + } + task.time = stTime + arw.taskC <- task +} + +func (arw *AheadReadWindow) doMultiAheadRead(offset, remainSize int, req *ExtentRequest, dnHosts []string, startTime time.Time) { + if len(arw.curTaskMap) > 0 && !arw.canAheadRead { + return + } + winCnt := 1 + if arw.canAheadRead { + 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++ { + if remainSize <= 0 { + id = 0 + var ( + dp *wrapper.DataPartition + err error + ) + curReq.ExtentKey = arw.streamer.getNextExtent(int(curReq.ExtentKey.FileOffset)) + if curReq.ExtentKey == nil { + return + } + if dp, err = arw.streamer.client.dataWrapper.GetDataPartition(curReq.ExtentKey.PartitionId); err != nil { + continue + } + dnHosts = dp.Hosts + remainSize = int(curReq.ExtentKey.Size) + } + + size := util.Min(remainSize, util.CacheReadBlockSize) + task := arw.getAheadReadTask(dnHosts, curReq, id, size) + if task != nil { + task.time = startTime + arw.taskC <- task + } + remainSize -= size + id++ + } +} + +func (arw *AheadReadWindow) getAheadReadTask(dnHosts []string, req *ExtentRequest, id int, size int) *AheadReadTask { + cacheOffset := int(id) * util.CacheReadBlockSize + p := NewReadPacket(req.ExtentKey, cacheOffset, size, arw.streamer.inode, req.FileOffset, true) + task := &AheadReadTask{ + p: p, + dnHosts: getRandomHostById(dnHosts, id), + req: req, + cacheSize: size, + } + return task +} + +func (arw *AheadReadWindow) evictCacheBlock(req *ExtentRequest) { + offset := req.FileOffset - int(req.ExtentKey.FileOffset) + int(req.ExtentKey.ExtentOffset) + cacheOffset := offset / util.CacheReadBlockSize * util.CacheReadBlockSize + key := fmt.Sprintf("%v-%v-%v-%v", arw.streamer.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, cacheOffset) + value, ok := arw.cache.blockCache.LoadAndDelete(key) + if ok { + bv := value.(*AheadReadBlock) + arw.cache.putAheadReadBlock(bv) + } +} + +func (s *Streamer) getNextExtent(offset int) (ek *proto.ExtentKey) { + s.extents.RLock() + defer s.extents.RUnlock() + s.extents.root.Ascend(func(i btree.Item) bool { + e := i.(*proto.ExtentKey) + if e.Size < util.CacheReadBlockSize { + return true + } + if e.FileOffset > uint64(offset) { + ek = e + return false + } + return true + }) + return +} + +func (s *Streamer) aheadRead(req *ExtentRequest) (readSize int, err error) { + var ( + offset int + cacheOffset int + cacheBlock *AheadReadBlock + step string + dp *wrapper.DataPartition + remainSize int + key string + ) + startTime := time.Now() + defer func() { + log.LogDebugf("aheadRead step: %v, inode(%v) key(%v) FileOffset(%v) reqSize(%v) readSize(%v) err(%v) %v", step, s.inode, key, req.FileOffset, req.Size, readSize, err, time.Since(startTime)) + }() + + if dp, err = s.client.dataWrapper.GetDataPartition(req.ExtentKey.PartitionId); err != nil { + return + } + + offset = req.FileOffset - int(req.ExtentKey.FileOffset) + int(req.ExtentKey.ExtentOffset) + cacheOffset = offset / util.CacheReadBlockSize * util.CacheReadBlockSize + remainSize = int(req.ExtentKey.Size) - cacheOffset + if remainSize < 0 { + remainSize = 0 + } + needSize := req.Size - readSize + for needSize > 0 { + key = fmt.Sprintf("%v-%v-%v-%v", s.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, cacheOffset) + val, ok := s.aheadReadWindow.cache.blockCache.Load(key) + if !ok { + break + } + step = "hit" + cacheBlock = val.(*AheadReadBlock) + cacheBlock.time = startTime.Unix() + log.LogDebugf("aheadRead cache hit inode(%v) FileOffset(%v) offset(%v) size(%v) cacheBlockOffset(%v) cacheBlockSize(%v) %v", s.inode, req.FileOffset, offset, needSize, cacheBlock.offset, cacheBlock.size, time.Since(startTime)) + if cacheBlock.offset <= uint64(offset) { + if (cacheBlock.offset + cacheBlock.size) > uint64(offset+needSize) { + copy(req.Data[readSize:req.Size], cacheBlock.data[offset-int(cacheBlock.offset):offset-int(cacheBlock.offset)+needSize]) + readSize += needSize + return + } else { + go s.aheadReadWindow.addNextTask(offset, dp.Hosts, req, startTime) + if offset-int(cacheBlock.offset) > int(cacheBlock.size) { + return + } + copy(req.Data, cacheBlock.data[offset-int(cacheBlock.offset):int(cacheBlock.size)]) + readSize += int(cacheBlock.size) + int(cacheBlock.offset) - offset + offset += readSize + cacheOffset = offset + needSize -= readSize + if needSize <= 0 { + return + } + continue + } + } + break + } + step = "pass" + go s.aheadReadWindow.doMultiAheadRead(offset, remainSize, req, dp.Hosts, startTime) + return +} + +func sendToNode(host string, p *Packet, getReply GetReplyFunc) (err error) { + var conn *net.TCPConn + if conn, err = StreamConnPool.GetConnect(host); err != nil { + return + } + defer StreamConnPool.PutConnect(conn, err != nil) + if err = p.WriteToConn(conn); err != nil { + return + } + err, _ = getReply(conn) + return +} + +func getRandomHostById(hosts []string, id int) []string { + if len(hosts) == 0 { + return nil + } + hIdx := id % len(hosts) + rhosts := make([]string, len(hosts)) + for i := 0; i < len(hosts); i++ { + if hIdx >= len(hosts) { + hIdx = 0 + } + rhosts[i] = hosts[hIdx] + hIdx++ + } + return rhosts +} diff --git a/sdk/data/stream/stream_reader.go b/sdk/data/stream/stream_reader.go index 380600b4e..b321b43d9 100644 --- a/sdk/data/stream/stream_reader.go +++ b/sdk/data/stream/stream_reader.go @@ -57,7 +57,9 @@ type Streamer struct { isCache bool openForWrite bool - rdonly bool + rdonly bool + aheadReadEnable bool + aheadReadWindow *AheadReadWindow } type bcacheKey struct { @@ -90,6 +92,10 @@ func NewStreamer(client *ExtentClient, inode uint64, openForWrite, isCache bool) s.setError() } } + if client.AheadRead != nil { + s.aheadReadEnable = client.AheadRead.enable + s.aheadReadWindow = NewAheadReadWindow(client.AheadRead, s) + } go s.server() go s.asyncBlockCache() return s @@ -203,7 +209,15 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32) total += req.Size log.LogDebugf("Stream read hole: ino(%v) req(%v) total(%v)", s.inode, req, total) } else { - log.LogDebugf("Stream read: ino(%v) req(%v) s.needBCache(%v) s.client.bcacheEnable(%v)", s.inode, req, s.needBCache, s.client.bcacheEnable) + 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 { + readBytes, err = s.aheadRead(req) + if err == nil && readBytes == req.Size { + total += readBytes + continue + } + log.LogDebugf("aheadRead inode(%v) FileOffset(%v) readBytes(%v) reqSize(%v) err(%v)", s.inode, req.FileOffset, readBytes, req.Size, err) + } if s.needBCache { bcacheMetric := exporter.NewCounter("fileReadL1Cache") bcacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName}) diff --git a/sdk/data/stream/stream_writer.go b/sdk/data/stream/stream_writer.go index c6f8fa6b3..b2d3fe734 100644 --- a/sdk/data/stream/stream_writer.go +++ b/sdk/data/stream/stream_writer.go @@ -409,6 +409,9 @@ begin: log.LogDebugf("action[streamer.write] ino %v doOverWriteByAppend extent key (%v)", s.inode, req.ExtentKey) writeSize, _, err, _ = s.doOverWriteByAppend(req, direct, storageClass, isMigration) } + if s.aheadReadEnable { + s.aheadReadWindow.evictCacheBlock(req) + } if s.client.bcacheEnable { cacheKey := util.GenerateKey(s.client.volumeName, s.inode, uint64(req.FileOffset)) go s.client.evictBcache(cacheKey) @@ -1031,6 +1034,9 @@ func (s *Streamer) open() { func (s *Streamer) release() error { s.refcnt-- + if s.client.AheadRead != nil { + s.aheadReadEnable = s.client.AheadRead.enable + } err := s.closeOpenHandler() if err != nil { s.abort() diff --git a/util/buf/buffer_pool.go b/util/buf/buffer_pool.go index a757beb70..746ef04ae 100644 --- a/util/buf/buffer_pool.go +++ b/util/buf/buffer_pool.go @@ -46,6 +46,7 @@ var ( HeadVerBuffersTotalLimit int64 HeadProtoVerBuffersTotalLimit int64 RepairBuffersTotalLimit int64 + cacheBuffersTotalLimit int64 ) var ( @@ -55,6 +56,7 @@ var ( headVerBuffersCount int64 headProtoVerBuffersCount int64 repairBuffersCount int64 + cacheBuffersCount int64 ) var ( @@ -63,6 +65,7 @@ var ( headBufVerAllocId uint64 headBufProtoVerAllocId uint64 repairBufAllocId uint64 + cacheBufAllocId uint64 ) var ( @@ -71,6 +74,7 @@ var ( headBufVerFreeId uint64 headBufProtoVerFreeId uint64 repairBufFreeId uint64 + cacheBufFreeId uint64 ) var ( @@ -80,6 +84,7 @@ var ( headVerBuffersRateLimit = rate.NewLimiter(rate.Limit(16), 16) headProtoVerBuffersRateLimit = rate.NewLimiter(rate.Limit(16), 16) repairBuffersRateLimit = rate.NewLimiter(rate.Limit(16), 16) + cacheBuffersRateLimit = rate.NewLimiter(rate.Limit(16), 16) ) func NewTinyBufferPool() *sync.Pool { @@ -143,6 +148,18 @@ func NewRepiarBufferPool() *sync.Pool { } } +func NewCacheBufferPool() *sync.Pool { + return &sync.Pool{ + New: func() interface{} { + if cacheBuffersTotalLimit != InvalidLimit && atomic.LoadInt64(&cacheBuffersCount) >= cacheBuffersTotalLimit { + ctx := context.Background() + cacheBuffersRateLimit.Wait(ctx) + } + return make([]byte, util.CacheReadBlockSize) + }, + } +} + // BufferPool defines the struct of a buffered pool with 4 objects. type BufferPool struct { headPools []chan []byte @@ -150,12 +167,14 @@ type BufferPool struct { headProtoVerPools []chan []byte normalPools []chan []byte repairPools []chan []byte + cachePools []chan []byte tinyPool *sync.Pool headPool *sync.Pool normalPool *sync.Pool headVerPool *sync.Pool headProtoVerPool *sync.Pool repairPool *sync.Pool + cachePool *sync.Pool } const slotCnt = 16 @@ -168,12 +187,14 @@ func NewBufferPool() (bufferP *BufferPool) { bufferP.headVerPools = make([]chan []byte, slotCnt) bufferP.headProtoVerPools = make([]chan []byte, slotCnt) bufferP.repairPools = make([]chan []byte, slotCnt) + bufferP.cachePools = make([]chan []byte, slotCnt) for i := 0; i < int(slotCnt); i++ { bufferP.headPools[i] = make(chan []byte, HeaderBufferPoolSize/slotCnt) bufferP.headVerPools[i] = make(chan []byte, HeaderBufferPoolSize/slotCnt) bufferP.headProtoVerPools[i] = make(chan []byte, HeaderBufferPoolSize/slotCnt) bufferP.normalPools[i] = make(chan []byte, HeaderBufferPoolSize/slotCnt) bufferP.repairPools[i] = make(chan []byte, HeaderBufferPoolSize/slotCnt) + bufferP.cachePools[i] = make(chan []byte, HeaderBufferPoolSize/slotCnt) } bufferP.tinyPool = NewTinyBufferPool() bufferP.headPool = NewHeadBufferPool() @@ -181,6 +202,7 @@ func NewBufferPool() (bufferP *BufferPool) { bufferP.headProtoVerPool = NewHeadProtoVerBufferPool() bufferP.normalPool = NewNormalBufferPool() bufferP.repairPool = NewRepiarBufferPool() + bufferP.cachePool = NewCacheBufferPool() return bufferP } @@ -247,6 +269,15 @@ func (bufferP *BufferPool) getTiny() (data []byte) { return bufferP.tinyPool.Get().([]byte) } +func (bufferP *BufferPool) getCache(id uint64) (data []byte) { + select { + case data = <-bufferP.cachePools[id%slotCnt]: + return + default: + return bufferP.cachePool.Get().([]byte) + } +} + // Get returns the data based on the given size. Different size corresponds to different object in the pool. func (bufferP *BufferPool) Get(size int) (data []byte, err error) { if size == util.PacketHeaderSize { @@ -272,6 +303,10 @@ func (bufferP *BufferPool) Get(size int) (data []byte, err error) { } else if size == util.DefaultTinySizeLimit { atomic.AddInt64(&tinyBuffersCount, 1) return bufferP.getTiny(), nil + } else if size == util.CacheReadBlockSize { + atomic.AddInt64(&cacheBuffersCount, 1) + id := atomic.AddUint64(&cacheBufAllocId, 1) + return bufferP.getCache(id), nil } return nil, fmt.Errorf("can only support 45 or 65536 bytes") } @@ -321,6 +356,15 @@ func (bufferP *BufferPool) putRepair(index int, data []byte) { } } +func (bufferP *BufferPool) putCache(index int, data []byte) { + select { + case bufferP.cachePools[index] <- data: + return + default: + bufferP.cachePool.Put(data) // nolint: staticcheck + } +} + // Put puts the given data into the buffer pool. func (bufferP *BufferPool) Put(data []byte) { if data == nil { @@ -350,5 +394,9 @@ func (bufferP *BufferPool) Put(data []byte) { } else if size == util.DefaultTinySizeLimit { bufferP.tinyPool.Put(data) // nolint: staticcheck atomic.AddInt64(&tinyBuffersCount, -1) + } else if size == util.CacheReadBlockSize { + atomic.AddInt64(&cacheBuffersCount, -1) + id := atomic.AddUint64(&cacheBufFreeId, 1) + bufferP.putCache(int(id%slotCnt), data) } } diff --git a/util/unit.go b/util/unit.go index 337884ee4..e737b4130 100644 --- a/util/unit.go +++ b/util/unit.go @@ -42,6 +42,7 @@ const ( BlockSize = 65536 * 2 ReadBlockSize = BlockSize RepairReadBlockSize = 512 * util.KB + CacheReadBlockSize = 4 * MB PerBlockCrcSize = 4 ExtentSize = BlockCount * BlockSize BlockHeaderSize = 4096