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): reply to client as soon as possible

Browse Source 
close:#23109728

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-03-19 09:56:09 +08:00 committed by zhumingze1108
parent 9ad005fbdd
commit 380b8dde52
9 changed files with 86 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
datanode/partition.go
View File

@ -727,7 +727,7 @@ func (dp *DataPartition) Stop() {
log.LogInfof("action[Stop]:dp(%v) store applyId %v", dp.info(), applyId)
err := dp.storeAppliedID(applyId)
if err != nil {
log.LogErrorf("action[Stop]: failed to store applied index")
log.LogErrorf("action[Stop]: failed to store applied index dp %v, err %v", dp.info(), err)
dp.checkIsDiskError(err, WriteFlag)
}
})

19
flashnode/cachengine/block.go
View File

@ -37,7 +37,7 @@ import (
const (
_cacheBlockOpenOpt = os.O_CREATE | os.O_RDWR
HeaderSize = 24
HeaderSize = 40
)
type CacheBlock struct {
@ -222,6 +222,14 @@ func (cb *CacheBlock) writeCacheBlockFileHeader(file *os.File) (err error) {
if _, err = file.Seek(0, 0); err != nil {
return
}
// add two reserverd
var reserved uint64 = 0
if err = binary.Write(file, binary.BigEndian, reserved); err != nil {
return
}
if err = binary.Write(file, binary.BigEndian, reserved); err != nil {
return
}
if err = binary.Write(file, binary.BigEndian, cb.getAllocSize()); err != nil {
return
}
@ -248,9 +256,18 @@ func (cb *CacheBlock) writeCacheBlockFileHeader(file *os.File) (err error) {
func (cb *CacheBlock) checkCacheBlockFileHeader(file *os.File) (allocSize, usedSize int64, expiredTime time.Time, err error) {
var stat os.FileInfo
var seconds int64
var reserved1, reserved2 uint64
if stat, err = file.Stat(); err != nil {
return
}
if err = binary.Read(file, binary.BigEndian, &reserved1); err != nil {
return
}
if err = binary.Read(file, binary.BigEndian, &reserved2); err != nil {
return
}
if err = binary.Read(file, binary.BigEndian, &allocSize); err != nil {
return
}

8
flashnode/flashnode.go
View File

@ -55,6 +55,7 @@ const (
_tcpServerTimeoutSec = 60 * 5
_connPoolIdleTimeout = 60 // 60s
_extentReadMaxRetry = 3
_cacheReadTimeoutSec = 1
_extentReadTimeoutSec = 3
_defaultDiskWriteIOCC = 64
_defaultDiskWriteFlow = 0 * util.GB
@ -345,8 +346,9 @@ 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)
f.limitRead = util.NewIOLimiterEx(f.diskReadFlow, f.diskReadIocc*len(f.disks), f.diskReadIoFactorFlow)
f.handleReadTimeout = _cacheReadTimeoutSec
f.limitWrite = util.NewIOLimiterEx(f.diskWriteFlow, f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow, f.handleReadTimeout)
f.limitRead = util.NewIOLimiterEx(f.diskReadFlow, f.diskReadIocc*len(f.disks), f.diskReadIoFactorFlow, f.handleReadTimeout)
lruFhCapacity := cfg.GetInt(cfgLruFhCapacity)
if lruFhCapacity <= 0 || lruFhCapacity >= 1000000 {
lruFhCapacity = _defaultLRUFhCapacity
@ -406,7 +408,7 @@ func (f *FlashNode) startCacheEngine() (err error) {
log.LogErrorf("startCacheEngine failed:%v", err)
return
}
f.SetTimeout(_extentReadTimeoutSec, _extentReadTimeoutSec)
f.SetTimeout(_cacheReadTimeoutSec, _extentReadTimeoutSec)
return f.cacheEngine.Start()
}

28
flashnode/flashnode_op.go
View File

@ -44,16 +44,6 @@ func (f *FlashNode) preHandle(conn net.Conn, p *proto.Packet) error {
}
func (f *FlashNode) handlePacket(conn net.Conn, p *proto.Packet) (err error) {
start := time.Now()
defer func() {
logContent := fmt.Sprintf("client:%v req:[%v](%v) cost %v ", conn.RemoteAddr().String(), p.GetOpMsg(),
p.GetReqID(), time.Since(start).String())
if err != nil {
log.LogErrorf("handlePacket: %v error:%v", logContent, err.Error())
} else {
log.LogDebugf("handlePacket: %v", logContent)
}
}()
switch p.Opcode {
case proto.OpFlashNodeHeartbeat:
err = f.opFlashNodeHeartbeat(conn, p)
@ -72,6 +62,8 @@ func (f *FlashNode) SetTimeout(handleReadTimeout int, readDataNodeTimeout int) {
if f.handleReadTimeout != handleReadTimeout && handleReadTimeout > 0 {
log.LogInfof("FlashNode set handleReadTimeout from %d to %d", f.handleReadTimeout, handleReadTimeout)
f.handleReadTimeout = handleReadTimeout
f.limitWrite.ResetIOEx(f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow, f.handleReadTimeout)
f.limitWrite.ResetFlow(f.diskWriteFlow)
}
f.cacheEngine.SetReadDataNodeTimeout(readDataNodeTimeout)
}
@ -137,7 +129,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
defer func() {
if err != nil {
log.LogErrorf("action[opCacheRead] volume:[%s], logMsg:%s", volume,
log.LogWarnf("action[opCacheRead] volume:[%s], logMsg:%s", volume,
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err))
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
if e := p.WriteToConn(conn); e != nil {
@ -173,7 +165,8 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
}
bgTime2 := stat.BeginStat()
missCacheMetric := exporter.NewTPCnt("MissCacheRead")
if writable := f.limitWrite.TryRun(int(req.Size_), func() {
// try to cache more miss data, but reply to client more quickly
if writable := f.limitWrite.TryRunWithContext(ctx, 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
@ -182,13 +175,13 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
}
}); !writable {
err = fmt.Errorf("create block cache limited")
stat.EndStat("MissCacheRead", err, bgTime2, 1)
stat.EndStat("MissCacheReadLimit", err, bgTime2, 1)
missCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return
}
select {
case <-ctx.Done():
stat.EndStat("MissCacheRead", ctx.Err(), bgTime2, 1)
stat.EndStat("MissCacheReadCancel", ctx.Err(), bgTime2, 1)
missCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return ctx.Err()
default:
@ -200,18 +193,19 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
return err
}
}
bgTime = stat.BeginStat()
bgTime2 := stat.BeginStat()
hitCacheMetric := exporter.NewTPCnt("HitCacheRead")
// reply to client as quick as possible if hit cache
err2 := f.limitRead.RunNoWait(int(req.Size_), false, func() {
err = f.doStreamReadRequest(ctx, conn, req, p, block)
})
if err2 != nil {
err = err2
stat.EndStat("HitCacheRead", err, bgTime, 1)
stat.EndStat("HitCacheRead", err, bgTime2, 1)
hitCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return
}
stat.EndStat("HitCacheRead", err, bgTime, 1)
stat.EndStat("HitCacheRead", err, bgTime2, 1)
hitCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
return
}

11
flashnode/serve_http.go
View File

@ -131,7 +131,7 @@ func (f *FlashNode) handleSetWriteDiskQos(w http.ResponseWriter, r *http.Request
f.diskWriteIoFactorFlow = _defaultDiskWriteFactor
}
if updated {
f.limitWrite.ResetIO(f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow)
f.limitWrite.ResetIOEx(f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow, f.handleReadTimeout)
f.limitWrite.ResetFlow(f.diskWriteFlow)
}
replyOK(w, r, nil)
@ -179,15 +179,16 @@ func (f *FlashNode) handleSetReadDiskQos(w http.ResponseWriter, r *http.Request)
}
func (f *FlashNode) handleGetDiskQos(w http.ResponseWriter, r *http.Request) {
writeStatus := LimiterStatus{Status: f.limitWrite.Status(), DiskNum: len(f.disks)}
readStatus := LimiterStatus{Status: f.limitRead.Status(), DiskNum: len(f.disks)}
writeStatus := LimiterStatus{Status: f.limitWrite.Status(), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
readStatus := LimiterStatus{Status: f.limitRead.Status(), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
info := LimiterStatusInfo{WriteStatus: writeStatus, ReadStatus: readStatus}
replyOK(w, r, info)
}
type LimiterStatus struct {
Status util.LimiterStatus
DiskNum int
Status util.LimiterStatus
DiskNum int
ReadTimeoutSec int
}
type LimiterStatusInfo struct {

2
master/config.go
View File

@ -115,7 +115,7 @@ const (
lowerLimitRWMetaPartition = 3 // lower limit of RW meta partition, equal defaultReplicaNum
defaultHttpReversePoolSize = 1024
defaultFlashNodeHandleReadTimeout = 3
defaultFlashNodeHandleReadTimeout = 1
defaultFlashNodeReadDataNodeTimeout = 3
)

1
sdk/data/stream/stream_remote_cache.go
View File

@ -111,7 +111,6 @@ func (s *Streamer) readFromRemoteCache(ctx context.Context, offset, size uint64,
err = fmt.Errorf("readFromRemoteCache failed: cannot find any flashGroups")
return
}
if read, err = s.client.RemoteCache.Read(ctx, fg, s.inode, req); err != nil {
log.LogWarnf("readFromRemoteCache: flashGroup read failed. offset(%v) size(%v) fg(%v) req(%v) err(%v)", offset, size, fg, req, err)
return

60
util/limit_io.go
View File

@ -51,7 +51,7 @@ type LimiterStatus struct {
}
var (
IOLimitTicket = time.Minute
IOLimitTicket = 60 // 1 min
IOLimitTicketInner = time.Millisecond * 100
LimitedIoError = errors.New("limited io error")
)
@ -59,16 +59,16 @@ 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, 0)
}
func NewIOLimiterEx(flowLimit, ioConcurrency, factor int) *IoLimiter {
func NewIOLimiterEx(flowLimit, ioConcurrency, factor, hangMaxSecond int) *IoLimiter {
flow := rate.NewLimiter(rate.Inf, 0)
if flowLimit > 0 {
flow = rate.NewLimiter(rate.Limit(flowLimit), flowLimit/2)
}
l := &IoLimiter{limit: flowLimit, flow: flow}
l.io.Store(newIOQueue(ioConcurrency, factor))
l.io.Store(newIOQueue(ioConcurrency, factor, hangMaxSecond))
return l
}
@ -88,7 +88,12 @@ func (l *IoLimiter) ResetFlow(flowLimit int) {
}
func (l *IoLimiter) ResetIO(ioConcurrency, factor int) {
q := l.io.Swap(newIOQueue(ioConcurrency, factor)).(*ioQueue)
q := l.io.Swap(newIOQueueEx(ioConcurrency, factor)).(*ioQueue)
q.Close()
}
func (l *IoLimiter) ResetIOEx(ioConcurrency, factor, hangMaxSecond int) {
q := l.io.Swap(newIOQueue(ioConcurrency, factor, hangMaxSecond)).(*ioQueue)
q.Close()
}
@ -123,6 +128,19 @@ func (l *IoLimiter) TryRun(size int, taskFn func()) bool {
return true
}
func (l *IoLimiter) TryRunWithContext(ctx context.Context, size int, taskFn func()) bool {
if size > 0 {
if err := l.flow.WaitN(ctx, size); err != nil {
log.LogWarnf("action[limitio] tryrun wait flow with %d %s", size, err.Error())
return false
}
}
if ok := l.getIO().TryRun(taskFn); !ok {
return false
}
return true
}
func (l *IoLimiter) Status() (st LimiterStatus) {
st = l.getIO().Status()
@ -142,7 +160,7 @@ func (l *IoLimiter) Status() (st LimiterStatus) {
}
func (l *IoLimiter) Close() {
q := l.io.Swap(newIOQueue(0, 0)).(*ioQueue)
q := l.io.Swap(newIOQueue(0, 0, 0)).(*ioQueue)
q.Close()
}
@ -154,17 +172,22 @@ type task struct {
}
type ioQueue struct {
wg sync.WaitGroup
once sync.Once
running uint32
concurrency int
stopCh chan struct{}
queue chan *task
midQueue chan *task
factor int
wg sync.WaitGroup
once sync.Once
running uint32
concurrency int
stopCh chan struct{}
queue chan *task
midQueue chan *task
factor int
hangMaxSecond int
}
func newIOQueue(concurrency, factor int) *ioQueue {
func newIOQueueEx(concurrency, factor int) *ioQueue {
return newIOQueue(concurrency, factor, 0)
}
func newIOQueue(concurrency, factor, hangMaxSecond int) *ioQueue {
q := &ioQueue{concurrency: concurrency}
if q.concurrency <= 0 {
return q
@ -173,6 +196,9 @@ func newIOQueue(concurrency, factor int) *ioQueue {
if factor <= 0 {
factor = defaultQueueFactor
}
if hangMaxSecond <= 0 {
q.hangMaxSecond = IOLimitTicket
}
q.factor = factor
q.midQueue = make(chan *task, 100)
q.stopCh = make(chan struct{})
@ -209,7 +235,7 @@ func (q *ioQueue) innerRun() {
case <-q.stopCh:
return
case task := <-q.midQueue:
if timeutil.GetCurrentTime().After(task.tm.Add(IOLimitTicket)) {
if timeutil.GetCurrentTime().After(task.tm.Add(time.Duration(q.hangMaxSecond) * time.Second)) {
task.err = LimitedIoError
close(task.done)
continue
@ -222,7 +248,7 @@ func (q *ioQueue) innerRun() {
case q.queue <- task:
stop = true
case <-tickerInner.C:
if timeutil.GetCurrentTime().After(task.tm.Add(IOLimitTicket)) {
if timeutil.GetCurrentTime().After(task.tm.Add(time.Duration(q.hangMaxSecond) * time.Second)) {
task.err = LimitedIoError
close(task.done)
stop = true

2
util/limit_io_test.go
View File

@ -93,7 +93,7 @@ func TestLimitIOTimeout(t *testing.T) {
t.Logf("Run rs: %+v", rs)
require.True(t, tVar == 1)
q.queue = make(chan *task)
IOLimitTicket = IOLimitTicketInner * 2
IOLimitTicket = 1 // 1 second
rs = q.Run(f, false)
require.True(t, rs == LimitedIoError)
}