diff --git a/blobstore/access/stream/stream_get_test.go b/blobstore/access/stream/stream_get_test.go index 9c6db5847..37e04b475 100644 --- a/blobstore/access/stream/stream_get_test.go +++ b/blobstore/access/stream/stream_get_test.go @@ -219,13 +219,9 @@ func TestAccessStreamGetShardTimeout(t *testing.T) { vuidController.Unblock(1001) }() { - startTime := time.Now() transfer, _ := streamer.Get(ctx(), bytes.NewBuffer(nil), *loc, uint64(size), 0) err := transfer() require.NoError(t, err) - - duration := time.Since(startTime) - require.GreaterOrEqual(t, vuidController.duration, duration, "greater duration: ", duration) } // delay one duration when blocking two shard, cos MinReadShardsX = 1 diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go index 3a41a303a..b7cd599eb 100644 --- a/datanode/data_partition_repair.go +++ b/datanode/data_partition_repair.go @@ -622,7 +622,7 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect dp.Disk().allocCheckLimit(proto.IopsReadType, 1) dp.Disk().allocCheckLimit(proto.FlowReadType, currReadSize) - if rs := dp.disk.limitRead.Run(int(currReadSize), func() { + if rs := dp.disk.limitRead.Run(int(currReadSize), false, func() { var crc uint32 crc, err = store.Read(reply.GetExtentID(), offset, int64(currReadSize), reply.GetData(), isRepairRead, p.GetOpcode() == proto.OpBackupRead) reply.SetCRC(crc) diff --git a/datanode/disk.go b/datanode/disk.go index 0f83f87d3..d05638fcb 100644 --- a/datanode/disk.go +++ b/datanode/disk.go @@ -149,8 +149,8 @@ func NewDisk(path string, reservedSpace, diskRdonlySpace uint64, maxErrCnt int, d.limitFactor[proto.FlowWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxFLowLimit), proto.QosDefaultBurst) d.limitFactor[proto.IopsReadType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) d.limitFactor[proto.IopsWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) - d.limitRead = newIOLimiter(space.dataNode.diskReadFlow, space.dataNode.diskReadIocc, false) - d.limitWrite = newIOLimiter(space.dataNode.diskWriteFlow, space.dataNode.diskWriteIocc, true) + d.limitRead = newIOLimiter(space.dataNode.diskReadFlow, space.dataNode.diskReadIocc) + d.limitWrite = newIOLimiter(space.dataNode.diskWriteFlow, space.dataNode.diskWriteIocc) err = d.initDecommissionStatus() if err != nil { diff --git a/datanode/limit_io.go b/datanode/limit_io.go index 15e1fbfa3..79890c7e0 100644 --- a/datanode/limit_io.go +++ b/datanode/limit_io.go @@ -21,6 +21,8 @@ import ( "sync/atomic" "time" + "github.com/cubefs/cubefs/util/timeutil" + "github.com/cubefs/cubefs/datanode/storage" "github.com/cubefs/cubefs/util/log" @@ -33,10 +35,9 @@ const ( ) type ioLimiter struct { - limit int - flow *rate.Limiter - io atomic.Value - allowHang bool + limit int + flow *rate.Limiter + io atomic.Value } type LimiterStatus struct { @@ -51,17 +52,17 @@ 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, allowHang bool) *ioLimiter { - return newIOLimiterEx(flowLimit, ioConcurrency, 0, allowHang) +func newIOLimiter(flowLimit, ioConcurrency int) *ioLimiter { + return newIOLimiterEx(flowLimit, ioConcurrency, 0) } -func newIOLimiterEx(flowLimit, ioConcurrency, factor int, allowHang bool) *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) } - l := &ioLimiter{limit: flowLimit, flow: flow, allowHang: allowHang} - l.io.Store(newIOQueue(ioConcurrency, factor, allowHang)) + l := &ioLimiter{limit: flowLimit, flow: flow} + l.io.Store(newIOQueue(ioConcurrency, factor)) return l } @@ -81,17 +82,17 @@ func (l *ioLimiter) ResetFlow(flowLimit int) { } func (l *ioLimiter) ResetIO(ioConcurrency, factor int) { - q := l.io.Swap(newIOQueue(ioConcurrency, factor, l.allowHang)).(*ioQueue) + q := l.io.Swap(newIOQueue(ioConcurrency, factor)).(*ioQueue) q.Close() } -func (l *ioLimiter) Run(size int, taskFn func()) (err error) { +func (l *ioLimiter) Run(size int, allowHang bool, taskFn func()) (err error) { 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()) } } - return l.getIO().Run(taskFn, l.allowHang) + return l.getIO().Run(taskFn, allowHang) } func (l *ioLimiter) TryRun(size int, taskFn func()) bool { @@ -126,7 +127,7 @@ func (l *ioLimiter) Status() (st LimiterStatus) { } func (l *ioLimiter) Close() { - q := l.io.Swap(newIOQueue(0, 0, l.allowHang)).(*ioQueue) + q := l.io.Swap(newIOQueue(0, 0)).(*ioQueue) q.Close() } @@ -147,7 +148,7 @@ type ioQueue struct { midQueue chan *task } -func newIOQueue(concurrency, factor int, allowHang bool) *ioQueue { +func newIOQueue(concurrency, factor int) *ioQueue { q := &ioQueue{concurrency: concurrency} if q.concurrency <= 0 { return q @@ -177,9 +178,7 @@ func newIOQueue(concurrency, factor int, allowHang bool) *ioQueue { }() } - if !allowHang { - go q.innerRun() - } + go q.innerRun() return q } @@ -193,23 +192,23 @@ func (q *ioQueue) innerRun() { case <-q.stopCh: return case task := <-q.midQueue: - if time.Now().After(task.tm.Add(IOLimitTicket)) { + if timeutil.GetCurrentTime().After(task.tm.Add(IOLimitTicket)) { task.err = storage.LimitedIoError close(task.done) - } else { - stop := false - for !stop { - select { - case <-q.stopCh: - return - case q.queue <- task: + continue + } + stop := false + for !stop { + select { + case <-q.stopCh: + return + case q.queue <- task: + stop = true + case <-tickerInner.C: + if timeutil.GetCurrentTime().After(task.tm.Add(IOLimitTicket)) { + task.err = storage.LimitedIoError + close(task.done) stop = true - case <-tickerInner.C: - if time.Now().After(task.tm.Add(IOLimitTicket)) { - task.err = storage.LimitedIoError - close(task.done) - stop = true - } } } } @@ -233,7 +232,7 @@ func (q *ioQueue) Run(taskFn func(), allowHang bool) (err error) { if !allowHang { ch = q.midQueue } - task := &task{fn: taskFn, done: make(chan struct{}), tm: time.Now()} + task := &task{fn: taskFn, done: make(chan struct{}), tm: timeutil.GetCurrentTime()} select { case <-q.stopCh: taskFn() diff --git a/datanode/limit_io_test.go b/datanode/limit_io_test.go index fc288e1ed..34ecb77eb 100644 --- a/datanode/limit_io_test.go +++ b/datanode/limit_io_test.go @@ -32,41 +32,41 @@ func TestLimitIOBase(t *testing.T) { {100, -1}, {1 << 20, 4}, } { - l := newIOLimiter(flowIO[0], flowIO[1], true) + l := newIOLimiter(flowIO[0], flowIO[1]) l.ResetFlow(flowIO[0]) l.ResetIO(flowIO[1], 0) - l.Run(0, f) - l.Run(10, f) + l.Run(0, true, f) + l.Run(10, true, f) require.True(t, l.TryRun(1, f)) l.Close() } { - l := newIOLimiter(1<<10, 0, true) - l.Run(10, f) + l := newIOLimiter(1<<10, 0) + l.Run(10, true, f) st := l.Status() t.Logf("status: %+v", st) require.Equal(t, 1<<10, st.FlowLimit) require.True(t, st.FlowUsed > 0) require.True(t, st.FlowUsed <= 10) require.True(t, l.TryRun(10, f)) - l.Run(1<<20, f) + l.Run(1<<20, true, f) l.TryRun(1<<20, f) l.Close() } { done := make(chan struct{}) - l := newIOLimiter(-1, 2, true) + l := newIOLimiter(-1, 2) st := l.Status() t.Logf("before status: %+v", st) for ii := 0; ii < st.IOConcurrency; ii++ { go func() { - l.Run(0, func() { <-done }) + l.Run(0, true, func() { <-done }) }() } for ii := 0; ii < st.IOQueue*2; ii++ { go func() { - l.Run(0, func() { <-done }) + l.Run(0, true, func() { <-done }) }() } time.Sleep(100 * time.Millisecond) @@ -87,7 +87,7 @@ func TestLimitIOTimeout(t *testing.T) { tVar = 1 t.Logf("func running!") } - l := newIOLimiter(-1, 1, false) + l := newIOLimiter(-1, 1) st := l.Status() t.Logf("before status: %+v", st) q := l.getIO() @@ -101,7 +101,7 @@ func TestLimitIOTimeout(t *testing.T) { } func TestLimitIOConcurrency(t *testing.T) { - l := newIOLimiter(1<<10, 10, true) + l := newIOLimiter(1<<10, 10) done := make(chan struct{}) go func() { for { @@ -127,7 +127,7 @@ func TestLimitIOConcurrency(t *testing.T) { time.Sleep(time.Microsecond) go func() { - l.Run(1, func() { <-done }) + l.Run(1, true, func() { <-done }) }() } }() diff --git a/datanode/nodeinfo.go b/datanode/nodeinfo.go index 83ac2cb3e..a43f32c83 100644 --- a/datanode/nodeinfo.go +++ b/datanode/nodeinfo.go @@ -17,7 +17,7 @@ const ( var ( IOLimitTicket = time.Minute - IOLimitTicketInner = time.Millisecond * 10 + IOLimitTicketInner = time.Millisecond * 100 nodeInfoStopC = make(chan struct{}) ) diff --git a/datanode/partition_op_by_raft.go b/datanode/partition_op_by_raft.go index 0cdedacbe..b27a2fc50 100644 --- a/datanode/partition_op_by_raft.go +++ b/datanode/partition_op_by_raft.go @@ -281,7 +281,7 @@ func (dp *DataPartition) ApplyRandomWrite(command []byte, raftApplyID uint64) (r syncWrite = true } - dp.disk.limitWrite.Run(int(opItem.size), func() { + dp.disk.limitWrite.Run(int(opItem.size), true, func() { param := &storage.WriteParam{ ExtentID: uint64(opItem.extentID), Offset: int64(opItem.offset), diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go index 62431129f..ace06f536 100644 --- a/datanode/wrap_operator.go +++ b/datanode/wrap_operator.go @@ -258,7 +258,7 @@ func (s *DataNode) handlePacketToCreateExtent(p *repl.Packet) { } partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - partition.disk.limitWrite.Run(0, func() { + partition.disk.limitWrite.Run(0, true, func() { err = partition.ExtentStore().Create(p.ExtentID) }) } @@ -759,7 +759,7 @@ func (s *DataNode) handleMarkDeletePacket(p *repl.Packet, c net.Conn) { log.LogInfof("handleMarkDeletePacket Delete PartitionID(%v)_Extent(%v)_Offset(%v)_Size(%v)", p.PartitionID, p.ExtentID, ext.ExtentOffset, ext.Size) partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - partition.disk.limitWrite.Run(0, func() { + partition.disk.limitWrite.Run(0, true, func() { log.LogInfof("[handleBatchMarkDeletePacket] vol(%v) dp(%v) mark delete extent(%v)", partition.config.VolName, partition.partitionID, p.ExtentID) err = partition.ExtentStore().MarkDelete(p.ExtentID, int64(ext.ExtentOffset), int64(ext.Size)) if err != nil { @@ -771,7 +771,7 @@ func (s *DataNode) handleMarkDeletePacket(p *repl.Packet, c net.Conn) { log.LogInfof("handleMarkDeletePacket Delete PartitionID(%v)_Extent(%v)", p.PartitionID, p.ExtentID) partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - if rs := partition.disk.limitWrite.Run(0, func() { + if rs := partition.disk.limitWrite.Run(0, true, func() { log.LogInfof("[handleBatchMarkDeletePacket] vol(%v) dp(%v) mark delete extent(%v)", partition.config.VolName, partition.partitionID, p.ExtentID) err = partition.ExtentStore().MarkDelete(p.ExtentID, 0, 0) if err != nil { diff --git a/datanode/wrap_operator_test.go b/datanode/wrap_operator_test.go index fa7418cf4..01ebeed97 100644 --- a/datanode/wrap_operator_test.go +++ b/datanode/wrap_operator_test.go @@ -48,8 +48,8 @@ func newDiskForOperatorTest(t *testing.T, dn *DataNode) (d *Disk) { d.limitFactor[proto.FlowWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxFLowLimit), proto.QosDefaultBurst) d.limitFactor[proto.IopsReadType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) d.limitFactor[proto.IopsWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) - d.limitRead = newIOLimiter(1*util.MB, 10, true) - d.limitWrite = newIOLimiter(1*util.MB, 10, true) + d.limitRead = newIOLimiter(1*util.MB, 10) + d.limitWrite = newIOLimiter(1*util.MB, 10) return }