diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go
index 27a33c710..ba5f3be0e 100644
--- a/datanode/data_partition_repair.go
+++ b/datanode/data_partition_repair.go
@@ -25,6 +25,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cubefs/cubefs/blobstore/util/bytespool"
 	"github.com/cubefs/cubefs/util/exporter"
 
 	"github.com/cubefs/cubefs/proto"
@@ -550,7 +551,7 @@ func (dp *DataPartition) ExtentWithHoleRepairRead(request repl.PacketInterface,
 		}
 		needReplySize -= int64(currReadSize)
 		offset += int64(currReadSize)
-		if currReadSize == util.ReadBlockSize {
+		if currReadSize == util.ReadBlockSize || currReadSize == util.RepairReadBlockSize {
 			proto.Buffers.Put(reply.GetData())
 		}
 		if connect.RemoteAddr() != nil { // conn in testcase may not initialize
@@ -604,7 +605,7 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect
 			}
 			reply.SetData(data)
 		} else {
-			reply.SetData(make([]byte, currReadSize))
+			reply.SetData(bytespool.Alloc(int(currReadSize)))
 		}
 		if !shallDegrade {
 			partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName)
@@ -642,10 +643,13 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect
 		}
 		needReplySize -= currReadSize
 		offset += int64(currReadSize)
-		if currReadSize == util.ReadBlockSize {
+		if currReadSize == util.ReadBlockSize || currReadSize == util.RepairReadBlockSize {
 			proto.Buffers.Put(reply.GetData())
+		} else {
+			bytespool.Free(reply.GetData())
 		}
-		if connect.RemoteAddr() != nil {
+
+		if log.EnableInfo() && connect.RemoteAddr() != nil {
 			logContent := fmt.Sprintf("action[operatePacket] %v.",
 				reply.LogMessage(reply.GetOpMsg(), connect.RemoteAddr().String(), reply.GetStartT(), err))
 			log.LogReadf(logContent)
diff --git a/datanode/data_partition_repair_test.go b/datanode/data_partition_repair_test.go
index 313600e54..728143fbf 100644
--- a/datanode/data_partition_repair_test.go
+++ b/datanode/data_partition_repair_test.go
@@ -368,7 +368,7 @@ func extentStoreSnapshotRwTest(t *testing.T, s *storage.ExtentStore, id uint64,
 
 func extentReloadCheckSnapshotCrc(t *testing.T, path string, id uint64, crc uint32) (s *storage.ExtentStore) {
 	var err error
-	s, err = storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, false)
+	s, err = storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
 	require.NoError(t, err)
 
 	offset := int64(util.ExtentSize)
@@ -395,7 +395,7 @@ func mockInitWorker(t *testing.T, role string) *repairWorker {
 	worker.packChannel = make(chan repl.PacketInterface, 100)
 	path, _, err := getSrcPathExtentStore(role)
 	assert.True(t, err == nil)
-	s, err := storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, true)
+	s, err := storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, true)
 
 	require.NoError(t, err)
 	worker.dp = mockMakeDp(path)
@@ -515,7 +515,7 @@ func testDoNormalRepair(t *testing.T, normalId uint64, data []byte, crc uint32,
 	recvWorker.dp.extentStore.Close()
 
 	var err error
-	recvWorker.dp.extentStore, err = storage.NewExtentStore(recvWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, false)
+	recvWorker.dp.extentStore, err = storage.NewExtentStore(recvWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
 	require.NoError(t, err)
 	extentReloadCheckNormalCrc(t, recvWorker.dp.extentStore, normalId, crc)
 	recvWorker.dp.extentStore.Close()
@@ -524,9 +524,9 @@ func testDoNormalRepair(t *testing.T, normalId uint64, data []byte, crc uint32,
 func testDoSnapshotRepair(t *testing.T, normalId uint64, data []byte, crc uint32, isCreate bool) {
 	var err error
 
-	recvWorker.dp.extentStore, err = storage.NewExtentStore(recvWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, false)
+	recvWorker.dp.extentStore, err = storage.NewExtentStore(recvWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
 	require.NoError(t, err)
-	sendWorker.dp.extentStore, err = storage.NewExtentStore(sendWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, false)
+	sendWorker.dp.extentStore, err = storage.NewExtentStore(sendWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
 	require.NoError(t, err)
 	if isCreate {
 		s1 := sendWorker.dp.extentStore
diff --git a/datanode/disk.go b/datanode/disk.go
index 1e9ec619a..3af7fb2c4 100644
--- a/datanode/disk.go
+++ b/datanode/disk.go
@@ -249,12 +249,12 @@ func (d *Disk) updateQosLimiter() {
 	for i := proto.IopsReadType; i < proto.FlowWriteType; i++ {
 		log.LogInfof("action[updateQosLimiter] type %v limit %v", proto.QosTypeString(i), d.limitFactor[i].Limit())
 	}
-	log.LogInfof("action[updateQosLimiter] read(iocc:%d iops:%d flow:%d) write(iocc:%d iops:%d flow:%d)",
+	log.LogWarnf("action[updateQosLimiter] read(iocc:%d iops:%d flow:%d) write(iocc:%d iops:%d flow:%d)",
 		d.dataNode.diskReadIocc, d.dataNode.diskReadIops, d.dataNode.diskReadFlow,
 		d.dataNode.diskWriteIocc, d.dataNode.diskWriteIops, d.dataNode.diskWriteFlow)
-	d.limitRead.ResetIO(d.dataNode.diskReadIocc)
+	d.limitRead.ResetIO(d.dataNode.diskReadIocc, 0)
 	d.limitRead.ResetFlow(d.dataNode.diskReadFlow)
-	d.limitWrite.ResetIO(d.dataNode.diskWriteIocc)
+	d.limitWrite.ResetIO(d.dataNode.diskWriteIocc, d.dataNode.diskWQueFactor)
 	d.limitWrite.ResetFlow(d.dataNode.diskWriteFlow)
 }
 
diff --git a/datanode/limit_io.go b/datanode/limit_io.go
index 3b4d8711d..cd13a1255 100644
--- a/datanode/limit_io.go
+++ b/datanode/limit_io.go
@@ -25,7 +25,10 @@ import (
 	"golang.org/x/time/rate"
 )
 
-const minusOne = ^uint32(0)
+const (
+	minusOne           = ^uint32(0)
+	defaultQueueFactor = 8
+)
 
 type ioLimiter struct {
 	limit int
@@ -46,12 +49,16 @@ type LimiterStatus struct {
 // flow rate limiter's burst is double limit.
 // max queue size of io is 8-times io concurrency.
 func newIOLimiter(flowLimit, ioConcurrency int) *ioLimiter {
+	return newIOLimiterEx(flowLimit, ioConcurrency, 0)
+}
+
+func newIOLimiterEx(flowLimit, ioConcurrency, factor int) *ioLimiter {
 	flow := rate.NewLimiter(rate.Inf, 0)
 	if flowLimit > 0 {
-		flow = rate.NewLimiter(rate.Limit(flowLimit), 2*flowLimit)
+		flow = rate.NewLimiter(rate.Limit(flowLimit), flowLimit/2)
 	}
 	l := &ioLimiter{limit: flowLimit, flow: flow}
-	l.io.Store(newIOQueue(ioConcurrency))
+	l.io.Store(newIOQueue(ioConcurrency, factor))
 	return l
 }
 
@@ -66,17 +73,17 @@ func (l *ioLimiter) ResetFlow(flowLimit int) {
 		l.flow.SetBurst(0)
 	} else {
 		l.flow.SetLimit(rate.Limit(flowLimit))
-		l.flow.SetBurst(2 * flowLimit)
+		l.flow.SetBurst(flowLimit / 2)
 	}
 }
 
-func (l *ioLimiter) ResetIO(ioConcurrency int) {
-	q := l.io.Swap(newIOQueue(ioConcurrency)).(*ioQueue)
+func (l *ioLimiter) ResetIO(ioConcurrency, factor int) {
+	q := l.io.Swap(newIOQueue(ioConcurrency, factor)).(*ioQueue)
 	q.Close()
 }
 
 func (l *ioLimiter) Run(size int, taskFn func()) {
-	if size > 0 {
+	if size > 0 && l.limit > 0 {
 		if err := l.flow.WaitN(context.Background(), size); err != nil {
 			log.LogWarnf("action[limitio] run wait flow with %d %s", size, err.Error())
 		}
@@ -116,7 +123,7 @@ func (l *ioLimiter) Status() (st LimiterStatus) {
 }
 
 func (l *ioLimiter) Close() {
-	q := l.io.Swap(newIOQueue(0)).(*ioQueue)
+	q := l.io.Swap(newIOQueue(0, 0)).(*ioQueue)
 	q.Close()
 }
 
@@ -134,14 +141,18 @@ type ioQueue struct {
 	queue       chan *task
 }
 
-func newIOQueue(concurrency int) *ioQueue {
+func newIOQueue(concurrency, factor int) *ioQueue {
 	q := &ioQueue{concurrency: concurrency}
 	if q.concurrency <= 0 {
 		return q
 	}
 
+	if factor <= 0 {
+		factor = defaultQueueFactor
+	}
+
 	q.stopCh = make(chan struct{})
-	q.queue = make(chan *task, 8*concurrency)
+	q.queue = make(chan *task, factor*concurrency)
 	q.wg.Add(concurrency)
 	for ii := 0; ii < concurrency; ii++ {
 		go func() {
diff --git a/datanode/limit_io_test.go b/datanode/limit_io_test.go
index 09e14e505..481f40674 100644
--- a/datanode/limit_io_test.go
+++ b/datanode/limit_io_test.go
@@ -32,7 +32,7 @@ func TestLimitIOBase(t *testing.T) {
 	} {
 		l := newIOLimiter(flowIO[0], flowIO[1])
 		l.ResetFlow(flowIO[0])
-		l.ResetIO(flowIO[1])
+		l.ResetIO(flowIO[1], 0)
 		l.Run(0, f)
 		l.Run(10, f)
 		require.True(t, l.TryRun(1, f))
@@ -92,7 +92,7 @@ func TestLimitIOConcurrency(t *testing.T) {
 
 			time.Sleep(time.Microsecond)
 			l.ResetFlow(1 << 10)
-			l.ResetIO(10)
+			l.ResetIO(10, 0)
 		}
 	}()
 
diff --git a/datanode/partition.go b/datanode/partition.go
index 68f9b2396..4c5a9ff1d 100644
--- a/datanode/partition.go
+++ b/datanode/partition.go
@@ -409,7 +409,7 @@ func newDataPartition(dpCfg *dataPartitionCfg, disk *Disk, isCreate bool) (dp *D
 
 	partition.replicasInit()
 	partition.extentStore, err = storage.NewExtentStore(partition.path, dpCfg.PartitionID, dpCfg.PartitionSize,
-		partition.partitionType, isCreate)
+		partition.partitionType, disk.dataNode.cacheCap, isCreate)
 	if err != nil {
 		log.LogWarnf("action[newDataPartition] dp %v NewExtentStore failed %v", partitionID, err.Error())
 		return
diff --git a/datanode/server.go b/datanode/server.go
index b86f177ed..a71b16baf 100644
--- a/datanode/server.go
+++ b/datanode/server.go
@@ -112,13 +112,14 @@ const (
 	ConfigKeySmuxTotalStream   = "sumxTotalStream"    // int
 
 	// rate limit control enable
-	ConfigDiskQosEnable = "diskQosEnable" // bool
-	ConfigDiskReadIocc  = "diskReadIocc"  // int
-	ConfigDiskReadIops  = "diskReadIops"  // int
-	ConfigDiskReadFlow  = "diskReadFlow"  // int
-	ConfigDiskWriteIocc = "diskWriteIocc" // int
-	ConfigDiskWriteIops = "diskWriteIops" // int
-	ConfigDiskWriteFlow = "diskWriteFlow" // int
+	ConfigDiskQosEnable  = "diskQosEnable"  // bool
+	ConfigDiskReadIocc   = "diskReadIocc"   // int
+	ConfigDiskReadIops   = "diskReadIops"   // int
+	ConfigDiskReadFlow   = "diskReadFlow"   // int
+	ConfigDiskWriteIocc  = "diskWriteIocc"  // int
+	ConfigDiskWriteIops  = "diskWriteIops"  // int
+	ConfigDiskWriteFlow  = "diskWriteFlow"  // int
+	ConfigDiskWQueFactor = "diskWQueFactor" // int
 
 	// load/stop dp limit
 	ConfigDiskCurrentLoadDpLimit = "diskCurrentLoadDpLimit"
@@ -132,6 +133,7 @@ const (
 
 	// disk status becomes unavailable if disk error partition count reaches this value
 	ConfigKeyDiskUnavailablePartitionErrorCount = "diskUnavailablePartitionErrorCount"
+	ConfigKeyCacheCap                           = "cacheCap"
 
 	// storage device media type, for hybrid cloud, in string: SDD or HDD
 	ConfigMediaType = "mediaType"
@@ -190,6 +192,7 @@ type DataNode struct {
 	diskWriteIocc           int
 	diskWriteIops           int
 	diskWriteFlow           int
+	diskWQueFactor          int
 	dpMaxRepairErrCnt       uint64
 	clusterUuid             string
 	clusterUuidEnable       bool
@@ -205,6 +208,7 @@ type DataNode struct {
 	diskUnavailablePartitionErrorCount uint64 // disk status becomes unavailable when disk error partition count reaches this value
 	started                            int32
 	dpBackupTimeout                    time.Duration
+	cacheCap                           int
 	mediaType                          uint32              // type of storage hardware medi
 	nodeForbidWriteOpOfProtoVer0       bool                // whether forbid by node granularity,
 	VolsForbidWriteOpOfProtoVer0       map[string]struct{} // whether forbid by volume granularity,
@@ -370,9 +374,8 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
 	}
 	s.port = port
 
-	/*for _, ip := range cfg.GetSlice(proto.MasterAddr) {
-		MasterClient.AddNode(ip.(string))
-	}*/
+	s.cacheCap = cfg.GetInt(ConfigKeyCacheCap)
+	log.LogWarnf("parseConfig: cache cap size %d", s.cacheCap)
 
 	updateInterval := cfg.GetInt(configNameResolveInterval)
 	if updateInterval <= 0 || updateInterval > 60 {
diff --git a/datanode/server_handler.go b/datanode/server_handler.go
index 5d62cbec9..9be7ad7f7 100644
--- a/datanode/server_handler.go
+++ b/datanode/server_handler.go
@@ -325,12 +325,13 @@ func (s *DataNode) setDiskQos(w http.ResponseWriter, r *http.Request) {
 
 	updated := false
 	for key, pVal := range map[string]*int{
-		ConfigDiskReadIocc:  &s.diskReadIocc,
-		ConfigDiskReadIops:  &s.diskReadIops,
-		ConfigDiskReadFlow:  &s.diskReadFlow,
-		ConfigDiskWriteIocc: &s.diskWriteIocc,
-		ConfigDiskWriteIops: &s.diskWriteIops,
-		ConfigDiskWriteFlow: &s.diskWriteFlow,
+		ConfigDiskReadIocc:   &s.diskReadIocc,
+		ConfigDiskReadIops:   &s.diskReadIops,
+		ConfigDiskReadFlow:   &s.diskReadFlow,
+		ConfigDiskWriteIocc:  &s.diskWriteIocc,
+		ConfigDiskWriteIops:  &s.diskWriteIops,
+		ConfigDiskWriteFlow:  &s.diskWriteFlow,
+		ConfigDiskWQueFactor: &s.diskWQueFactor,
 	} {
 		val, err, has := parser(key)
 		if err != nil {
diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go
index 083936171..374d56d37 100644
--- a/datanode/wrap_operator.go
+++ b/datanode/wrap_operator.go
@@ -89,7 +89,9 @@ func (s *DataNode) OperatePacket(p *repl.Packet, c net.Conn) (err error) {
 		tpLabels map[string]string
 		tpObject *exporter.TimePointCount
 	)
-	log.LogDebugf("action[OperatePacket] %v, pack [%v]", p.GetOpMsg(), p)
+	if log.EnableDebug() {
+		log.LogDebugf("action[OperatePacket] %v, pack [%v]", p.GetOpMsg(), p)
+	}
 	shallDegrade := p.ShallDegrade()
 	sz := p.Size
 	if !shallDegrade {
@@ -111,7 +113,7 @@ func (s *DataNode) OperatePacket(p *repl.Packet, c net.Conn) (err error) {
 			} else {
 				log.LogErrorf(logContent)
 			}
-		} else {
+		} else if log.EnableInfo() {
 			logContent := fmt.Sprintf("action[OperatePacket] %v.",
 				p.LogMessage(p.GetOpMsg(), c.RemoteAddr().String(), start, nil))
 			switch p.Opcode {
@@ -138,6 +140,11 @@ func (s *DataNode) OperatePacket(p *repl.Packet, c net.Conn) (err error) {
 		if !shallDegrade {
 			tpObject.SetWithLabels(err, tpLabels)
 		}
+
+		if p.IsReadOperation() {
+			now := time.Now().UnixNano()
+			exporter.RecodCost("data_read_cost", (now-start)/1e3)
+		}
 	}()
 
 	switch p.Opcode {
diff --git a/datanode/wrap_operator_test.go b/datanode/wrap_operator_test.go
index 39c205d60..71c027d75 100644
--- a/datanode/wrap_operator_test.go
+++ b/datanode/wrap_operator_test.go
@@ -29,7 +29,7 @@ import (
 func newExtentStoreForOperatorTest(t *testing.T) (store *storage.ExtentStore) {
 	path, err := os.MkdirTemp("", "")
 	require.NoError(t, err)
-	store, err = storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, true)
+	store, err = storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, true)
 	require.NoError(t, err)
 	return
 }
diff --git a/datanode/wrap_prepare.go b/datanode/wrap_prepare.go
index e136fdb4d..d2b2ab4e1 100644
--- a/datanode/wrap_prepare.go
+++ b/datanode/wrap_prepare.go
@@ -120,7 +120,10 @@ func (s *DataNode) checkPacketAndPrepare(p *repl.Packet) error {
 		err      error
 	)
 
-	log.LogDebugf("action[prepare.checkPacketAndPrepare] pack opcode (%v) p.IsLeaderPacket(%v) p (%v)", p.Opcode, p.IsLeaderPacket(), p)
+	if log.EnableDebug() {
+		log.LogDebugf("action[prepare.checkPacketAndPrepare] pack opcode (%v) p.IsLeaderPacket(%v) p (%v)", p.Opcode,
+			p.IsLeaderPacket(), p)
+	}
 	if p.IsRandomWrite() || p.IsSnapshotModWriteAppendOperation() || p.IsNormalWriteOperation() {
 		if err = partition.CheckWriteVer(p); err != nil {
 			return err
diff --git a/metanode/partition_fsmop_extend.go b/metanode/partition_fsmop_extend.go
index ef6b78d0f..8d615f269 100644
--- a/metanode/partition_fsmop_extend.go
+++ b/metanode/partition_fsmop_extend.go
@@ -47,7 +47,7 @@ func (mp *metaPartition) fsmLockDir(req *proto.LockDirRequest) (resp *proto.Lock
 
 	log.LogDebugf("fsmLockDir: req info %s, val %s", req.String(), newVal)
 
-	var newExtend = NewExtend(req.Inode)
+	newExtend := NewExtend(req.Inode)
 	treeItem := mp.extendTree.CopyGet(newExtend)
 
 	var oldValue []byte
@@ -72,7 +72,7 @@ func (mp *metaPartition) fsmLockDir(req *proto.LockDirRequest) (resp *proto.Lock
 	if err != nil {
 		log.LogErrorf("fsmLockDir: parse req failed, req %s, old %s, err %s", req.String(), string(oldValue), err.Error())
 		resp.Status = proto.OpExistErr
-		return 
+		return
 	}
 
 	log.LogDebugf("fsmLockDir: get old lock dir info, req %v, old %d, expire %d", req, oldLkId, oldExpire)
@@ -106,7 +106,7 @@ func (mp *metaPartition) fsmUnlockDir(req *proto.LockDirRequest) (resp *proto.Lo
 	newVal := fmt.Sprintf("%d|%d", req.LockId, newExpire)
 	log.LogDebugf("fsmUnlockDir: req info %s, val %s", req, newVal)
 
-	var newExtend = NewExtend(req.Inode)
+	newExtend := NewExtend(req.Inode)
 	treeItem := mp.extendTree.CopyGet(newExtend)
 
 	var oldValue []byte
diff --git a/proto/packet.go b/proto/packet.go
index d305fc83b..c655957a1 100644
--- a/proto/packet.go
+++ b/proto/packet.go
@@ -364,7 +364,8 @@ type Packet struct {
 	// version-1: from v3.4
 	ProtoVersion uint32
 
-	VerList []*VolVersionInfo
+	VerList  []*VolVersionInfo
+	noPrefix bool
 }
 
 func IsTinyExtentType(extentType uint8) bool {
@@ -503,6 +504,8 @@ func (p *Packet) GetOpMsg() (m string) {
 		m = "OpMetaReadDir"
 	case OpMetaReadDirLimit:
 		m = "OpMetaReadDirLimit"
+	case OpMetaLockDir:
+		m = "OpMetaLockDir"
 	case OpMetaInodeGet:
 		m = "OpMetaInodeGet"
 	case OpMetaBatchInodeGet:
@@ -1234,7 +1237,7 @@ func (p *Packet) GetUniqueLogId() (m string) {
 		m = m + fmt.Sprintf("_ResultMesg(%v)", p.GetResultMsg())
 	}()
 	if p.HasPrepare {
-		m = p.mesg
+		m = p.GetMsg()
 		return
 	}
 	m = fmt.Sprintf("Req(%v)_Partition(%v)_", p.ReqID, p.PartitionID)
@@ -1267,7 +1270,23 @@ func (p *Packet) GetUniqueLogId() (m string) {
 	return
 }
 
+func (p *Packet) GetMsg() string {
+	if p.noPrefix {
+		p.mesg = fmt.Sprintf("Req(%v)_Partition(%v)_Extent(%v)_ExtentOffset(%v)_KernelOffset(%v)_"+
+			"Size(%v)_Opcode(%v)_CRC(%v), m(%s)",
+			p.ReqID, p.PartitionID, p.ExtentID, p.ExtentOffset,
+			p.KernelOffset, p.Size, p.GetOpMsg(), p.CRC, p.mesg)
+		return p.mesg
+	}
+	return p.mesg
+}
+
 func (p *Packet) setPacketPrefix() {
+	if !log.EnableDebug() && p.IsReadOperation() {
+		p.noPrefix = true
+		return
+	}
+
 	p.mesg = fmt.Sprintf("Req(%v)_Partition(%v)_", p.ReqID, p.PartitionID)
 	if (p.Opcode == OpSplitMarkDelete || (IsTinyExtentType(p.ExtentType) && p.Opcode == OpMarkDelete)) && len(p.Data) > 0 {
 		ext := new(TinyExtentDeleteRecord)
diff --git a/repl/repl_protocol.go b/repl/repl_protocol.go
index 211d7c0f4..81bfb940b 100644
--- a/repl/repl_protocol.go
+++ b/repl/repl_protocol.go
@@ -162,8 +162,10 @@ func (ft *FollowerTransport) readFollowerResult(request *FollowerPacket) (err er
 		err = fmt.Errorf(string(reply.Data[:reply.Size]))
 		return
 	}
-	log.LogDebugf("action[ActionReceiveFromFollower] %v.", reply.LogMessage(ActionReceiveFromFollower,
-		ft.addr, request.StartT, err))
+	if log.EnableDebug() {
+		log.LogDebugf("action[ActionReceiveFromFollower] %v.", reply.LogMessage(ActionReceiveFromFollower,
+			ft.addr, request.StartT, err))
+	}
 	return
 }
 
@@ -416,7 +418,10 @@ func (rp *ReplProtocol) writeResponse(reply *Packet) {
 	defer func() {
 		reply.clean()
 	}()
-	log.LogDebugf("writeResponse.opcode %v reply %v conn(%v)", reply.Opcode, reply.GetUniqueLogId(), rp.sourceConn.RemoteAddr().String())
+	if log.EnableDebug() {
+		log.LogDebugf("writeResponse.opcode %v reply %v conn(%v)", reply.Opcode, reply.GetUniqueLogId(),
+			rp.sourceConn.RemoteAddr().String())
+	}
 	if reply.IsErrPacket() {
 		err = fmt.Errorf(reply.LogMessage(ActionWriteToClient, rp.sourceConn.RemoteAddr().String(),
 			reply.StartT, fmt.Errorf(string(reply.Data[:reply.Size]))))
@@ -429,7 +434,9 @@ func (rp *ReplProtocol) writeResponse(reply *Packet) {
 		}
 		rp.Stop()
 	}
-	log.LogDebugf("try rsp opcode %v %v %v", rp.replId, reply.Opcode, rp.sourceConn.RemoteAddr().String())
+	if log.EnableDebug() {
+		log.LogDebugf("try rsp opcode %v %v %v", rp.replId, reply.Opcode, rp.sourceConn.RemoteAddr().String())
+	}
 	// execute the post-processing function
 	rp.postFunc(reply)
 	if !reply.NeedReply {
@@ -445,8 +452,10 @@ func (rp *ReplProtocol) writeResponse(reply *Packet) {
 		log.LogErrorf(err.Error())
 		rp.Stop()
 	}
-	log.LogDebugf(reply.LogMessage(ActionWriteToClient,
-		rp.sourceConn.RemoteAddr().String(), reply.StartT, err))
+	if log.EnableDebug() {
+		log.LogDebugf(reply.LogMessage(ActionWriteToClient,
+			rp.sourceConn.RemoteAddr().String(), reply.StartT, err))
+	}
 }
 
 // Stop stops the replication protocol.
diff --git a/sdk/data/manager/limiter.go b/sdk/data/manager/limiter.go
index 4c9a39f5b..ec4df50d4 100644
--- a/sdk/data/manager/limiter.go
+++ b/sdk/data/manager/limiter.go
@@ -501,11 +501,17 @@ func (limitManager *LimitManager) SetClientLimit(limit *proto.LimitRsp2Client) {
 }
 
 func (limitManager *LimitManager) ReadAlloc(ctx context.Context, size int) {
+	if !limitManager.enable {
+		return
+	}
 	limitManager.WaitN(ctx, limitManager.limitMap[proto.IopsReadType], 1)
 	limitManager.WaitN(ctx, limitManager.limitMap[proto.FlowReadType], size)
 }
 
 func (limitManager *LimitManager) WriteAlloc(ctx context.Context, size int) {
+	if !limitManager.enable {
+		return
+	}
 	limitManager.WaitN(ctx, limitManager.limitMap[proto.IopsWriteType], 1)
 	limitManager.WaitN(ctx, limitManager.limitMap[proto.FlowWriteType], size)
 }
diff --git a/sdk/data/stream/extent_cache.go b/sdk/data/stream/extent_cache.go
index 8bda1ce03..31db42652 100644
--- a/sdk/data/stream/extent_cache.go
+++ b/sdk/data/stream/extent_cache.go
@@ -123,7 +123,9 @@ func (cache *ExtentCache) update(gen, size uint64, force bool, eks []proto.Exten
 	cache.root.Clear(false)
 	for _, ek := range eks {
 		extent := ek
-		log.LogDebugf("action[update] update cache ino(%v) replace or insert ek [%v]", cache.inode, ek.String())
+		if log.EnableDebug() {
+			log.LogDebugf("action[update] update cache ino(%v) replace or insert ek [%v]", cache.inode, ek.String())
+		}
 		cache.root.ReplaceOrInsert(&extent)
 	}
 }
diff --git a/sdk/data/stream/extent_client.go b/sdk/data/stream/extent_client.go
index 477414855..d73f33845 100644
--- a/sdk/data/stream/extent_client.go
+++ b/sdk/data/stream/extent_client.go
@@ -24,38 +24,23 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/cubefs/cubefs/depends/bazil.org/fuse"
 	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/sdk/data/manager"
 	"github.com/cubefs/cubefs/sdk/data/wrapper"
 	"github.com/cubefs/cubefs/sdk/meta"
 	"github.com/cubefs/cubefs/util"
 	"github.com/cubefs/cubefs/util/errors"
+	"github.com/cubefs/cubefs/util/exporter"
 	"github.com/cubefs/cubefs/util/log"
 	"github.com/cubefs/cubefs/util/stat"
-	"github.com/prometheus/client_golang/prometheus"
 
 	"golang.org/x/time/rate"
 )
 
-var (
-	clientMetric = prometheus.NewSummaryVec(
-		prometheus.SummaryOpts{
-			Namespace:  "cubefs",
-			Subsystem:  "client",
-			Name:       "client_cost_time",
-			Help:       "time cost in cubefs sdk",
-			Objectives: map[float64]float64{0.5: 0.05, 0.75: 0.025, 0.9: 0.01, 0.95: 0.005, 0.99: 0.001, 0.999: 0.0001, 0.9999: 0.00001},
-		}, []string{"api"})
-	readReqCountMetric = prometheus.NewGauge(
-		prometheus.GaugeOpts{
-			Name: "read_req_cnt",
-		})
-)
+var reqChanSize = defaultChanSize
 
-func init() {
-	prometheus.MustRegister(clientMetric)
-	prometheus.MustRegister(readReqCountMetric)
-}
+const defaultChanSize = 64
 
 type (
 	SplitExtentKeyFunc            func(parentInode, inode uint64, key proto.ExtentKey, storageClass uint32) error
@@ -120,6 +105,12 @@ func init() {
 	}}
 }
 
+func SetReqChansize(size int) {
+	if size > defaultChanSize {
+		reqChanSize = size
+	}
+}
+
 type ExtentConfig struct {
 	Volume            string
 	Masters           []string
@@ -215,6 +206,10 @@ func (client *ExtentClient) UidIsLimited(uid uint32) bool {
 	return false
 }
 
+func (client *ExtentClient) readLimit() bool {
+	return client.readLimiter.Limit() != rate.Inf
+}
+
 func (client *ExtentClient) evictStreamer() bool {
 	// remove from list
 	item := client.streamerList.Back()
@@ -463,6 +458,30 @@ func (client *ExtentClient) OpenStream(inode uint64, openForWrite, isCache bool)
 	return s.IssueOpenRequest()
 }
 
+func (client *ExtentClient) OpenStreamRdonly(inode uint64, rdonly bool) error {
+	client.streamerLock.Lock()
+	s, ok := client.streamers[inode]
+	if !ok {
+		s = NewStreamer(client, inode, false, false)
+		client.streamers[inode] = s
+		s.rdonly = rdonly
+	}
+
+	if s.rdonly {
+		defer client.streamerLock.Unlock()
+		// stream is rdonly, but open again by writable, return err
+		if !rdonly {
+			log.LogErrorf("OpenStreamRdonly: rdonly stream can't be open again for write, s %s, rdonly %v", s.String(), rdonly)
+			return fuse.EPERM
+		}
+
+		s.refcnt++
+		return nil
+	}
+
+	return s.IssueOpenRequest()
+}
+
 // Open request shall grab the lock until request is sent to the request channel
 func (client *ExtentClient) OpenStreamWithCache(inode uint64, needBCache, openForWrite, isCache bool) error {
 	client.streamerLock.Lock()
@@ -478,7 +497,7 @@ func (client *ExtentClient) OpenStreamWithCache(inode uint64, needBCache, openFo
 	if !s.isOpen && !client.disableMetaCache {
 		s.isOpen = true
 		log.LogDebugf("open stream again, ino(%v)", s.inode)
-		s.request = make(chan interface{}, 64)
+		s.request = make(chan interface{}, reqChanSize)
 		s.pendingCache = make(chan bcacheKey, 1)
 		go s.server()
 		go s.asyncBlockCache()
@@ -494,7 +513,16 @@ func (client *ExtentClient) CloseStream(inode uint64) error {
 		client.streamerLock.Unlock()
 		return nil
 	}
-	log.LogDebugf("CloseStream streamer(%v)", s)
+
+	if log.EnableDebug() {
+		log.LogDebugf("CloseStream: stream(%s)", s.String())
+	}
+	if s.rdonly {
+		s.refcnt--
+		client.streamerLock.Unlock()
+		return nil
+	}
+
 	return s.IssueReleaseRequest()
 }
 
@@ -506,6 +534,22 @@ func (client *ExtentClient) EvictStream(inode uint64) error {
 		client.streamerLock.Unlock()
 		return nil
 	}
+
+	log.LogDebugf("EvictStream: stream(%v)", s)
+
+	if s.rdonly {
+		defer client.streamerLock.Unlock()
+		if s.refcnt > 0 || len(s.request) != 0 {
+			log.LogWarnf("evict: streamer(%v) refcnt(%v)", s.String(), s.refcnt)
+			return nil
+		}
+
+		if s.client.disableMetaCache || !s.needBCache {
+			delete(s.client.streamers, s.inode)
+		}
+		return nil
+	}
+
 	if s.isOpen {
 		err := s.IssueEvictRequest()
 		if err != nil {
@@ -658,16 +702,15 @@ func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int
 	// log.LogErrorf("======> ExtentClient Read Enter, inode(%v), len(data)=(%v), offset(%v), size(%v) storageClass(%v) isMigration(%v)",
 	//	inode, len(data), offset, size, storageClass, isMigration)
 	// t1 := time.Now()
-	beg := time.Now()
-	defer func() {
-		clientMetric.WithLabelValues("Read").Observe(float64(time.Since(beg).Microseconds()))
-	}()
-
-	readReqCountMetric.Inc()
 	if size == 0 {
 		return
 	}
 
+	beg := time.Now()
+	defer func() {
+		exporter.RecodCost("Read", time.Since(beg).Microseconds())
+	}()
+
 	s := client.GetStreamer(inode)
 	if s == nil {
 		log.LogErrorf("Read: stream is not opened yet, ino(%v) offset(%v) size(%v)", inode, offset, size)
@@ -676,9 +719,7 @@ func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int
 
 	var errGetExtents error
 	s.once.Do(func() {
-		beg = time.Now()
 		errGetExtents = s.GetExtents(isMigration)
-		clientMetric.WithLabelValues("Read_GetExtents").Observe(float64(time.Since(beg).Microseconds()))
 		if log.EnableDebug() {
 			log.LogDebugf("Read: ino(%v) offset(%v) size(%v) storageClass(%v) isMigration(%v) errGetExtents(%v)",
 				inode, offset, size, storageClass, isMigration, errGetExtents)
@@ -690,18 +731,15 @@ func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int
 		return 0, err
 	}
 
-	beg = time.Now()
-	err = s.IssueFlushRequest()
-	if err != nil {
-		return
+	if !s.rdonly || s.dirty {
+		err = s.IssueFlushRequest()
+		if err != nil {
+			return
+		}
 	}
-	clientMetric.WithLabelValues("Read_Flush").Observe(float64(time.Since(beg).Microseconds()))
 
-	beg = time.Now()
 	read, err = s.read(data, offset, size, storageClass)
-	clientMetric.WithLabelValues("Read_read").Observe(float64(time.Since(beg).Microseconds()))
 	// log.LogErrorf("======> ExtentClient Read Exit, inode(%v), time[%v us].", inode, time.Since(t1).Microseconds())
-	readReqCountMetric.Dec()
 	return
 }
 
@@ -789,7 +827,7 @@ func (client *ExtentClient) GetStreamer(inode uint64) *Streamer {
 	}
 	if !s.isOpen {
 		s.isOpen = true
-		s.request = make(chan interface{}, 64)
+		s.request = make(chan interface{}, reqChanSize)
 		s.pendingCache = make(chan bcacheKey, 1)
 		go s.server()
 		go s.asyncBlockCache()
diff --git a/sdk/data/stream/extent_handler.go b/sdk/data/stream/extent_handler.go
index 551338b06..5795b48a9 100644
--- a/sdk/data/stream/extent_handler.go
+++ b/sdk/data/stream/extent_handler.go
@@ -329,7 +329,9 @@ func (eh *ExtentHandler) sender() {
 			}
 			packet.StartT = time.Now().UnixNano()
 
-			log.LogDebugf("ExtentHandler sender: extent allocated, eh(%v) dp(%v) extID(%v) packet(%v)", eh, eh.dp, eh.extID, packet.GetUniqueLogId())
+			if log.EnableDebug() {
+				log.LogDebugf("ExtentHandler sender: extent allocated, eh(%v) dp(%v) extID(%v) packet(%v)", eh, eh.dp, eh.extID, packet.GetUniqueLogId())
+			}
 
 			if err = packet.writeToConn(eh.conn); err != nil {
 				log.LogWarnf("sender writeTo: failed, eh(%v) err(%v) packet(%v)", eh, err, packet)
@@ -520,7 +522,7 @@ func (eh *ExtentHandler) cleanup() (err error) {
 			eh.conn = nil
 			// TODO unhandled error
 			status := eh.getStatus()
-			StreamConnPool.PutConnect(conn, status >= ExtentStatusRecovery)
+			StreamWriteConnPool.PutConnect(conn, status >= ExtentStatusRecovery)
 		}
 		close(eh.stop)
 	})
@@ -731,7 +733,7 @@ func (eh *ExtentHandler) allocateExtent() (err error) {
 			extID = int(eh.key.ExtentId)
 		}
 
-		if conn, err = StreamConnPool.GetConnect(dp.Hosts[0]); err != nil {
+		if conn, err = StreamWriteConnPool.GetConnect(dp.Hosts[0]); err != nil {
 			log.LogWarnf("allocateExtent: failed to create connection, eh(%v) err(%v) dp(%v) exclude(%v)",
 				eh, err, dp, exclude)
 			// If storeMode is tinyExtentType and can't create connection, we also check host status.
@@ -766,13 +768,13 @@ func (eh *ExtentHandler) createExtent(dp *wrapper.DataPartition) (extID int, err
 		stat.EndStat("createExtent", err, bgTime, 1)
 	}()
 
-	conn, err := StreamConnPool.GetConnect(dp.Hosts[0])
+	conn, err := StreamWriteConnPool.GetConnect(dp.Hosts[0])
 	if err != nil {
 		return extID, errors.Trace(err, "createExtent: failed to create connection, eh(%v) datapartionHosts(%v)", eh, dp.Hosts[0])
 	}
 
 	defer func() {
-		StreamConnPool.PutConnectEx(conn, err)
+		StreamWriteConnPool.PutConnectEx(conn, err)
 	}()
 
 	p := NewCreateExtentPacket(dp, eh.inode)
diff --git a/sdk/data/stream/stream_conn.go b/sdk/data/stream/stream_conn.go
index 9ef58a769..5d37441aa 100644
--- a/sdk/data/stream/stream_conn.go
+++ b/sdk/data/stream/stream_conn.go
@@ -53,7 +53,10 @@ type StreamConn struct {
 	maxRetryTimeout time.Duration
 }
 
-var StreamConnPool = util.NewConnectPool()
+var (
+	StreamConnPool      = util.NewConnectPool()
+	StreamWriteConnPool = util.NewConnectPool()
+)
 
 // NewStreamConn returns a new stream connection.
 func NewStreamConn(dp *wrapper.DataPartition, follower bool, timeout time.Duration) (sc *StreamConn) {
@@ -157,7 +160,7 @@ func (sc *StreamConn) sendToDataPartition(req *Packet, retry *bool, getReply Get
 		log.LogDebugf("req opcode %v, conn %v", req.Opcode, conn)
 		err = sc.sendToConn(conn, req, getReply)
 		if err == nil {
-			StreamConnPool.PutConnect(conn, false)
+			StreamConnPool.PutConnectV2(conn, false, sc.currAddr)
 			return
 		}
 		log.LogWarnf("sendToDataPartition: send to curr addr failed, addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
@@ -182,7 +185,7 @@ func (sc *StreamConn) sendToDataPartition(req *Packet, retry *bool, getReply Get
 		sc.dp.LeaderAddr = addr
 		err = sc.sendToConn(conn, req, getReply)
 		if err == nil {
-			StreamConnPool.PutConnect(conn, false)
+			StreamConnPool.PutConnectV2(conn, false, addr)
 			return
 		}
 		StreamConnPool.PutConnect(conn, true)
@@ -242,6 +245,8 @@ func sortByStatus(dp *wrapper.DataPartition, selectAll bool) (hosts []string) {
 		dpHosts = dp.Hosts
 	}
 
+	hosts = make([]string, 0, len(dpHosts))
+
 	for _, addr := range dpHosts {
 		status, ok := hostsStatus[addr]
 		if ok {
diff --git a/sdk/data/stream/stream_reader.go b/sdk/data/stream/stream_reader.go
index a46b52b31..fc07a4f27 100644
--- a/sdk/data/stream/stream_reader.go
+++ b/sdk/data/stream/stream_reader.go
@@ -56,6 +56,8 @@ type Streamer struct {
 	needUpdateVer        int32
 	isCache              bool
 	openForWrite         bool
+
+	rdonly bool
 }
 
 type bcacheKey struct {
@@ -71,7 +73,7 @@ func NewStreamer(client *ExtentClient, inode uint64, openForWrite, isCache bool)
 	s.inode = inode
 	s.parentInode = 0
 	s.extents = NewExtentCache(inode)
-	s.request = make(chan interface{}, 64)
+	s.request = make(chan interface{}, reqChanSize)
 	s.done = make(chan struct{})
 	s.dirtylist = NewDirtyExtentList()
 	s.isOpen = true
@@ -153,7 +155,9 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
 	)
 	log.LogDebugf("action[streamer.read] ino(%v) offset %v size %v", s.inode, offset, size)
 	ctx := context.Background()
-	s.client.readLimiter.Wait(ctx)
+	if s.client.readLimit() {
+		s.client.readLimiter.Wait(ctx)
+	}
 	s.client.LimitManager.ReadAlloc(ctx, size)
 	requests = s.extents.PrepareReadRequests(offset, size, data)
 	for _, req := range requests {
@@ -208,8 +212,8 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
 			// skip hole,ek is not nil,read block cache firstly
 			log.LogDebugf("Stream read: ino(%v) req(%v) s.client.bcacheEnable(%v) s.client.bcacheOnlyForNotSSD(%v) s.needBCache(%v)",
 				s.inode, req, s.client.bcacheEnable, s.client.bcacheOnlyForNotSSD, s.needBCache)
-			cacheKey := util.GenerateRepVolKey(s.client.volumeName, s.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, req.ExtentKey.FileOffset)
 			if s.client.bcacheEnable && s.needBCache && filesize <= bcache.MaxFileSize {
+				cacheKey := util.GenerateRepVolKey(s.client.volumeName, s.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, req.ExtentKey.FileOffset)
 				inodeInfo, err := s.client.getInodeInfo(s.inode)
 				if err != nil {
 					log.LogErrorf("Streamer read: getInodeInfo failed. ino(%v) req(%v) err(%v)", s.inode, req, err)
@@ -229,6 +233,7 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
 								s.inode, inodeInfo.StorageClass, cacheKey, readBytes, err)
 							continue
 						}
+						log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) loadBcache(%v)", cacheKey, s.client.loadBcache)
 					}
 					log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) loadBcache(%v)", cacheKey, s.client.loadBcache)
 				} else {
@@ -255,6 +260,7 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
 					log.LogErrorf("Streamer read: getInodeInfo failed. ino(%v) req(%v) err(%v)", s.inode, req, err)
 					return 0, err
 				}
+				cacheKey := util.GenerateRepVolKey(s.client.volumeName, s.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, req.ExtentKey.FileOffset)
 				// limit big block cache
 				if s.exceedBlockSize(req.ExtentKey.Size) && atomic.LoadInt32(&s.client.inflightL1BigBlock) > 10 {
 					// do nothing
@@ -269,9 +275,7 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
 				}
 			}
 
-			beg := time.Now()
 			readBytes, err = reader.Read(req)
-			clientMetric.WithLabelValues("Streamer_read_Read").Observe(float64(time.Since(beg).Microseconds()))
 			log.LogDebugf("TRACE Stream read: ino(%v) req(%v) readBytes(%v) err(%v)", s.inode, req, readBytes, err)
 
 			total += readBytes
diff --git a/sdk/data/stream/stream_writer.go b/sdk/data/stream/stream_writer.go
index 4d43d99b0..47033a0ba 100644
--- a/sdk/data/stream/stream_writer.go
+++ b/sdk/data/stream/stream_writer.go
@@ -139,6 +139,9 @@ func (s *Streamer) IssueWriteRequest(offset int, data []byte, flags int, checkFu
 }
 
 func (s *Streamer) IssueFlushRequest() error {
+	if s.rdonly {
+		return nil
+	}
 	request := flushRequestPool.Get().(*FlushRequest)
 	request.done = make(chan struct{}, 1)
 	s.request <- request
@@ -214,6 +217,10 @@ func (s *Streamer) server() {
 			log.LogDebugf("done server: evict, streamer(%v)", s)
 			return
 		case <-t.C:
+			if s.rdonly {
+				log.LogDebugf("server: rdonly stream no need to start server routine. ino %d", s.inode)
+				return
+			}
 			s.traverse()
 			s.client.streamerLock.Lock()
 			if s.refcnt <= 0 {
diff --git a/sdk/meta/operation.go b/sdk/meta/operation.go
index 829841d3c..aa85f0e97 100644
--- a/sdk/meta/operation.go
+++ b/sdk/meta/operation.go
@@ -2896,14 +2896,6 @@ func (mw *MetaWrapper) lockDir(mp *MetaPartition, inode uint64, lease uint64, lo
 		stat.EndStat("lockDir", err, bgTime, 1)
 	}()
 
-	req := &proto.LockDirRequest{
-		VolName:     mw.volname,
-		PartitionId: mp.PartitionID,
-		Inode:       inode,
-		LockId:      lockId,
-		Lease:       lease,
-	}
-
 	if lockId == 0 && lease != 0 {
 		status, uniqId, err1 := mw.consumeUniqID(mp)
 		if err1 != nil || status != statusOK {
@@ -2915,6 +2907,14 @@ func (mw *MetaWrapper) lockDir(mp *MetaPartition, inode uint64, lease uint64, lo
 		log.LogDebugf("lockDir: get lockId success, id %d, ino %d", lockId, inode)
 	}
 
+	req := &proto.LockDirRequest{
+		VolName:     mw.volname,
+		PartitionId: mp.PartitionID,
+		Inode:       inode,
+		LockId:      lockId,
+		Lease:       lease,
+	}
+
 	packet := proto.NewPacketReqID()
 	packet.Opcode = proto.OpMetaLockDir
 	packet.PartitionID = mp.PartitionID
diff --git a/storage/extent.go b/storage/extent.go
index 137bcf502..dd101bd36 100644
--- a/storage/extent.go
+++ b/storage/extent.go
@@ -27,7 +27,9 @@ import (
 	"sync/atomic"
 	"syscall"
 	"time"
+	"unsafe"
 
+	"github.com/cubefs/cubefs/blobstore/util/bytespool"
 	"github.com/cubefs/cubefs/depends/tiglabs/raft/logger"
 	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/util"
@@ -44,8 +46,33 @@ const (
 
 const (
 	ExtentMaxSize = 1024 * 1024 * 1024 * 1024 * 4 // 4TB
+	pageSize      = 4096
+	alignSize     = 4096
 )
 
+func alignment(block []byte, AlignSize int) int {
+	return int(uintptr(unsafe.Pointer(&block[0])) & uintptr(AlignSize-1))
+}
+
+// alignedBlock returns []byte of size BlockSize aligned to a multiple
+// of AlignSize in memory (must be power of two)
+func alignedBlock(blkSize int, block []byte) []byte {
+	a := alignment(block, alignSize)
+	offset := 0
+	if a != 0 {
+		offset = alignSize - a
+	}
+	block = block[offset : offset+blkSize]
+	// Can't check alignment of a zero sized block
+	if blkSize != 0 {
+		a = alignment(block, alignSize)
+		if a != 0 {
+			log.LogFatal("Failed to align block")
+		}
+	}
+	return block
+}
+
 type WriteParam struct {
 	ExtentID                                uint64
 	Offset                                  int64
@@ -183,6 +210,7 @@ func (extInfos SortedExtentInfos) Swap(i, j int) {
 // Header of extent include inode value of this extent block and Crc blocks of data blocks.
 type Extent struct {
 	file            *os.File
+	readFile        *os.File
 	filePath        string
 	extentID        uint64
 	modifyTime      int64
@@ -232,6 +260,9 @@ func (e *Extent) Close() (err error) {
 	if err = e.file.Close(); err != nil {
 		return
 	}
+	if err = e.readFile.Close(); err != nil {
+		return
+	}
 	return
 }
 
@@ -254,6 +285,10 @@ func (e *Extent) InitToFS() (err error) {
 		return err
 	}
 
+	if e.readFile, err = os.OpenFile(e.filePath, os.O_RDONLY|syscall.O_DIRECT, 0o666); err != nil {
+		return err
+	}
+
 	if IsTinyExtent(e.extentID) {
 		e.dataSize = 0
 		return
@@ -308,6 +343,15 @@ func (e *Extent) RestoreFromFS() (err error) {
 		}
 		return err
 	}
+
+	if e.readFile, err = os.OpenFile(e.filePath, os.O_RDONLY|syscall.O_DIRECT, 0o666); err != nil {
+		if os.IsNotExist(err) {
+			err = ExtentNotFoundError
+			log.LogWarnf("RestoreFromFS: open file by direct failed, file %s, err %s", e.filePath, err.Error())
+		}
+		return err
+	}
+
 	var info os.FileInfo
 	if info, err = e.file.Stat(); err != nil {
 		err = fmt.Errorf("stat file %v: %v", e.file.Name(), err)
@@ -415,7 +459,7 @@ func (e *Extent) Write(param *WriteParam, crcFunc UpdateCrcFunc) (status uint8,
 	if err = e.checkWriteOffsetAndSize(param); err != nil {
 		log.LogErrorf("action[Extent.Write] checkWriteOffsetAndSize write param(%v) err %v", param, err)
 		err = newParameterError("extent current size=%d write param(%v)", e.dataSize, param)
-		log.LogInfof("action[Extent.Write] newParameterError path %v write param(%v) err %v", e.filePath, param, err)
+		log.LogErrorf("action[Extent.Write] newParameterError path %v write param(%v) err %v", e.filePath, param, err)
 		status = proto.OpTryOtherExtent
 		return
 	}
@@ -515,7 +559,9 @@ func (e *Extent) Read(data []byte, offset, size int64, isRepairRead bool) (crc u
 	}
 
 	var rSize int
-	if rSize, err = e.file.ReadAt(data[:size], offset); err != nil {
+	if size < util.BlockSize {
+		err = e.ReadAligned(data, offset, size)
+	} else if rSize, err = e.file.ReadAt(data[:size], offset); err != nil {
 		log.LogErrorf("action[Extent.Read]extent %v offset %v size %v err %v realsize %v", e.extentID, offset, size, err, rSize)
 		return
 	}
@@ -523,6 +569,33 @@ func (e *Extent) Read(data []byte, offset, size int64, isRepairRead bool) (crc u
 	return
 }
 
+func (e *Extent) ReadAligned(data []byte, offset, size int64) error {
+	start := offset / pageSize * pageSize
+	end := (offset + size + pageSize - 1) / pageSize * pageSize
+
+	newSize := end - start
+
+	block := bytespool.Alloc(int(newSize) + alignSize)
+	defer bytespool.Free(block)
+
+	newData := alignedBlock(int(newSize), block)
+
+	n, err := e.readFile.ReadAt(newData, start)
+	if err != nil && err != io.EOF {
+		return err
+	}
+
+	newEnd := offset - start + size
+	if n < int(newEnd) {
+		return fmt.Errorf("read data size %d less than req, off %d, start %d, size %d",
+			n, offset, start, size)
+	}
+
+	copy(data, newData[offset-start:offset-start+size])
+
+	return nil
+}
+
 // ReadTiny read data from a tiny extent.
 func (e *Extent) ReadTiny(data []byte, offset, size int64, isRepairRead bool) (crc uint32, err error) {
 	_, err = e.file.ReadAt(data[:size], offset)
diff --git a/storage/extent_cache.go b/storage/extent_cache.go
index 134241adc..75cf8f085 100644
--- a/storage/extent_cache.go
+++ b/storage/extent_cache.go
@@ -22,6 +22,10 @@ import (
 	"github.com/cubefs/cubefs/util/log"
 )
 
+const (
+	minCacheCap = 100
+)
+
 // ExtentMapItem stores the extent entity pointer and the element
 // pointer of the extent entity in a cache list.
 type ExtentMapItem struct {
@@ -41,6 +45,10 @@ type ExtentCache struct {
 
 // NewExtentCache creates and returns a new ExtentCache instance.
 func NewExtentCache(capacity int) *ExtentCache {
+	if capacity <= 0 {
+		capacity = minCacheCap
+	}
+
 	return &ExtentCache{
 		extentMap:   make(map[uint64]*ExtentMapItem),
 		extentList:  list.New(),
diff --git a/storage/extent_store.go b/storage/extent_store.go
index 520f1b16d..9063c7492 100644
--- a/storage/extent_store.go
+++ b/storage/extent_store.go
@@ -156,7 +156,7 @@ func MkdirAll(name string) (err error) {
 	return os.MkdirAll(name, 0o755)
 }
 
-func NewExtentStore(dataDir string, partitionID uint64, storeSize, dpType int, isCreate bool) (s *ExtentStore, err error) {
+func NewExtentStore(dataDir string, partitionID uint64, storeSize, dpType, cap int, isCreate bool) (s *ExtentStore, err error) {
 	begin := time.Now()
 	defer func() {
 		log.LogInfof("[NewExtentStore] load dp(%v) new extent store using time(%v)", partitionID, time.Since(begin))
@@ -234,7 +234,7 @@ func NewExtentStore(dataDir string, partitionID uint64, storeSize, dpType int, i
 
 	s.extentInfoMap = make(map[uint64]*ExtentInfo)
 	s.extentLockMap = make(map[uint64]proto.GcFlag)
-	s.cache = NewExtentCache(100)
+	s.cache = NewExtentCache(cap)
 	if err = s.initBaseFileID(); err != nil {
 		err = fmt.Errorf("init base field ID: %v", err)
 		return
@@ -737,8 +737,10 @@ func (s *ExtentStore) Read(extentID uint64, offset, size int64, nbuf []byte, isR
 	log.LogDebugf("[Read] dp %v extent[%d] offset[%d] size[%d] isRepairRead[%v] extentLock[%v]",
 		s.partitionID, extentID, offset, size, isRepairRead, s.extentLock)
 	defer func() {
-		log.LogDebugf("[Read] dp %v extent[%d] offset[%d] size[%d] isRepairRead[%v] extentLock[%v] cost %v",
-			s.partitionID, extentID, offset, size, isRepairRead, s.extentLock, time.Since(begin).String())
+		if log.EnableDebug() {
+			log.LogDebugf("[Read] dp %v extent[%d] offset[%d] size[%d] isRepairRead[%v] extentLock[%v] cost %v",
+				s.partitionID, extentID, offset, size, isRepairRead, s.extentLock, time.Since(begin).String())
+		}
 	}()
 
 	ei, _ := s.GetExtentInfo(extentID)
@@ -780,8 +782,10 @@ func (s *ExtentStore) Read(extentID uint64, offset, size int64, nbuf []byte, isR
 	log.LogDebugf("[Read]dp %v extent %v offset %v size %v  ei.Size %v e.dataSize %v isRepairRead %v",
 		s.partitionID, extentID, offset, size, ei.Size, e.dataSize, isRepairRead)
 	crc, err = e.Read(nbuf, offset, size, isRepairRead)
-	log.LogDebugf("[Read]dp %v extent %v offset %v size %v  ei.Size %v e.dataSize %v isRepairRead %v,cost %v",
-		s.partitionID, extentID, offset, size, ei.Size, e.dataSize, isRepairRead, time.Since(begin2).String())
+	if log.EnableDebug() {
+		log.LogDebugf("[Read]dp %v extent %v offset %v size %v  ei.Size %v e.dataSize %v isRepairRead %v,cost %v",
+			s.partitionID, extentID, offset, size, ei.Size, e.dataSize, isRepairRead, time.Since(begin2).String())
+	}
 
 	return
 }
diff --git a/storage/extent_store_test.go b/storage/extent_store_test.go
index a59df6770..572d5a65d 100644
--- a/storage/extent_store_test.go
+++ b/storage/extent_store_test.go
@@ -171,7 +171,7 @@ func reopenExtentStoreTest(t *testing.T, dpType int) {
 	path, clean, err := getTestPathExtentStore()
 	require.NoError(t, err)
 	defer clean()
-	s, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, true)
+	s, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, 0, true)
 	require.NoError(t, err)
 	defer s.Close()
 	id, err := s.NextExtentID()
@@ -198,7 +198,7 @@ func reopenExtentStoreTest(t *testing.T, dpType int) {
 	firstSnap, err := s.SnapShot()
 	require.NoError(t, err)
 	s.Close()
-	newStor, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, false)
+	newStor, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, 0, false)
 	require.NoError(t, err)
 	defer newStor.Close()
 	// read data
@@ -230,7 +230,7 @@ func staleExtentStoreTest(t *testing.T, dpType int) {
 	extDirName := filepath.Base(path)
 	require.NoError(t, err)
 	defer clean()
-	s, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, true)
+	s, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, 0, true)
 	require.NoError(t, err)
 	id, err := s.NextExtentID()
 	require.NoError(t, err)
@@ -239,7 +239,7 @@ func staleExtentStoreTest(t *testing.T, dpType int) {
 	s.Close()
 
 	// reopen1
-	newS1, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, true)
+	newS1, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, 0, true)
 	require.NoError(t, err)
 	fileList, err := os.ReadDir(filepath.Dir(path))
 	require.NoError(t, err)
@@ -255,7 +255,7 @@ func staleExtentStoreTest(t *testing.T, dpType int) {
 	newS1.Close()
 
 	// reopen2
-	newS2, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, true)
+	newS2, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, 0, true)
 	require.NoError(t, err)
 	fileList, err = os.ReadDir(filepath.Dir(path))
 	require.NoError(t, err)
@@ -276,7 +276,7 @@ func ExtentStoreTest(t *testing.T, dpType int) {
 	path, clean, err := getTestPathExtentStore()
 	require.NoError(t, err)
 	defer clean()
-	s, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, true)
+	s, err := storage.NewExtentStore(path, 0, 1*util.GB, dpType, 0, true)
 	require.NoError(t, err)
 	defer s.Close()
 	extentStoreLogicalTest(t, s)
diff --git a/util/config/config.go b/util/config/config.go
index 492a57e28..87313ca9b 100644
--- a/util/config/config.go
+++ b/util/config/config.go
@@ -43,6 +43,10 @@ func newConfig() *Config {
 	return result
 }
 
+func NewConfig() *Config {
+	return newConfig()
+}
+
 // LoadConfigFile loads config information from a JSON file.
 func LoadConfigFile(filename string) (*Config, error) {
 	result := newConfig()
@@ -98,6 +102,10 @@ func (c *Config) SetString(key, val string) {
 	c.data[key] = val
 }
 
+func (c *Config) SetNewVal(key string, val interface{}) {
+	c.data[key] = val
+}
+
 // GetFloat returns a float value for the config key.
 func (c *Config) GetFloat(key string) float64 {
 	x, present := c.data[key]
diff --git a/util/conn_pool.go b/util/conn_pool.go
index 1f517270a..063be2323 100644
--- a/util/conn_pool.go
+++ b/util/conn_pool.go
@@ -115,6 +115,10 @@ func (cp *ConnectPool) ReleaseAll(addr net.Addr) {
 }
 
 func (cp *ConnectPool) PutConnect(c *net.TCPConn, forceClose bool) {
+	cp.PutConnectV2(c, forceClose, "")
+}
+
+func (cp *ConnectPool) PutConnectV2(c *net.TCPConn, forceClose bool, addr string) {
 	if c == nil {
 		return
 	}
@@ -128,7 +132,9 @@ func (cp *ConnectPool) PutConnect(c *net.TCPConn, forceClose bool) {
 		return
 	default:
 	}
-	addr := c.RemoteAddr().String()
+	if addr == "" {
+		addr = c.RemoteAddr().String()
+	}
 	cp.RLock()
 	pool, ok := cp.pools[addr]
 	cp.RUnlock()
diff --git a/util/exporter/exporter.go b/util/exporter/exporter.go
index dceed0a3a..b8873d5c0 100644
--- a/util/exporter/exporter.go
+++ b/util/exporter/exporter.go
@@ -23,6 +23,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/cubefs/cubefs/proto"
@@ -290,4 +291,42 @@ func collect() {
 	go collectGauge()
 	go collectHistogram()
 	go collectAlarm()
+	prometheus.Register(Recoder)
+}
+
+var Recoder = prometheus.NewHistogramVec(
+	prometheus.HistogramOpts{
+		Name:    "request_const_us",
+		Help:    "recode cost time by us",
+		Buckets: []float64{50, 100, 200, 500, 1000, 2000, 5000, 10000, 200000, 500000},
+	},
+	[]string{"api"},
+)
+
+var (
+	obMap = map[string]prometheus.Observer{}
+	obLk  = sync.RWMutex{}
+)
+
+func RecodCost(api string, costUs int64) {
+	obLk.RLock()
+	ob := obMap[api]
+	obLk.RUnlock()
+	if ob != nil {
+		ob.Observe(float64(costUs))
+		return
+	}
+
+	obLk.Lock()
+	defer obLk.Unlock()
+
+	ob = obMap[api]
+	if ob != nil {
+		ob.Observe(float64(costUs))
+		return
+	}
+
+	ob = Recoder.WithLabelValues(api)
+	obMap[api] = ob
+	ob.Observe(float64(costUs))
 }
diff --git a/util/stat/statistic.go b/util/stat/statistic.go
index e2c561f17..8d226307f 100644
--- a/util/stat/statistic.go
+++ b/util/stat/statistic.go
@@ -394,6 +394,9 @@ func shiftFiles() error {
 }
 
 func StatBandWidth(typeName string, Size uint32) {
+	if gSt == nil {
+		return
+	}
 	EndStat(typeName+"[FLOW_KB]", nil, nil, Size/1024)
 }