improve lister

ydb-platform · Dec 2, 2024 · 0a25cd0 · 0a25cd0
1 parent c4247d6
commit 0a25cd0
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 11 deletions.
diff --git a/cloud/tasks/config/config.proto b/cloud/tasks/config/config.proto
@@ -64,4 +64,6 @@ message TasksConfig {
     // Warning: if this flag is false, then information about ended tasks
     // will not be cleared from the database.
     optional bool RegularSystemTasksEnabled = 33 [default = true];
+
+    optional uint64 TasksToListLimit = 34 [default = 10000];
 }
diff --git a/cloud/tasks/lister.go b/cloud/tasks/lister.go
@@ -20,6 +20,7 @@ type listTasksFunc = func(
 ) ([]storage.TaskInfo, error)
 
 type lister struct {
+	tasksToListLimit      uint64
 	listTasks             listTasksFunc
 	channels              []*channel
 	pollForTasksPeriodMin time.Duration
@@ -64,9 +65,7 @@ func (l *lister) loop(ctx context.Context) {
 			continue
 		}
 
-		limit := channelsLen - inflightTaskCount
-
-		tasks, err := l.listTasks(ctx, uint64(limit))
+		tasks, err := l.listTasks(ctx, l.tasksToListLimit)
 		if err == nil {
 			logging.Debug(ctx, "lister listed %v tasks", len(tasks))
 
@@ -170,6 +169,7 @@ func (l *lister) getInflightTaskCount() uint32 {
 
 func newLister(
 	ctx context.Context,
+	tasksToListLimit uint64,
 	listTasks listTasksFunc,
 	channelsCount uint64,
 	pollForTasksPeriodMin time.Duration,
@@ -178,6 +178,7 @@ func newLister(
 ) *lister {
 
 	lister := &lister{
+		tasksToListLimit:      tasksToListLimit,
 		listTasks:             listTasks,
 		channels:              make([]*channel, channelsCount),
 		pollForTasksPeriodMin: pollForTasksPeriodMin,

diff --git a/cloud/tasks/runner.go b/cloud/tasks/runner.go
@@ -900,6 +900,7 @@ func StartRunners(
 
 	listerReadyToRun := newLister(
 		ctx,
+		config.GetTasksToListLimit(),
 		func(ctx context.Context, limit uint64) ([]storage.TaskInfo, error) {
 			return taskStorage.ListTasksReadyToRun(
 				ctx,
@@ -914,6 +915,7 @@ func StartRunners(
 	)
 	listerReadyToCancel := newLister(
 		ctx,
+		config.GetTasksToListLimit(),
 		func(ctx context.Context, limit uint64) ([]storage.TaskInfo, error) {
 			return taskStorage.ListTasksReadyToCancel(
 				ctx,
@@ -951,6 +953,7 @@ func StartRunners(
 
 	listerStallingWhileRunning := newLister(
 		ctx,
+		config.GetTasksToListLimit(),
 		func(ctx context.Context, limit uint64) ([]storage.TaskInfo, error) {
 			return taskStorage.ListTasksStallingWhileRunning(
 				ctx,
@@ -966,6 +969,7 @@ func StartRunners(
 	)
 	listerStallingWhileCancelling := newLister(
 		ctx,
+		config.GetTasksToListLimit(),
 		func(ctx context.Context, limit uint64) ([]storage.TaskInfo, error) {
 			return taskStorage.ListTasksStallingWhileCancelling(
 				ctx,

diff --git a/cloud/tasks/runner_test.go b/cloud/tasks/runner_test.go
@@ -1421,13 +1421,14 @@ func testListerLoop(
 	}
 	lister := newLister(
 		ctx,
+		100, // tasksToListLimit
 		func(ctx context.Context, limit uint64) ([]storage.TaskInfo, error) {
 			return tasks, nil
 		},
-		uint64(channelCount),
-		50*time.Millisecond,
-		100*time.Millisecond,
-		make(map[string]int64),
+		uint64(channelCount),   // channelsCount
+		50*time.Millisecond,    // pollForTasksPeriodMin
+		100*time.Millisecond,   // pollForTasksPeriodMax
+		make(map[string]int64), // inflightTaskLimits
 	)
 
 	receivedTasks := make([]storage.TaskInfo, 0)
@@ -1497,13 +1498,14 @@ func TestListerLoopCancellingWhileReceiving(t *testing.T) {
 	}
 	lister := newLister(
 		ctx,
+		100, // tasksToListLimit
 		func(ctx context.Context, limit uint64) ([]storage.TaskInfo, error) {
 			return tasks, nil
 		},
-		uint64(channelCount),
-		50*time.Millisecond,
-		100*time.Millisecond,
-		make(map[string]int64),
+		uint64(channelCount),   // channelsCount
+		50*time.Millisecond,    // pollForTasksPeriodMin
+		100*time.Millisecond,   // pollForTasksPeriodMax
+		make(map[string]int64), // inflightTaskLimits
 	)
 
 	receivedTasks := make([]storage.TaskInfo, 0)