perf(scheduler): optimize delete process

with: #23065620

close #3721

Signed-off-by: JasonHu520 <huzongchao@oppo.com>

blobstore/scheduler/base/kafka_consumer_group.go

@@ -131,6 +131,10 @@ func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
 
 		// reset batch msgs and ticker
 		msgs = msgs[:0]
+		select {
+		case <-tk.C:
+		default:
+		}
 		tk.Reset(time.Second * time.Duration(consumer.maxWaitTimeS))
 	}
 }

blobstore/scheduler/blob_deleter.go

@@ -174,12 +174,13 @@ type BlobDeleteConfig struct {
 	MessagePunishTimeM     int `json:"message_punish_time_m"`
 	MessageSlowDownTimeS   int `json:"message_slow_down_time_s"`
 
-	TaskPoolSize    int              `json:"task_pool_size"`
-	MaxBatchSize    int              `json:"max_batch_size"`
-	BatchIntervalS  int              `json:"batch_interval_s"`
-	SafeDelayTimeH  int64            `json:"safe_delay_time_h"`
-	DeleteHourRange HourRange        `json:"delete_hour_range"`
-	DeleteLog       recordlog.Config `json:"delete_log"`
+	TaskPoolSize     int              `json:"task_pool_size"`
+	FailTaskPoolSize int              `json:"fail_task_pool_size"`
+	MaxBatchSize     int              `json:"max_batch_size"`
+	BatchIntervalS   int              `json:"batch_interval_s"`
+	SafeDelayTimeH   int64            `json:"safe_delay_time_h"`
+	DeleteHourRange  HourRange        `json:"delete_hour_range"`
+	DeleteLog        recordlog.Config `json:"delete_log"`
 }
 
 func (cfg *BlobDeleteConfig) topics() []string {
@@ -199,6 +200,7 @@ type BlobDeleteMgr struct {
 	closer.Closer
 	taskSwitch      *taskswitch.TaskSwitch
 	taskPool        *taskpool.TaskPool
+	failTaskPool    *taskpool.TaskPool
 	clusterTopology IClusterTopology
 	blobnodeCli     client.BlobnodeAPI
 
@@ -245,10 +247,12 @@ func NewBlobDeleteMgr(
 	}
 
 	tp := taskpool.New(cfg.TaskPoolSize, cfg.TaskPoolSize)
+	ftp := taskpool.New(cfg.FailTaskPoolSize, cfg.FailTaskPoolSize)
 
 	mgr := &BlobDeleteMgr{
 		taskSwitch:             taskSwitch,
 		taskPool:               &tp,
+		failTaskPool:           &ftp,
 		clusterTopology:        clusterTopology,
 		blobnodeCli:            blobnodeCli,
 		delSuccessCounter:      base.NewCounter(cfg.ClusterID, "delete", base.KindSuccess),
@@ -361,13 +365,15 @@ func (mgr *BlobDeleteMgr) Consume(msgs []*sarama.ConsumerMessage, consumerPause
 	defer mgr.maybeSlowDownConsume(delItems, consumerPause)
 	defer mgr.recordAllResult(delItems)
 
+	pool := mgr.getTaskPool(msgs[0].Topic)
+
 	span, ctx := trace.StartSpanFromContextWithTraceID(context.Background(), "BlobDeleteConsume", tracePrefix)
 	span.Infof("start delete msgs len[%d], topic[%s], partition[%d], offset[%d]", len(msgs), msgs[0].Topic, msgs[0].Partition, msgs[0].Offset)
 	wg := sync.WaitGroup{}
 	wg.Add(len(delItems))
 	for i := range delItems {
 		idx := i
-		mgr.taskPool.Run(func() {
+		pool.Run(func() {
 			mgr.handleOneMsg(ctx, &delItems[idx], consumerPause)
 			wg.Done()
 		})
@@ -382,6 +388,17 @@ func (mgr *BlobDeleteMgr) Consume(msgs []*sarama.ConsumerMessage, consumerPause
 	return true
 }
 
+func (mgr *BlobDeleteMgr) getTaskPool(topic string) *taskpool.TaskPool {
+	switch topic {
+	case mgr.cfg.Kafka.TopicFailed:
+		return mgr.failTaskPool
+	case mgr.cfg.Kafka.TopicNormal:
+		return mgr.taskPool
+	default:
+		return mgr.taskPool
+	}
+}
+
 func (mgr *BlobDeleteMgr) preProcessMsg(msgs []*sarama.ConsumerMessage) (ret []delBlobRet, batchTraceId string) {
 	ret = make([]delBlobRet, len(msgs))
 	firstValidMsg, batchTraceId := true, ""

blobstore/scheduler/blob_deleter_test.go

@@ -66,10 +66,12 @@ func newBlobDeleteMgr(t *testing.T) *BlobDeleteMgr {
 	delLogger.EXPECT().Close().AnyTimes().Return(nil)
 	delLogger.EXPECT().Encode(any).AnyTimes().Return(nil)
 	tp := taskpool.New(2, 2)
+	ftp := taskpool.New(2, 2)
 
 	return &BlobDeleteMgr{
-		taskSwitch: taskSwitch,
-		taskPool:   &tp,
+		taskSwitch:   taskSwitch,
+		taskPool:     &tp,
+		failTaskPool: &ftp,
 
 		safeDelayTime:   time.Hour,
 		clusterTopology: clusterTopology,

blobstore/scheduler/config.go

@@ -286,6 +286,7 @@ func (c *Config) fixBlobDeleteConfig() error {
 	}
 	c.BlobDelete.ClusterID = c.ClusterID
 	defaulter.LessOrEqual(&c.BlobDelete.TaskPoolSize, defaultTaskPoolSize)
+	defaulter.LessOrEqual(&c.BlobDelete.FailTaskPoolSize, defaultTaskPoolSize)
 	defaulter.LessOrEqual(&c.BlobDelete.DeleteLog.ChunkBits, defaultDeleteLogChunkSize)
 	defaulter.LessOrEqual(&c.BlobDelete.MessagePunishThreshold, defaultMessagePunishThreshold)
 	defaulter.LessOrEqual(&c.BlobDelete.MessagePunishTimeM, defaultMessagePunishTimeM)