gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor(datanode): refactor data read and write performance. #22504692

Browse Source 
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2024-11-11 16:26:40 +08:00 committed by AmazingChi
parent 6e9caa9131
commit ab8217379e
30 changed files with 393 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
datanode/data_partition_repair.go
View File

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

10
datanode/data_partition_repair_test.go
View File

@ -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

6
datanode/disk.go
View File

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

31
datanode/limit_io.go
View File

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

4
datanode/limit_io_test.go
View File

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

2
datanode/partition.go
View File

@ -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

23
datanode/server.go
View File

@ -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 {

13
datanode/server_handler.go
View File

@ -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 {

11
datanode/wrap_operator.go
View File

@ -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 {

2
datanode/wrap_operator_test.go
View File

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

5
datanode/wrap_prepare.go
View File

@ -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

6
metanode/partition_fsmop_extend.go
View File

@ -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

23
proto/packet.go
View File

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

21
repl/repl_protocol.go
View File

@ -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.

6
sdk/data/manager/limiter.go
View File

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

4
sdk/data/stream/extent_cache.go
View File

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

114
sdk/data/stream/extent_client.go
View File

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

12
sdk/data/stream/extent_handler.go
View File

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

11
sdk/data/stream/stream_conn.go
View File

@ -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 {

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

@ -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

7
sdk/data/stream/stream_writer.go
View File

@ -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 {

16
sdk/meta/operation.go
View File

@ -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

77
storage/extent.go
View File

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

8
storage/extent_cache.go
View File

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

16
storage/extent_store.go
View File

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

12
storage/extent_store_test.go
View File

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

8
util/config/config.go
View File

@ -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]

8
util/conn_pool.go
View File

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

39
util/exporter/exporter.go
View File

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

3
util/stat/statistic.go
View File

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