From 46703bdc75cf3efac7119fe07136b303d417daec Mon Sep 17 00:00:00 2001 From: mawei029 Date: Thu, 7 Sep 2023 19:16:47 +0800 Subject: [PATCH] refactor(blobnode): optimize the qos of blobnode Signed-off-by: mawei029 --- blobstore/blobnode/base/qos/config.go | 7 +- blobstore/blobnode/base/qos/qos.go | 131 +++++++++--------- blobstore/blobnode/base/qos/qos_test.go | 25 +++- blobstore/blobnode/config.go | 19 --- blobstore/blobnode/core/chunk/chunk_test.go | 10 +- blobstore/blobnode/core/chunk/compact_test.go | 2 +- blobstore/blobnode/core/config.go | 19 +-- blobstore/blobnode/core/disk/disk.go | 6 +- .../blobnode/core/storage/datafile_test.go | 6 +- blobstore/blobnode/shard.go | 37 +---- blobstore/blobnode/shard_test.go | 10 +- blobstore/blobnode/startup.go | 3 - blobstore/blobnode/svr.go | 3 - blobstore/util/taskpool/iopool.go | 16 +-- blobstore/util/taskpool/iopool_mock.go | 14 -- blobstore/util/taskpool/iopool_test.go | 8 +- 16 files changed, 124 insertions(+), 192 deletions(-) diff --git a/blobstore/blobnode/base/qos/config.go b/blobstore/blobnode/base/qos/config.go index 2a1b86519..900335756 100644 --- a/blobstore/blobnode/base/qos/config.go +++ b/blobstore/blobnode/base/qos/config.go @@ -23,14 +23,15 @@ import ( const ( defaultMaxBandwidthMBPS = 1024 defaultBackgroundBandwidthMBPS = 10 - defaultMaxWaitCount = 512 + defaultMaxWaitCount = 1024 ) type Config struct { StatGetter flow.StatGetter `json:"-"` // Identify: a io flow DiskViewer iostat.IOViewer `json:"-"` // Identify: io viewer - ReadQueueLen int `json:"-"` - WriteQueueLen int `json:"-"` + ReadQueueDepth int `json:"-"` // equal $queueDepth of io pool: The number of elements in the queue + WriteQueueDepth int `json:"-"` // equal $queueDepth of io pool: The number of elements in the queue + WriteChanQueCnt int `json:"-"` // The number of chan queues, equal $chanCnt of write io pool MaxWaitCount int `json:"max_wait_count"` DiskBandwidthMBPS int64 `json:"disk_bandwidth_mbps"` BackgroundMBPS int64 `json:"background_mbps"` diff --git a/blobstore/blobnode/base/qos/qos.go b/blobstore/blobnode/base/qos/qos.go index 997028cd5..c40c8979b 100644 --- a/blobstore/blobnode/base/qos/qos.go +++ b/blobstore/blobnode/base/qos/qos.go @@ -46,36 +46,44 @@ const ( ReadType IOTypeRW = iota WriteType - writePoolCnt = 2 - reserve = 2 percent = 100 - ratioQueueLow = 50 - ratioQueueMiddle = 75 + ratioQueueLow = 1 + ratioQueueMiddle = 3 ratioDiscardLow = 30 ratioDiscardMid = 50 - ratioDiscardHigh = 80 + ratioDiscardHigh = 70 ) -type IoQosInternal struct { - maxQueueDepth int32 // max queue depth - queueLen int32 // current queue length - discardRatio int32 - rand *rand.Rand +type IoQosDiscard struct { + queueDepth int32 // queue depth, const + currentCnt int32 // current element cnt in queue, may be $queueDepth < $currentCnt < $maxWaitCnt + discardRatio int32 + rand *rand.Rand closer.Closer } -func newIoQosLimit(length int) *IoQosInternal { - qos := &IoQosInternal{ - maxQueueDepth: int32(length), - discardRatio: ratioDiscardLow, - rand: rand.New(rand.NewSource(time.Now().UnixNano())), - Closer: closer.New(), +func newIoQosDiscard(depth int) *IoQosDiscard { + qos := &IoQosDiscard{ + queueDepth: int32(depth), + discardRatio: ratioDiscardLow, + rand: rand.New(rand.NewSource(time.Now().UnixNano())), + Closer: closer.New(), } - go qos.adjustDiscardRatio() + go qos.adjustDiscardRatio() // dynamic adjust discard ratio return qos } -func (q *IoQosInternal) tryAcquire(ctx context.Context) bool { +// $depth: equal $queueDepth of io pool. The number of elements in the queue +// $cnt: The number of chan queues, equal $chanCnt of write io pool +func newWriteIoQosLimit(depth int, cnt int) []*IoQosDiscard { + w := make([]*IoQosDiscard, cnt) + for i := range w { + w[i] = newIoQosDiscard(depth) + } + return w +} + +func (q *IoQosDiscard) tryAcquire(ctx context.Context) bool { ioType := bnapi.GetIoType(ctx) if ioType.IsHighLevel() { q.incCount() @@ -84,40 +92,42 @@ func (q *IoQosInternal) tryAcquire(ctx context.Context) bool { return q.allowLowLevel() } -func (q *IoQosInternal) incCount() { - atomic.AddInt32(&q.queueLen, 1) +func (q *IoQosDiscard) incCount() { + atomic.AddInt32(&q.currentCnt, 1) } -func (q *IoQosInternal) decCount() { - atomic.AddInt32(&q.queueLen, -1) +func (q *IoQosDiscard) decCount() { + atomic.AddInt32(&q.currentCnt, -1) } -func (q *IoQosInternal) isNotFullLoad() bool { - return atomic.LoadInt32(&q.queueLen) < q.maxQueueDepth +func (q *IoQosDiscard) isNotFullLoad() bool { + return atomic.LoadInt32(&q.currentCnt) < q.queueDepth } -func (q *IoQosInternal) allowLowLevel() bool { +func (q *IoQosDiscard) allowLowLevel() bool { if q.isNotFullLoad() { q.incCount() return true } - + // If your score(rand.Intn) is lower than the threshold, return false and discard this IO if q.rand.Intn(percent) < q.getDiscardRatio() { return false } + q.incCount() return true } -func (q *IoQosInternal) release() { +func (q *IoQosDiscard) release() { q.decCount() } -func (q *IoQosInternal) getDiscardRatio() int { +func (q *IoQosDiscard) getDiscardRatio() int { return int(atomic.LoadInt32(&q.discardRatio)) } -func (q *IoQosInternal) adjustDiscardRatio() { +// dynamic adjust discard ratio +func (q *IoQosDiscard) adjustDiscardRatio() { tk := time.NewTicker(time.Second) defer tk.Stop() @@ -128,7 +138,13 @@ func (q *IoQosInternal) adjustDiscardRatio() { return } - ratio := atomic.LoadInt32(&q.queueLen) * percent / q.maxQueueDepth / reserve + currentCnt := atomic.LoadInt32(&q.currentCnt) + if currentCnt < q.queueDepth { + continue + } + + // adjust discard ratio, when $queueDepth <= $currentCnt <= $maxWaitCnt + ratio := (currentCnt - q.queueDepth) / q.queueDepth switch { case ratio < ratioQueueLow: atomic.StoreInt32(&q.discardRatio, ratioDiscardLow) @@ -140,31 +156,23 @@ func (q *IoQosInternal) adjustDiscardRatio() { } } -func newWriteIoQosLimit(length int) [writePoolCnt]*IoQosInternal { - w := [writePoolCnt]*IoQosInternal{ - newIoQosLimit(length), - newIoQosLimit(length), - } - return w -} - type IoQueueQos struct { - bpsLimiters []*rate.Limiter - maxWaitCnt chan struct{} - writeLimit [writePoolCnt]*IoQosInternal - readLimit *IoQosInternal - conf Config + maxWaitCnt int32 // const, max wait IO count + ioCnt int32 // current IO count + bpsLimiters []*rate.Limiter // limit bandwidth + readDiscard *IoQosDiscard // discard some low level IO + writeDiscard []*IoQosDiscard + conf Config closer.Closer } func NewIoQueueQos(conf Config) (Qos, error) { - maxWaitCnt := make(chan struct{}, conf.MaxWaitCount) qos := &IoQueueQos{ - maxWaitCnt: maxWaitCnt, - readLimit: newIoQosLimit(conf.ReadQueueLen), - writeLimit: newWriteIoQosLimit(conf.WriteQueueLen), - conf: conf, - Closer: closer.New(), + maxWaitCnt: int32(conf.MaxWaitCount), + readDiscard: newIoQosDiscard(conf.ReadQueueDepth), + writeDiscard: newWriteIoQosLimit(conf.WriteQueueDepth, conf.WriteChanQueCnt), + conf: conf, + Closer: closer.New(), } qos.initBpsLimiters() @@ -265,16 +273,15 @@ func (qos *IoQueueQos) Reader(ctx context.Context, ioType bnapi.IOType, reader i // whether beyond max wait num func (qos *IoQueueQos) Allow() bool { - select { - case qos.maxWaitCnt <- struct{}{}: - return true - default: + if atomic.AddInt32(&qos.ioCnt, 1) > qos.maxWaitCnt { + atomic.AddInt32(&qos.ioCnt, -1) return false } + return true } func (qos *IoQueueQos) Release() { - <-qos.maxWaitCnt + atomic.AddInt32(&qos.ioCnt, -1) } func (qos *IoQueueQos) TryAcquireIO(ctx context.Context, chunkId uint64, rwType IOTypeRW) bool { @@ -285,10 +292,10 @@ func (qos *IoQueueQos) TryAcquireIO(ctx context.Context, chunkId uint64, rwType ret := false switch rwType { case WriteType: - idx := chunkId % uint64(len(qos.writeLimit)) - ret = qos.writeLimit[idx].tryAcquire(ctx) + idx := chunkId % uint64(len(qos.writeDiscard)) + ret = qos.writeDiscard[idx].tryAcquire(ctx) case ReadType: - ret = qos.readLimit.tryAcquire(ctx) + ret = qos.readDiscard.tryAcquire(ctx) } return ret } @@ -296,17 +303,17 @@ func (qos *IoQueueQos) TryAcquireIO(ctx context.Context, chunkId uint64, rwType func (qos *IoQueueQos) ReleaseIO(chunkId uint64, rwType IOTypeRW) { switch rwType { case WriteType: - idx := chunkId % uint64(len(qos.writeLimit)) - qos.writeLimit[idx].release() + idx := chunkId % uint64(len(qos.writeDiscard)) + qos.writeDiscard[idx].release() case ReadType: - qos.readLimit.release() + qos.readDiscard.release() } qos.Release() } func (qos *IoQueueQos) Close() { - qos.readLimit.Close() - for _, w := range qos.writeLimit { + qos.readDiscard.Close() + for _, w := range qos.writeDiscard { w.Close() } qos.Closer.Close() diff --git a/blobstore/blobnode/base/qos/qos_test.go b/blobstore/blobnode/base/qos/qos_test.go index 9f2cd25a1..80c13915f 100644 --- a/blobstore/blobnode/base/qos/qos_test.go +++ b/blobstore/blobnode/base/qos/qos_test.go @@ -7,6 +7,7 @@ import ( "os" "strings" "testing" + "time" "github.com/cubefs/cubefs/blobstore/blobnode/base/flow" "github.com/cubefs/cubefs/blobstore/common/iostat" @@ -29,8 +30,9 @@ func TestNewQosManager(t *testing.T) { BackgroundMBPS: 1, DiskViewer: diskView, StatGetter: iom, - ReadQueueLen: 100, - WriteQueueLen: 100, + ReadQueueDepth: 100, + WriteQueueDepth: 100, + WriteChanQueCnt: 2, MaxWaitCount: 2 * 100, } qos, err := NewIoQueueQos(conf) @@ -101,8 +103,9 @@ func TestQosTryAcquire(t *testing.T) { BackgroundMBPS: 10, DiskViewer: diskView, StatGetter: statGet, - ReadQueueLen: 2, - WriteQueueLen: 2, + ReadQueueDepth: 2, + WriteQueueDepth: 2, + WriteChanQueCnt: 2, MaxWaitCount: 2 * 2, } qos, err := NewIoQueueQos(conf) @@ -133,7 +136,7 @@ func TestQosTryAcquire(t *testing.T) { { ctx = bnapi.SetIoType(ctx, bnapi.BackgroundIO) - for i := 0; i < q.conf.WriteQueueLen; i++ { + for i := 0; i < q.conf.WriteQueueDepth; i++ { ok = q.TryAcquireIO(ctx, 1, WriteType) require.True(t, ok) } @@ -152,3 +155,15 @@ func TestQosTryAcquire(t *testing.T) { require.Less(t, 0.0, ratio) } } + +func TestDynamicDiscard(t *testing.T) { + rdQueDepth := 128 + qos := newIoQosDiscard(rdQueDepth) + defer qos.Close() + require.Equal(t, int32(ratioDiscardLow), qos.discardRatio) + + time.Sleep(time.Millisecond * 100) + qos.currentCnt = 128 * 2 + time.Sleep(time.Millisecond * 1200) + require.Equal(t, int32(ratioDiscardMid), qos.discardRatio) +} diff --git a/blobstore/blobnode/config.go b/blobstore/blobnode/config.go index 47df7ba20..b83579b27 100644 --- a/blobstore/blobnode/config.go +++ b/blobstore/blobnode/config.go @@ -42,9 +42,6 @@ const ( DefaultChunkProtectionPeriodSec = 48 * 60 * 60 // 48 hour DefaultDiskStatusCheckIntervalSec = 2 * 60 // 2 min - DefaultPutQpsLimitPerDisk = 128 - DefaultGetQpsLimitPerDisk = 512 - DefaultGetQpsLimitPerKey = 64 DefaultDeleteQpsLimitPerDisk = 128 ) @@ -75,9 +72,6 @@ type Config struct { CleanExpiredStatIntervalSec int `json:"clean_expired_stat_interval_S"` DiskStatusCheckIntervalSec int `json:"disk_status_check_interval_S"` - PutQpsLimitPerDisk int `json:"put_qps_limit_per_disk"` - GetQpsLimitPerDisk int `json:"get_qps_limit_per_disk"` - GetQpsLimitPerKey int `json:"get_qps_limit_per_key"` DeleteQpsLimitPerDisk int `json:"delete_qps_limit_per_disk"` } @@ -115,18 +109,6 @@ func configInit(config *Config) { config.CleanExpiredStatIntervalSec = DefaultCleanExpiredStatIntervalSec } - if config.PutQpsLimitPerDisk <= 0 { - config.PutQpsLimitPerDisk = DefaultPutQpsLimitPerDisk - } - - if config.GetQpsLimitPerDisk == 0 { - config.GetQpsLimitPerDisk = DefaultGetQpsLimitPerDisk - } - - if config.GetQpsLimitPerKey == 0 { - config.GetQpsLimitPerKey = DefaultGetQpsLimitPerKey - } - if config.DeleteQpsLimitPerDisk <= 0 { config.DeleteQpsLimitPerDisk = DefaultDeleteQpsLimitPerDisk } @@ -135,7 +117,6 @@ func configInit(config *Config) { func (s *Service) changeLimit(ctx context.Context, c Config) { span := trace.SpanFromContextSafe(ctx) configInit(&c) - s.PutQpsLimitPerDisk.Reset(c.PutQpsLimitPerDisk) s.DeleteQpsLimitPerDisk.Reset(c.DeleteQpsLimitPerDisk) span.Info("hot reload limit config success.") } diff --git a/blobstore/blobnode/core/chunk/chunk_test.go b/blobstore/blobnode/core/chunk/chunk_test.go index f7a87d746..87784a59c 100644 --- a/blobstore/blobnode/core/chunk/chunk_test.go +++ b/blobstore/blobnode/core/chunk/chunk_test.go @@ -88,7 +88,7 @@ func TestNewChunkStorage(t *testing.T) { ioPool := taskpool.NewMockIoPool(ctr) ioPool.EXPECT().Submit(gomock.Any(), gomock.Any()).AnyTimes() - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 2, WriteQueueLen: 2, MaxWaitCount: 4}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 2, WriteQueueDepth: 2, MaxWaitCount: 4, WriteChanQueCnt: 2}) defer ioQos.Close() cs, err := NewChunkStorage(context.TODO(), datapath, vm, ioPool, ioPool, func(option *core.Option) { option.Conf = conf @@ -151,7 +151,7 @@ func TestChunkStorage_ReadWrite(t *testing.T) { } ioPool := newIoPoolMock(t) - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 2, WriteQueueLen: 2, MaxWaitCount: 4}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 2, WriteQueueDepth: 2, MaxWaitCount: 4, WriteChanQueCnt: 2}) defer ioQos.Close() cs, err := NewChunkStorage(ctx, datapath, vm, ioPool, ioPool, func(option *core.Option) { option.Conf = conf @@ -275,7 +275,7 @@ func TestChunkStorage_ReadWriteInline(t *testing.T) { } ioPool := newIoPoolMock(t) - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 2, WriteQueueLen: 2, MaxWaitCount: 4}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 2, WriteQueueDepth: 2, MaxWaitCount: 4, WriteChanQueCnt: 2}) defer ioQos.Close() cs, err := NewChunkStorage(ctx, datapath, vm, ioPool, ioPool, func(option *core.Option) { option.Conf = conf @@ -393,7 +393,7 @@ func TestChunkStorage_DeleteOp(t *testing.T) { } ioPool := newIoPoolMock(t) - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 2, WriteQueueLen: 2, MaxWaitCount: 4}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 2, WriteQueueDepth: 2, MaxWaitCount: 4, WriteChanQueCnt: 2}) defer ioQos.Close() cs, err := NewChunkStorage(ctx, datapath, vm, ioPool, ioPool, func(option *core.Option) { option.Conf = conf @@ -503,7 +503,7 @@ func TestChunkStorage_Finalizer(t *testing.T) { Status: bnapi.ChunkStatusNormal, } ioPool := newIoPoolMock(t) - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 2, WriteQueueLen: 2, MaxWaitCount: 4}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 2, WriteQueueDepth: 2, MaxWaitCount: 4, WriteChanQueCnt: 2}) defer ioQos.Close() cs, err := NewChunkStorage(ctx, datapath, vm, ioPool, ioPool, func(option *core.Option) { option.Conf = conf diff --git a/blobstore/blobnode/core/chunk/compact_test.go b/blobstore/blobnode/core/chunk/compact_test.go index cbdd594b1..1885f794a 100644 --- a/blobstore/blobnode/core/chunk/compact_test.go +++ b/blobstore/blobnode/core/chunk/compact_test.go @@ -171,7 +171,7 @@ func createTestChunk(t *testing.T, ctx context.Context, diskRoot string, vuid pr ctr := gomock.NewController(t) ioPool := taskpool.NewMockIoPool(ctr) ioPool.EXPECT().Submit(gomock.Any(), gomock.Any()).Do(func(taskId uint64, taskFn func()) { taskFn() }).AnyTimes() - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 200, WriteQueueLen: 200, MaxWaitCount: 400}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 200, WriteQueueDepth: 200, MaxWaitCount: 400, WriteChanQueCnt: 2}) defer ioQos.Close() chunk, err := NewChunkStorage(ctx, dataPath, vm, ioPool, ioPool, func(option *core.Option) { option.Conf = conf diff --git a/blobstore/blobnode/core/config.go b/blobstore/blobnode/core/config.go index 342b7cc01..006175fbf 100644 --- a/blobstore/blobnode/core/config.go +++ b/blobstore/blobnode/core/config.go @@ -44,9 +44,9 @@ const ( DefaultMetricReportIntervalS = int64(300) // 300 Sec DefaultBlockBufferSize = int64(64 * 1024) // 64k DefaultCompactEmptyRateThreshold = float64(0.8) // 80% rate - defaultWriteThreadCnt = 4 - defaultReadThreadCnt = 2 - defaultIOQueueLength = 256 + defaultWriteThreadCnt = 2 + defaultReadThreadCnt = 1 + defaultIOQueueDepth = 256 ) // Config for disk @@ -84,8 +84,8 @@ type RuntimeConfig struct { BlockBufferSize int64 `json:"block_buffer_size"` WriteThreadCnt int `json:"write_thread_cnt"` ReadThreadCnt int `json:"read_thread_cnt"` - WriteQueueLen int `json:"write_queue_len"` - ReadQueueLen int `json:"read_queue_len"` + WriteQueueDepth int `json:"write_queue_depth"` + ReadQueueDepth int `json:"read_queue_depth"` DataQos qos.Config `json:"data_qos"` } @@ -148,10 +148,11 @@ func InitConfig(conf *Config) error { defaulter.LessOrEqual(&conf.WriteThreadCnt, defaultWriteThreadCnt) defaulter.LessOrEqual(&conf.ReadThreadCnt, defaultReadThreadCnt) - defaulter.LessOrEqual(&conf.WriteQueueLen, defaultIOQueueLength) - defaulter.LessOrEqual(&conf.ReadQueueLen, defaultIOQueueLength) - conf.DataQos.ReadQueueLen = conf.ReadQueueLen - conf.DataQos.WriteQueueLen = conf.WriteQueueLen + defaulter.LessOrEqual(&conf.WriteQueueDepth, defaultIOQueueDepth) + defaulter.LessOrEqual(&conf.ReadQueueDepth, defaultIOQueueDepth) + conf.DataQos.ReadQueueDepth = conf.ReadQueueDepth + conf.DataQos.WriteQueueDepth = conf.WriteQueueDepth + conf.DataQos.WriteChanQueCnt = conf.WriteThreadCnt // $WriteChanQueCnt is equal to $WriteThreadCnt, one-to-one qos.InitAndFixQosConfig(&conf.DataQos) return nil diff --git a/blobstore/blobnode/core/disk/disk.go b/blobstore/blobnode/core/disk/disk.go index 180e76ead..ee4aa0af1 100644 --- a/blobstore/blobnode/core/disk/disk.go +++ b/blobstore/blobnode/core/disk/disk.go @@ -101,7 +101,7 @@ type DiskStorage struct { CreateAt int64 LastUpdateAt int64 - // io schedulers + // io pools writePool taskpool.IoPool readPool taskpool.IoPool } @@ -489,8 +489,8 @@ func newDiskStorage(ctx context.Context, conf core.Config) (ds *DiskStorage, err } // setting io pools - writePool := taskpool.NewWritePool(conf.WriteThreadCnt, conf.WriteQueueLen) - readPool := taskpool.NewReadPool(conf.ReadThreadCnt, conf.ReadQueueLen) + writePool := taskpool.NewWritePool(conf.WriteThreadCnt, conf.WriteQueueDepth) + readPool := taskpool.NewReadPool(conf.ReadThreadCnt, conf.ReadQueueDepth) ds = &DiskStorage{ DiskID: dm.DiskID, diff --git a/blobstore/blobnode/core/storage/datafile_test.go b/blobstore/blobnode/core/storage/datafile_test.go index 8cf4d2774..db26270d1 100644 --- a/blobstore/blobnode/core/storage/datafile_test.go +++ b/blobstore/blobnode/core/storage/datafile_test.go @@ -122,7 +122,7 @@ func TestChunkData_Write(t *testing.T) { } ioPool := newIoPoolMock(t) - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 2, WriteQueueLen: 2, MaxWaitCount: 4}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 2, WriteQueueDepth: 2, MaxWaitCount: 4, WriteChanQueCnt: 2}) defer ioQos.Close() cd, err := NewChunkData(ctx, core.VuidMeta{}, chunkname, diskConfig, true, ioQos, ioPool, ioPool) require.NoError(t, err) @@ -190,7 +190,7 @@ func TestChunkData_ConcurrencyWrite(t *testing.T) { concurrency := 10 ioPool := newIoPoolMock(t) - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: concurrency, WriteQueueLen: concurrency, MaxWaitCount: 2 * concurrency}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: concurrency, WriteQueueDepth: concurrency, MaxWaitCount: 2 * concurrency, WriteChanQueCnt: 2}) defer ioQos.Close() cd, err := NewChunkData(ctx, core.VuidMeta{}, chunkname, diskConfig, true, ioQos, ioPool, ioPool) require.NoError(t, err) @@ -271,7 +271,7 @@ func TestChunkData_Delete(t *testing.T) { RuntimeConfig: core.RuntimeConfig{BlockBufferSize: 64 * 1024}, } ioPool := newIoPoolMock(t) - ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueLen: 100, WriteQueueLen: 100, MaxWaitCount: 200}) + ioQos, _ := qos.NewIoQueueQos(qos.Config{ReadQueueDepth: 100, WriteQueueDepth: 100, MaxWaitCount: 200, WriteChanQueCnt: 2}) defer ioQos.Close() cd, err := NewChunkData(ctx, core.VuidMeta{}, chunkname, diskConfig, true, ioQos, ioPool, ioPool) require.NoError(t, err) diff --git a/blobstore/blobnode/shard.go b/blobstore/blobnode/shard.go index 73569e235..f18245f3c 100644 --- a/blobstore/blobnode/shard.go +++ b/blobstore/blobnode/shard.go @@ -35,7 +35,6 @@ import ( const ( ShardListPageLimit = 65536 - BidBatchReadLimit = 1024 ) /* @@ -98,28 +97,6 @@ func (s *Service) ShardGet(c *rpc.Context) { return } - start := time.Now() - limitKey := args.Bid - err = s.GetQpsLimitPerKey.Acquire(limitKey) - span.AppendTrackLog("lk.key", start, err) - if err != nil { - c.RespondError(bloberr.ErrOverload) - span.Warnf("shard get overload. args:%v err:%v", args, err) - return - } - defer s.GetQpsLimitPerKey.Release(limitKey) - - start = time.Now() - limitDiskKey := cs.Disk().ID() - err = s.GetQpsLimitPerDisk.Acquire(limitDiskKey) - span.AppendTrackLog("lk.disk", start, err) - if err != nil { - c.RespondError(bloberr.ErrOverload) - span.Warnf("shard get overload. args:%v err:%v", args, err) - return - } - defer s.GetQpsLimitPerDisk.Release(limitDiskKey) - // build shard reader shard := core.NewShardReader(args.Bid, args.Vuid, from, to, w) @@ -503,18 +480,6 @@ func (s *Service) ShardPut(c *rpc.Context) { return } - start := time.Now() - - limitKey := cs.Disk().ID() - err = s.PutQpsLimitPerDisk.Acquire(limitKey) - span.AppendTrackLog("lk.disk", start, err) - if err != nil { - span.Errorf("shard put overload. args:%v err:%v", args, err) - c.RespondError(bloberr.ErrOverload) - return - } - defer s.PutQpsLimitPerDisk.Release(limitKey) - if !cs.HasEnoughSpace(args.Size) { span.Errorf("cs has no enougn space. args:%v, chunk info:%v, disk:%v", args, cs.ChunkInfo(ctx), cs.Disk().Stats()) @@ -524,7 +489,7 @@ func (s *Service) ShardPut(c *rpc.Context) { shard := core.NewShardWriter(args.Bid, args.Vuid, uint32(args.Size), c.Request.Body) - start = time.Now() + start := time.Now() err = cs.Write(ctx, shard) span.AppendTrackLog("disk.put", start, err) diff --git a/blobstore/blobnode/shard_test.go b/blobstore/blobnode/shard_test.go index 93d00550c..f1a12040b 100644 --- a/blobstore/blobnode/shard_test.go +++ b/blobstore/blobnode/shard_test.go @@ -31,7 +31,6 @@ import ( bnapi "github.com/cubefs/cubefs/blobstore/api/blobnode" bloberr "github.com/cubefs/cubefs/blobstore/common/errors" "github.com/cubefs/cubefs/blobstore/common/proto" - "github.com/cubefs/cubefs/blobstore/util/limit/keycount" ) func noLimitClient() bnapi.StorageAPI { @@ -875,8 +874,6 @@ func TestShardGetConcurrency(t *testing.T) { client := noLimitClient() ctx := context.TODO() - service.GetQpsLimitPerKey = keycount.New(1) - diskID := proto.DiskID(101) vuid := proto.Vuid(2001) bid := proto.BlobID(30001) @@ -935,11 +932,8 @@ func TestShardGetConcurrency(t *testing.T) { for i := 0; i < concurrency; i++ { e := <-errChan - if e != nil { - return - } + require.NoError(t, e) } - t.Fatalf("get shard concurrency limit failed") } func TestShardPutConcurrency(t *testing.T) { @@ -950,8 +944,6 @@ func TestShardPutConcurrency(t *testing.T) { client := noLimitClient() ctx := context.TODO() - service.PutQpsLimitPerDisk = keycount.New(1) - diskID := proto.DiskID(101) vuid := proto.Vuid(2001) bid := proto.BlobID(30001) diff --git a/blobstore/blobnode/startup.go b/blobstore/blobnode/startup.go index e2062971b..07c9cdb66 100644 --- a/blobstore/blobnode/startup.go +++ b/blobstore/blobnode/startup.go @@ -247,9 +247,6 @@ func NewService(conf Config) (svr *Service, err error) { Disks: make(map[proto.DiskID]core.DiskAPI), Conf: &conf, - PutQpsLimitPerDisk: keycount.New(conf.PutQpsLimitPerDisk), - GetQpsLimitPerDisk: keycount.NewBlockingKeyCountLimit(conf.GetQpsLimitPerDisk), - GetQpsLimitPerKey: keycount.NewBlockingKeyCountLimit(conf.GetQpsLimitPerKey), DeleteQpsLimitPerDisk: keycount.New(conf.DeleteQpsLimitPerDisk), DeleteQpsLimitPerKey: keycount.NewBlockingKeyCountLimit(1), ChunkLimitPerVuid: keycount.New(1), diff --git a/blobstore/blobnode/svr.go b/blobstore/blobnode/svr.go index e09949655..5b7740c37 100644 --- a/blobstore/blobnode/svr.go +++ b/blobstore/blobnode/svr.go @@ -49,9 +49,6 @@ type Service struct { Conf *Config // limiter - PutQpsLimitPerDisk limit.ResettableLimiter - GetQpsLimitPerDisk limit.Limiter - GetQpsLimitPerKey limit.Limiter DeleteQpsLimitPerKey limit.Limiter DeleteQpsLimitPerDisk limit.ResettableLimiter ChunkLimitPerVuid limit.Limiter diff --git a/blobstore/util/taskpool/iopool.go b/blobstore/util/taskpool/iopool.go index aedf7730f..0fcc4f4db 100644 --- a/blobstore/util/taskpool/iopool.go +++ b/blobstore/util/taskpool/iopool.go @@ -22,7 +22,6 @@ import ( type IoPool interface { Submit(taskId uint64, taskFn func()) - TrySubmit(taskId uint64, taskFn func()) bool Close() } @@ -42,6 +41,7 @@ func NewWritePool(threadCnt, queueDepth int) IoPool { } func NewReadPool(threadCnt, queueDepth int) IoPool { + // Multiple $threadCnt share a same chan queue return newCommonIoPool(1, threadCnt, queueDepth) } @@ -66,7 +66,7 @@ func newCommonIoPool(chanCnt, threadCnt, queueDepth int) *ioPoolSimple { task.fn() task.done <- struct{}{} } - log.Warn("close io pool, exit") + log.Debug("close io pool") }() } return pool @@ -79,17 +79,6 @@ func (p *ioPoolSimple) Submit(taskId uint64, taskFn func()) { <-task.done } -func (p *ioPoolSimple) TrySubmit(taskId uint64, taskFn func()) bool { - idx, task := p.generateTask(taskId, taskFn) - - select { - case p.queue[idx] <- task: - return true - default: - return false - } -} - func (p *ioPoolSimple) generateTask(taskId uint64, taskFn func()) (idx uint64, task *taskInfo) { idx = taskId % uint64(len(p.queue)) @@ -106,4 +95,5 @@ func (p *ioPoolSimple) Close() { close(p.queue[i]) } p.wg.Wait() // wait all io task done + log.Info("close all io pool, exit") } diff --git a/blobstore/util/taskpool/iopool_mock.go b/blobstore/util/taskpool/iopool_mock.go index 0841dea6c..41f6ad6ee 100644 --- a/blobstore/util/taskpool/iopool_mock.go +++ b/blobstore/util/taskpool/iopool_mock.go @@ -56,17 +56,3 @@ func (mr *MockIoPoolMockRecorder) Submit(arg0, arg1 interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Submit", reflect.TypeOf((*MockIoPool)(nil).Submit), arg0, arg1) } - -// TrySubmit mocks base method. -func (m *MockIoPool) TrySubmit(arg0 uint64, arg1 func()) bool { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "TrySubmit", arg0, arg1) - ret0, _ := ret[0].(bool) - return ret0 -} - -// TrySubmit indicates an expected call of TrySubmit. -func (mr *MockIoPoolMockRecorder) TrySubmit(arg0, arg1 interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TrySubmit", reflect.TypeOf((*MockIoPool)(nil).TrySubmit), arg0, arg1) -} diff --git a/blobstore/util/taskpool/iopool_test.go b/blobstore/util/taskpool/iopool_test.go index 5ef45abf4..8eecb095b 100644 --- a/blobstore/util/taskpool/iopool_test.go +++ b/blobstore/util/taskpool/iopool_test.go @@ -33,16 +33,16 @@ func TestIoPoolSimple(t *testing.T) { { n := 0 closePool := NewReadPool(1, 2) - ok := closePool.TrySubmit(1, func() { + go closePool.Submit(1, func() { time.Sleep(time.Millisecond * 10) n++ }) - require.True(t, ok) - ok = closePool.TrySubmit(2, func() { + go closePool.Submit(2, func() { time.Sleep(time.Millisecond * 10) n++ }) - require.True(t, ok) + + time.Sleep(time.Second) closePool.Close() require.Equal(t, 2, n) // two task func }