feat(client): "client support aheadread"

Signed-off-by: yanbin027 <yanbin027@ke.com>

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)

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))
 	}

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())

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同时开启，会优先执行预读
+:::

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
 }

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

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
+}

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})

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()

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)
 	}
 }

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