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

fix(flashnode): use limitWrite for flashnode instead of disk

Browse Source 
close:#23123752

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-03-10 19:12:33 +08:00 committed by zhumingze1108
parent be79605ea0
commit 9a1adf8212
8 changed files with 61 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
flashnode/cachengine/block.go
View File

@ -29,7 +29,6 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
)
@ -106,7 +105,7 @@ func (cb *CacheBlock) Delete(reason string) (err error) {
if cb.Exist() {
err = os.Remove(cb.filePath)
auditlog.LogFlashNodeOp("BlockDelete", fmt.Sprintf("delete block %v, by :%v",
cb.filePath, reason), err)
cb.info(), reason), err)
}
return
}
@ -166,21 +165,14 @@ func (cb *CacheBlock) WriteAt(data []byte, offset, size int64) (err error) {
log.LogWarnf("[WriteAt] GetOrOpenFileHandler (%v) err %v", cb.filePath, err)
return
}
if writable := cb.disk.LimitWrite.TryRun(int(size), func() {
bgTime2 := stat.BeginStat()
if _, err = file.WriteAt(data[:size], offset+HeaderSize); err != nil {
log.LogWarnf("[WriteAt] WriteAt (%v) err %v", cb.filePath, err)
stat.EndStat("CacheBlock:PersistToLocal", err, bgTime2, 1)
return
}
stat.EndStat("CacheBlock:PersistToLocal", nil, bgTime2, 1)
}); !writable {
err = errors.NewErrorf("write io limitd")
return
}
if err != nil {
bgTime2 := stat.BeginStat()
if _, err = file.WriteAt(data[:size], offset+HeaderSize); err != nil {
log.LogWarnf("[WriteAt] WriteAt (%v) err %v", cb.filePath, err)
stat.EndStat("CacheBlock:PersistToLocal", err, bgTime2, 1)
return
}
stat.EndStat("CacheBlock:PersistToLocal", nil, bgTime2, 1)
cb.maybeUpdateUsedSize(offset + size)
return
}
@ -356,7 +348,7 @@ func (cb *CacheBlock) initFilePath(isLoad bool) (err error) {
}
_, err = os.Stat(cb.filePath)
if !isLoad {
msg := fmt.Sprintf("init cache block(%s) to local: err %v", cb.filePath, err)
msg := fmt.Sprintf("init cache block(%s) to local : err %v", cb.info(), err)
log.LogDebugf("%v", msg)
auditlog.LogFlashNodeOp("BlockInit", msg, err)
}
@ -515,7 +507,7 @@ func (cb *CacheBlock) InitOnce(engine *CacheEngine, sources []*proto.DataSource)
select {
case <-cb.closeCh:
engine.deleteCacheBlock(cb.blockKey)
auditlog.LogFlashNodeOp("BlockInit", fmt.Sprintf("%v is closed", cb.blockKey), nil)
auditlog.LogFlashNodeOp("BlockInit", fmt.Sprintf("%v is closed", cb.info()), nil)
default:
}
}
@ -553,7 +545,7 @@ func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto
select {
case <-cb.closeCh:
engine.deleteCacheBlock(cb.blockKey)
auditlog.LogFlashNodeOp("BlockInit", fmt.Sprintf("%v is closed", cb.blockKey), nil)
auditlog.LogFlashNodeOp("BlockInit", fmt.Sprintf("%v is closed", cb.info()), nil)
default:
}
})
@ -621,3 +613,7 @@ func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNode
}
cb.notifyReady()
}
func (cb *CacheBlock) info() string {
return fmt.Sprintf("path(%v)_from(%v)", cb.filePath, cb.clientIP)
}

8
flashnode/cachengine/block_test.go
View File

@ -73,9 +73,7 @@ func TestBlockWriteCache(t *testing.T) {
func testWriteSingleFile(t *testing.T) {
disk := new(Disk)
disk.Path = testTmpFS
disk.LimitWrite = util.NewIOLimiter(DefaultDiskWriteFlow, DefaultDiskWriteIOCC)
disk.Status = proto.ReadWrite
cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE,
nil, "", disk)
cacheBlock.cacheEngine = &CacheEngine{}
@ -98,7 +96,6 @@ func testWriteSingleFile(t *testing.T) {
func testWriteSingleFileError(t *testing.T) {
disk := new(Disk)
disk.Path = testTmpFS
disk.LimitWrite = util.NewIOLimiter(DefaultDiskWriteFlow, DefaultDiskWriteIOCC)
disk.Status = proto.ReadWrite
cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE,
nil, "", disk)
@ -124,7 +121,6 @@ func testWriteCacheBlockFull(t *testing.T) {
var err error
disk := new(Disk)
disk.Path = testTmpFS
disk.LimitWrite = util.NewIOLimiter(DefaultDiskWriteFlow, DefaultDiskWriteIOCC)
disk.Status = proto.ReadWrite
cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE,
nil, "", disk)
@ -157,7 +153,6 @@ func testWriteCacheBlockFull(t *testing.T) {
func newCacheBlockWithDiffInode(volume string, index int, allocSize uint64) (cacheBlock *CacheBlock, err error) {
disk := new(Disk)
disk.Path = testTmpFS
disk.LimitWrite = util.NewIOLimiter(DefaultDiskWriteFlow, DefaultDiskWriteIOCC)
disk.Status = proto.ReadWrite
cacheBlock = NewCacheBlock(testTmpFS, volume, uint64(index), 1024, 112456871, allocSize,
nil, "", disk)
@ -178,7 +173,6 @@ func newCacheBlockWithDiffInode(volume string, index int, allocSize uint64) (cac
func newCacheBlockWithDiffVolume(volume string, index int, allocSize uint64) (cacheBlock *CacheBlock, err error) {
disk := new(Disk)
disk.Path = testTmpFS
disk.LimitWrite = util.NewIOLimiter(DefaultDiskWriteFlow, DefaultDiskWriteIOCC)
disk.Status = proto.ReadWrite
newVolume := fmt.Sprintf("%s_%d", volume, index)
cacheBlock = NewCacheBlock(testTmpFS, newVolume, 1, 1024, 112456871, allocSize,
@ -233,7 +227,6 @@ func TestBlockReadCache(t *testing.T) {
defer func() { require.NoError(t, umount()) }()
disk := new(Disk)
disk.Path = testTmpFS
disk.LimitWrite = util.NewIOLimiter(DefaultDiskWriteFlow, DefaultDiskWriteIOCC)
disk.Status = proto.ReadWrite
cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 2568748711, proto.CACHE_BLOCK_SIZE,
nil, "", disk)
@ -274,7 +267,6 @@ func testParallelOperation(t *testing.T) {
defer func() { require.NoError(t, umount()) }()
disk := new(Disk)
disk.Path = testTmpFS
disk.LimitWrite = util.NewIOLimiter(DefaultDiskWriteFlow, DefaultDiskWriteIOCC)
disk.Status = proto.ReadWrite
cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE,
nil, "", disk)

16
flashnode/cachengine/engine.go
View File

@ -31,7 +31,6 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/errors"
@ -63,7 +62,6 @@ type Disk struct {
Path string
TotalSpace int64 // actual disk space configured for caching
Capacity int // lru capacity
LimitWrite *util.IoLimiter
Status int32
}
@ -653,7 +651,7 @@ func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64,
v, ok = c.keyToDiskMap.Load(key)
if ok {
cacheItem := v.(*lruCacheItem)
if atomic.LoadInt32(&cacheItem.status) == proto.ReadWrite {
if atomic.LoadInt32(&cacheItem.disk.Status) == proto.ReadWrite {
if blockValue, got := cacheItem.lruCache.Peek(key); got {
block = blockValue.(*CacheBlock)
return
@ -758,6 +756,8 @@ func (c *CacheEngine) CreateBlock(req *proto.CacheRequest, clientIP string, isPr
return nil, fmt.Errorf("no source data")
}
if block, err = c.createCacheBlock(req.Volume, req.Inode, req.FixedFileOffset, req.Version, req.TTL, computeAllocSize(req.Sources), clientIP, isPrepare); err != nil {
log.LogWarnf("action[CreateBlock] createCacheBlock(%v) failed err %v ",
GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version), err)
c.deleteCacheBlock(GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version))
return nil, err
}
@ -1006,13 +1006,3 @@ func (c *CacheEngine) SetReadDataNodeTimeout(timeout int) {
func (c *CacheEngine) GetReadDataNodeTimeout() int {
return c.readDataNodeTimeout
}
func (d *Disk) UpdateQosLimiter(iocc, flow, factor int) {
if atomic.LoadInt32(&d.Status) == proto.Unavailable {
log.LogWarnf("[updateQosLimiter] disk(%v) is broken", d.Path)
return
}
log.LogInfof("action[updateQosLimiter] disk %v write(iocc:%v flow:%v)", d.Path, iocc, flow)
d.LimitWrite.ResetIO(iocc, factor)
d.LimitWrite.ResetFlow(flow)
}

4
flashnode/cachengine/engine_test.go
View File

@ -52,7 +52,6 @@ func TestEngineNew(t *testing.T) {
}
disk := &Disk{Path: testTmpFS, TotalSpace: 200 * util.MB, Capacity: 1024, Status: proto.ReadWrite}
disk.LimitWrite = util.NewIOLimiter(0, 8)
disks := make([]*Disk, 0)
disks = append(disks, disk)
if !enabledTmpfs() {
@ -80,7 +79,6 @@ func TestEngineOverFlow(t *testing.T) {
}
disk := &Disk{Path: testTmpFS, TotalSpace: 200 * util.MB, Capacity: 1024, Status: proto.ReadWrite}
disk.LimitWrite = util.NewIOLimiter(0, 8)
disks := make([]*Disk, 0)
disks = append(disks, disk)
if !enabledTmpfs() {
@ -154,7 +152,6 @@ func TestEngineTTL(t *testing.T) {
require.NoError(t, err)
}
disk := &Disk{Path: testTmpFS, TotalSpace: 200 * util.MB, Capacity: 1024, Status: proto.ReadWrite}
disk.LimitWrite = util.NewIOLimiter(0, 8)
disks := make([]*Disk, 0)
disks = append(disks, disk)
if !enabledTmpfs() {
@ -217,7 +214,6 @@ func TestEngineLru(t *testing.T) {
}
disk := &Disk{Path: testTmpFS, TotalSpace: 200 * util.MB, Capacity: 1024, Status: proto.ReadWrite}
disk.LimitWrite = util.NewIOLimiter(0, 8)
disks := make([]*Disk, 0)
disks = append(disks, disk)
if !enabledTmpfs() {

20
flashnode/flashnode.go
View File

@ -56,9 +56,9 @@ const (
_connPoolIdleTimeout = 60 // 60s
_extentReadMaxRetry = 3
_extentReadTimeoutSec = 3
_defaultDiskWriteIOCC = 8
_defaultDiskWriteIOCC = 64
_defaultDiskWriteFlow = 0 * util.GB
_defaultDiskWriteFactor = 0
_defaultDiskWriteFactor = 8
)
// Configuration keys
@ -122,6 +122,8 @@ type FlashNode struct {
diskWriteIocc int
diskWriteFlow int
diskWriteIoFactorFlow int
limitWrite *util.IoLimiter
}
// Start starts up the flash node with the specified configuration.
@ -230,8 +232,8 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) {
f.diskWriteFlow = _defaultDiskWriteFlow
}
f.diskWriteIoFactorFlow = cfg.GetInt(cfgDiskWriteIoFactor)
if f.diskWriteFlow <= 0 {
f.diskWriteFlow = _defaultDiskWriteFactor
if f.diskWriteIoFactorFlow <= 0 {
f.diskWriteIoFactorFlow = _defaultDiskWriteFactor
}
if percent <= 1e-2 || percent > 1.0 {
percent = 1.0
@ -264,7 +266,6 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) {
disk := new(cachengine.Disk)
disk.TotalSpace = int64(f.memTotal)
disk.Path = f.memDataPath
disk.LimitWrite = util.NewIOLimiter(f.diskWriteFlow, f.diskWriteIocc)
disk.Status = proto.ReadWrite
disks := make([]*cachengine.Disk, 0)
disks = append(disks, disk)
@ -307,7 +308,6 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) {
disk := new(cachengine.Disk)
disk.TotalSpace = totalSpace
disk.Path = path
disk.LimitWrite = util.NewIOLimiter(f.diskWriteFlow, f.diskWriteIocc)
disk.Status = proto.ReadWrite
disks = append(disks, disk)
}
@ -319,7 +319,7 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) {
}
f.disks = disks
}
f.limitWrite = util.NewIOLimiterEx(f.diskWriteFlow, f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow)
lruFhCapacity := cfg.GetInt(cfgLruFhCapacity)
if lruFhCapacity <= 0 || lruFhCapacity >= 1000000 {
lruFhCapacity = _defaultLRUFhCapacity
@ -424,9 +424,3 @@ func (f *FlashNode) register() error {
}
}
}
func (f *FlashNode) updateQosLimit() {
for _, disk := range f.disks {
disk.UpdateQosLimiter(f.diskWriteIocc, f.diskWriteFlow, f.diskWriteIoFactorFlow)
}
}

23
flashnode/flashnode_op.go
View File

@ -169,12 +169,27 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
errMetric.AddWithLabels(1, map[string]string{exporter.FlashNode: f.localAddr, exporter.Disk: dataPath, exporter.Err: "LowerHitRate"})
}
}
if block, err = f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String(), false); err != nil {
log.LogWarnf("opCacheRead: CreateBlock failed, req(%v) err(%v)", req, err)
if writable := f.limitWrite.TryRun(int(req.Size_), func() {
if block2, err := f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String(), false); err != nil {
log.LogWarnf("opCacheRead: CreateBlock failed, req(%v) err(%v)", req, err)
return
} else {
block2.InitOnceForCacheRead(f.cacheEngine, cr.Sources)
}
}); !writable {
err = errors.NewErrorf("write io is limited")
return
}
select {
case <-ctx.Done():
return ctx.Err()
default:
block, err = f.cacheEngine.GetCacheBlockForRead(volume, cr.Inode, cr.FixedFileOffset, cr.Version, req.Size_)
}
if err != nil {
return err
}
// only execute it once for the first request block is created.
go block.InitOnceForCacheRead(f.cacheEngine, cr.Sources)
}
err = f.doStreamReadRequest(ctx, conn, req, p, block)
return

25
flashnode/serve_http.go
View File

@ -126,23 +126,22 @@ func (f *FlashNode) handleSetDiskQos(w http.ResponseWriter, r *http.Request) {
*pVal = val
}
}
if f.diskWriteIoFactorFlow == 0 {
f.diskWriteIoFactorFlow = _defaultDiskWriteFactor
}
if updated {
f.updateQosLimit()
f.limitWrite.ResetIO(f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow)
f.limitWrite.ResetFlow(f.diskWriteFlow)
}
replyOK(w, r, nil)
}
func (f *FlashNode) handleGetDiskQos(w http.ResponseWriter, r *http.Request) {
disks := make([]interface{}, 0)
for _, diskItem := range f.disks {
disk := &struct {
Path string `json:"path"`
Write util.LimiterStatus `json:"write"`
}{
Path: diskItem.Path,
Write: diskItem.LimitWrite.Status(),
}
disks = append(disks, disk)
}
replyOK(w, r, disks)
status := LimiterStatus{Status: f.limitWrite.Status(), DiskNum: len(f.disks)}
replyOK(w, r, status)
}
type LimiterStatus struct {
Status util.LimiterStatus
DiskNum int
}

8
util/limit_io.go
View File

@ -46,6 +46,7 @@ type LimiterStatus struct {
IOQueue int
IORunning int
IOWaiting int
Factor int
}
var (
@ -57,10 +58,10 @@ var (
// 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)
return NewIOLimiterEx(flowLimit, ioConcurrency, 0)
}
func newIOLimiterEx(flowLimit, ioConcurrency, factor int) *IoLimiter {
func NewIOLimiterEx(flowLimit, ioConcurrency, factor int) *IoLimiter {
flow := rate.NewLimiter(rate.Inf, 0)
if flowLimit > 0 {
flow = rate.NewLimiter(rate.Limit(flowLimit), flowLimit/2)
@ -150,6 +151,7 @@ type ioQueue struct {
stopCh chan struct{}
queue chan *task
midQueue chan *task
factor int
}
func newIOQueue(concurrency, factor int) *ioQueue {
@ -161,6 +163,7 @@ func newIOQueue(concurrency, factor int) *ioQueue {
if factor <= 0 {
factor = defaultQueueFactor
}
q.factor = factor
q.midQueue = make(chan *task, 100)
q.stopCh = make(chan struct{})
q.queue = make(chan *task, factor*concurrency)
@ -278,6 +281,7 @@ func (q *ioQueue) Status() (st LimiterStatus) {
st.IOQueue = cap(q.queue)
st.IORunning = int(atomic.LoadUint32(&q.running))
st.IOWaiting = len(q.queue)
st.Factor = q.factor
return
}