[PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O opened a new pull request, #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336

   ## Purpose of the PR
   
   - Design a new scheduler
   
   ## Main Changes
   
   - Support priority/elder/depends based scheduling.
   
   ## Verifying these changes
   
   <!-- Please pick the proper options below -->
   
   - [ ] Trivial rework / code cleanup without any test coverage. (No Need)
   - [ ] Already covered by existing tests, such as *(please modify tests here)*.
   - [x] Need tests and can be verified as follows.
   
   I have write some scripts in /test folder.
   
   
   ## Does this PR potentially affect the following parts?
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [ ]  Nope
   - [ ]  Dependencies (add/update license info) <!-- Don't forget to add/update the info in "LICENSE" & "NOTICE" files (both in root & dist module) -->
   - [x]  Modify configurations
   - [x]  The public API
   - [ ]  Other affects (typed here)
   
   ## Documentation Status
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [x]  `Doc - TODO` <!-- Your PR changes impact docs and you will update later -->
   - [ ]  `Doc - Done` <!-- Related docs have been already added or updated -->
   - [ ]  `Doc - No Need` <!-- Your PR changes don't impact/need docs -->
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447297562


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)

Review Comment:
   1. This will be checked while running. 
   2. Have checked by taskID.
   3. We cannot limit the depth.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(vermeer): support task priority based scheduling [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin merged PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444649666


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -202,6 +420,7 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 		}
 	}()
 
+	// TODO: Is here need a lock? TOCTTOU

Review Comment:
   **潜在的并发安全问题**: 在第 423 行检查任务状态后到实际启动任务之间存在时间窗口,这是经典的 TOCTTOU (Time-of-check to Time-of-use) 问题。
   
   **问题**:
   - 在检查 taskInfo.State 和调用 taskStarter.StartTask() 之间,任务状态可能被其他 goroutine 修改
   - 这可能导致任务被重复启动或状态不一致
   
   **建议**:
   1. 在 taskStarter.StartTask() 内部进行原子性的状态检查和转换
   2. 使用互斥锁保护整个检查和启动过程
   3. 使用 atomic.CompareAndSwap 来原子地检查和更新状态
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447305400


##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {
+					return nil, fmt.Errorf("priority should be non-negative")
+				}
+				if p > math.MaxInt32 {
+					return nil, fmt.Errorf("priority exceeds maximum value: %d", math.MaxInt32)
+				}
+				taskInfo.Priority = int32(p)
+			} else {
+				logrus.Warnf("priority convert to int32 error:%v", err)
+				return nil, err
+			}
+		}
+		if preorders, ok := params["preorders"]; ok {
+			preorderList := strings.Split(preorders, ",")
+			for _, preorder := range preorderList {
+				if pid, err := strconv.ParseInt(preorder, 10, 32); err == nil {
+					if taskMgr.GetTaskByID(int32(pid)) == nil {

Review Comment:
   Have solved



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422651658


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +89,133 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+
+	// TODO: NEED TO JUDGE IF THE TASK CAN CONCURRENTLY RUNNING
+	// NOT only by user setting, but also by scheduler setting
+
+	// step 3: return the task with the highest priority or small tasks which can be executed immediately
+	taskToWorkerGroupMap := s.taskManager.GetTaskToWorkerGroupMap()
+	nextTasks, err := s.algorithmManager.ScheduleNextTasks(allTasks, taskToWorkerGroupMap, idleWorkerGroups, concurrentWorkerGroups, softSchedule)
+	if err != nil {
+		logrus.Errorf("failed to schedule next tasks: %v", err)
+		return err
+	}
+	logrus.Debugf("scheduled %d tasks", len(nextTasks))
+	// step 4: send to start channel
+	for _, task := range nextTasks {
+		if task == nil {
+			logrus.Warnf("received a nil task from algorithm manager")
+			continue

Review Comment:
   **Missing error handling for channel operations**
   
   If the start channel is full, the task is dropped with only a warning. This could lead to tasks never being executed without clear notification to the user.
   
   **Suggested fix:**
   ```go
   for _, task := range nextTasks {
       if task == nil {
           logrus.Warnf("received a nil task from algorithm manager")
           continue
       }
       if task.State != structure.TaskStateWaiting {
           logrus.Warnf("task '%d' is not in waiting state, current state: %s", task.ID, task.State)
           continue
       }
       logrus.Infof("scheduling task '%d' with type '%s' to start channel", task.ID, task.Type)
       select {
       case s.startChan <- task:
           logrus.Infof("task '%d' sent to start channel", task.ID)
       default:
           // Critical: Set task to error state instead of silently dropping
           errMsg := fmt.Sprintf("start channel is full, cannot schedule task %d", task.ID)
           logrus.Errorf(errMsg)
           taskMgr.SetError(task, errMsg)
       }
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444647531


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 
-	go s.dispatch()
+	if s.cronManager.CheckCronExpression(taskInfo.CronExpr) == nil {
+		if err := s.cronManager.AddCronTask(taskInfo); err != nil {
+			logrus.Errorf("failed to add cron task: %v", err)
+			return false, err
+		}
+		logrus.Infof("added cron task for task '%d' with expression '%s'", taskInfo.ID, taskInfo.CronExpr)
+	}
 
 	return ok, nil
 }
 
-func (s *ScheduleBl) CancelTask(taskInfo *structure.TaskInfo) error {
-	if taskInfo == nil {
-		return errors.New("the argument `taskInfo` is nil")
+/*
+* @Description: QueueTaskFromTemplate queues the task from the template.
+* @Note: This function will queue the task from the template. This function is used by cron tasks.
+* @Param template
+* @Return int32, error
+ */
+func (s *ScheduleBl) QueueTaskFromTemplate(template *structure.TaskInfo) (int32, error) {
+	if template == nil {
+		return -1, errors.New("the argument `template` is nil")
 	}
 
-	s.Lock()
-	isHeadTask := s.spaceQueue.IsHeadTask(taskInfo.ID)
-	task := s.spaceQueue.RemoveTask(taskInfo.ID)
-	s.Unlock(nil)
+	bc := &baseCreator{}
+	taskInfo, err := bc.CopyTaskInfo(template)
+	if err != nil {
+		logrus.Errorf("failed to copy task info from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
+	}
+	bc.saveTaskInfo(taskInfo)
 
-	isInQueue := false
-	if task != nil {
-		logrus.Infof("removed task '%d' from space queue", task.ID)
-		isInQueue = true
+	ok, err := s.QueueTask(taskInfo)
+	if err != nil || !ok {
+		logrus.Errorf("failed to queue task from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
 	}
 
-	if isInQueue && !isHeadTask {
-		if err := taskMgr.SetState(taskInfo, structure.TaskStateCanceled); err != nil {
-			return err
-		}
+	logrus.Infof("queued task '%d' from template '%d'", taskInfo.ID, template.ID)
 
-		logrus.Infof("set task '%d' to TaskStateCanceled", taskInfo.ID)
-	} else {
-		logrus.Infof("sending task '%d' to task canceler", taskInfo.ID)
-		return s.handleCancelTask(taskInfo)
+	return taskInfo.ID, nil
+}
+
+/*
+* @Description: BatchQueueTask batches the task.
+* @Note: This function will batch the task.
+* @Param taskInfos
+* @Return []bool, []error
+ */
+func (s *ScheduleBl) BatchQueueTask(taskInfos []*structure.TaskInfo) ([]bool, []error) {
+	if len(taskInfos) == 0 {
+		return []bool{}, []error{}
 	}
 
-	return nil
-}
+	s.PauseDispatch()
 
-func (s *ScheduleBl) IsDispatchPaused() bool {
-	return s.isDispatchPaused
-}
-func (s *ScheduleBl) PauseDispatch() {
-	s.isDispatchPaused = true
-}
+	defer s.ResumeDispatch()
+	// defer s.Unlock(s.Lock())
 
-func (s *ScheduleBl) ResumeDispatch() {
-	s.isDispatchPaused = false
+	errors := make([]error, len(taskInfos))
+	oks := make([]bool, len(taskInfos))
+
+	for _, taskInfo := range taskInfos {
+		ok, err := s.QueueTask(taskInfo)
+		if err != nil {
+			logrus.Errorf("failed to queue task '%d': %v", taskInfo.ID, err)
+		}
+		errors = append(errors, err)
+		oks = append(oks, ok)
+	}
+
+	return oks, errors
 }
 
-func (s *ScheduleBl) AllTasksInQueue() []*structure.TaskInfo {
-	return s.spaceQueue.AllTasks()
+// ******** CloseCurrent ********
+func (s *ScheduleBl) CloseCurrent(taskId int32, removeWorkerName ...string) error {
+	defer s.Unlock(s.Lock())
+
+	// trace tasks need these workers, check if these tasks are available
+	s.taskManager.RemoveTask(taskId)
+	// release the worker group
+	s.resourceManager.ReleaseByTaskID(taskId)
+
+	if len(removeWorkerName) > 0 {
+		// stop the cron job if exists when need remove worker, otherwise the task is just closed normally
+		s.cronManager.DeleteTask(taskId)
+		// remove the worker from resource manager
+		workerName := removeWorkerName[0]
+		if workerName == "" {
+			return errors.New("the argument `removeWorkerName` is empty")
+		}
+		logrus.Infof("removing worker '%s' from resource manager", workerName)
+		s.ChangeWorkerStatus(workerName, schedules.WorkerOngoingStatusDeleted)
+	}
+
+	logrus.Infof("invoke dispatch when task '%d' is closed", taskId)
+	s.TryScheduleNextTasks(true)
+	return nil
 }
 
-func (s *ScheduleBl) TasksInQueue(space string) []*structure.TaskInfo {
-	return s.spaceQueue.SpaceTasks(space)
+// This will be called when a worker is offline.
+// This will be called when a worker is online.
+func (s *ScheduleBl) ChangeWorkerStatus(workerName string, status schedules.WorkerOngoingStatus) (bool, error) {
+	defer s.Unlock(s.Lock())
+	s.resourceManager.ChangeWorkerStatus(workerName, status)
+
+	logrus.Infof("worker '%s' status changed to '%s'", workerName, status)
+	// After changing the worker status, we may need to reschedule tasks
+	s.TryScheduleNextTasks(true)
+
+	return true, nil
 }

Review Comment:
   **线程安全问题**:  从 channel 接收任务后直接调用 ,如果  执行时间较长,可能会阻塞其他任务的接收。
   
   **建议**:
   ```go
   func (s *ScheduleBl) waitingStartedTask() {
       for taskInfo := range s.startChan {
           if taskInfo == nil {
               logrus.Warnf("received a nil task from startChan")
               continue
           }
           
           logrus.Infof("chan received task '%d' to start", taskInfo.ID)
           // 使用 goroutine 异步处理,避免阻塞 channel 接收
           go s.handleStartTask(taskInfo)
       }
   }
   ```
   
   这样可以提高并发性能。



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 

Review Comment:
   **资源泄漏风险**: 如果 cron 任务调度成功但后续  失败,cron 任务不会被清理。
   
   **建议**:
   ```go
   if s.cronManager.CheckCronExpression(taskInfo.CronExpr) == nil {
       // 先队列任务,成功后再添加 cron
       ok, err := s.taskManager.QueueTask(taskInfo)
       if err != nil {
           taskMgr.SetError(taskInfo, err.Error())
           return ok, err
       }
       
       if err := s.cronManager.AddCronTask(taskInfo); err != nil {
           logrus.Errorf("failed to add cron task: %v", err)
           // 清理已队列的任务
           s.taskManager.RemoveTask(taskInfo.ID)
           return false, err
       }
   }
   ```
   
   或者使用事务性操作来确保原子性。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444645561


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)

Review Comment:
   **验证不足**: 依赖任务的验证只检查了任务是否存在,但没有检查:
   1. 依赖任务的状态(是否已完成、是否失败等)
   2. 是否存在循环依赖
   3. 依赖链的深度是否合理
   
   **建议**:
   ```go
   // 检查依赖任务状态
   if depTask.State == structure.TaskStateFailed {
       return false, fmt.Errorf("dependency task %d has failed", depTaskID)
   }
   
   // 检查循环依赖
   if s.taskManager.HasCircularDependency(taskInfo.ID, depTaskID) {
       return false, fmt.Errorf("circular dependency detected")
   }
   ```



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 

Review Comment:
   **配置值硬编码**: 这里使用字符串字面量 "3" 作为默认值,建议定义为常量。
   
   **建议**:
   ```go
   const (
       defaultStartChanSize = "10"
       defaultTickerInterval = "3"  // 秒
       defaultSoftSchedule = "true"
   )
   ```
   
   这样更易于维护和文档化。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447186755


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}

Review Comment:
   Have solved.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2381807620


##########
vermeer/test/functional/load_local.go:
##########
@@ -43,3 +46,28 @@ func (lt *LoadTaskLocal) TaskLoadBody() map[string]string {
 		"load.vertex_backend": vertexBackends[rand.Intn(len(vertexBackends))],
 	}
 }
+
+// TaskLoadBodyWithNum creates load configuration with specified number of files.
+// If num <= 10, it will be automatically adjusted to 30 to ensure minimum test coverage.
+func (lt *LoadTaskLocal) TaskLoadBodyWithNum(num int) map[string]string {
+	vertexBackends := []string{"db", "mem"}
+
+	if num <= 10 {
+		num = 30
+	}

Review Comment:
   Magic numbers 10 and 30 should be defined as named constants to improve code maintainability and explain why these specific values are chosen.



##########
vermeer/apps/master/schedules/scheduler_task_manager.go:
##########
@@ -0,0 +1,269 @@
+package schedules
+
+import (
+	"errors"
+	"vermeer/apps/structure"
+
+	"github.com/sirupsen/logrus"
+)
+
+/*
+* @Description: SchedulerTaskManager is the manager for the scheduler task.
+* @Note: This is the manager for the scheduler task.
+ */
+type SchedulerTaskManager struct {
+	structure.MutexLocker
+	// This struct is responsible for managing tasks in the scheduling system.
+	// A map from task ID to TaskInfo can be used to track tasks.
+	allTaskMap   map[int32]*structure.TaskInfo
+	allTaskQueue []*structure.TaskInfo
+	// For debug or test, get task start sequence
+	startTaskQueue []*structure.TaskInfo
+	// onGoingTasks
+	notCompleteTasks map[int32]*structure.TaskInfo
+	// A map from task ID to worker group can be used to track which worker group is handling which task.
+	taskToworkerGroupMap map[int32]string
+}
+
+/*
+* @Description: Init initializes the SchedulerTaskManager.
+* @Note: This function will initialize the SchedulerTaskManager.
+* @Return *SchedulerTaskManager
+ */
+func (t *SchedulerTaskManager) Init() *SchedulerTaskManager {
+	t.allTaskMap = make(map[int32]*structure.TaskInfo)
+	t.notCompleteTasks = make(map[int32]*structure.TaskInfo)
+	t.taskToworkerGroupMap = make(map[int32]string)
+	return t
+}
+
+/*
+* @Description: QueueTask queues the task.
+* @Note: This function will queue the task.
+* @Param taskInfo
+* @Return bool, error
+ */
+func (t *SchedulerTaskManager) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
+	if taskInfo == nil {
+		return false, errors.New("the argument `taskInfo` is nil")
+	}
+
+	if taskInfo.SpaceName == "" {
+		return false, errors.New("the property `SpaceName` of taskInfo is empty")
+	}
+
+	defer t.Unlock(t.Lock())
+
+	// Add the task to the task map
+	t.allTaskMap[taskInfo.ID] = taskInfo
+	t.allTaskQueue = append(t.allTaskQueue, taskInfo)
+	t.notCompleteTasks[taskInfo.ID] = taskInfo
+	t.AssignGroup(taskInfo)
+	return true, nil
+}
+
+/*
+* @Description: RefreshTaskToWorkerGroupMap refreshes the task to worker group map.
+* @Note: This function will refresh the task to worker group map.
+ */
+func (t *SchedulerTaskManager) RefreshTaskToWorkerGroupMap() {
+	defer t.Unlock(t.Lock())
+
+	for _, taskInfo := range t.GetAllTasksNotComplete() {
+		if taskInfo == nil {
+			continue
+		}
+		t.AssignGroup(taskInfo)
+		t.taskToworkerGroupMap[taskInfo.ID] = workerMgr.ApplyGroup(taskInfo.SpaceName, taskInfo.GraphName)
+	}
+}
+
+// Only for debug or test, get task start sequence
+/*
+* @Description: AddTaskStartSequence adds the task start sequence.
+* @Note: This function will add the task start sequence.
+* @Param taskID
+* @Return error
+ */
+func (t *SchedulerTaskManager) AddTaskStartSequence(taskID int32) error {
+	if _, exists := t.allTaskMap[taskID]; !exists {
+		return errors.New("task not found")
+	}
+	t.startTaskQueue = append(t.startTaskQueue, t.allTaskMap[taskID])
+	return nil
+}
+
+/*
+* @Description: RemoveTask removes the task.
+* @Note: This function will remove the task.
+* @Param taskID
+* @Return error
+ */
+func (t *SchedulerTaskManager) RemoveTask(taskID int32) error {
+	if _, exists := t.allTaskMap[taskID]; !exists {
+		return errors.New("task not found")
+	}
+	defer t.Unlock(t.Lock())
+	delete(t.allTaskMap, taskID)
+	// remove from queue
+	for i, task := range t.allTaskQueue {
+		if task.ID == taskID {
+			t.allTaskQueue = append(t.allTaskQueue[:i], t.allTaskQueue[i+1:]...)
+			break
+		}
+	}
+	delete(t.taskToworkerGroupMap, taskID)
+	delete(t.notCompleteTasks, taskID)
+	return nil
+}
+
+/*
+* @Description: MarkTaskComplete marks the task complete.
+* @Note: This function will mark the task complete.
+* @Param taskID
+* @Return error
+ */
+func (t *SchedulerTaskManager) MarkTaskComplete(taskID int32) error {
+	if _, exists := t.allTaskMap[taskID]; !exists {
+		return errors.New("task not found")
+	}
+	defer t.Unlock(t.Lock())
+	delete(t.notCompleteTasks, taskID)
+	return nil
+}
+
+// update or create a task in the task map
+/*
+* @Description: AssignGroup assigns the group.
+* @Note: This function will assign the group.
+* @Param taskInfo
+* @Return error
+ */
+func (t *SchedulerTaskManager) AssignGroup(taskInfo *structure.TaskInfo) error {
+	group := workerMgr.ApplyGroup(taskInfo.SpaceName, taskInfo.GraphName)
+	if group == "" {
+		return errors.New("failed to assign group for task")
+	}
+	t.taskToworkerGroupMap[taskInfo.ID] = group
+	return nil
+}
+
+/*
+* @Description: GetTaskByID gets the task by ID.
+* @Note: This function will get the task by ID.
+* @Param taskID
+* @Return *structure.TaskInfo, error
+ */
+func (t *SchedulerTaskManager) GetTaskByID(taskID int32) (*structure.TaskInfo, error) {
+	task, exists := t.allTaskMap[taskID]
+	if !exists {
+		return nil, errors.New("task not found")
+	}
+	return task, nil
+}
+
+/*
+* @Description: GetLastTask gets the last task.
+* @Note: This function will get the last task.
+* @Param spaceName
+* @Return *structure.TaskInfo
+ */
+func (t *SchedulerTaskManager) GetLastTask(spaceName string) *structure.TaskInfo {
+	// Implement logic to get the last task in the queue for the given space
+	if len(t.allTaskQueue) == 0 {
+		return nil
+	}
+	for i := len(t.allTaskQueue) - 1; i >= 0; i-- {
+		if t.allTaskQueue[i].SpaceName == spaceName {
+			return t.allTaskQueue[i]
+		}
+	}
+	return nil
+}
+
+/*
+* @Description: GetAllTasks gets all tasks.
+* @Note: This function will get all tasks.
+* @Return []*structure.TaskInfo
+ */
+func (t *SchedulerTaskManager) GetAllTasks() []*structure.TaskInfo {
+	tasks := make([]*structure.TaskInfo, 0, len(t.allTaskMap))
+	for _, task := range t.allTaskMap {
+		tasks = append(tasks, task)
+	}
+	return tasks
+}
+
+func (t *SchedulerTaskManager) GetAllTasksNotComplete() []*structure.TaskInfo {
+	tasks := make([]*structure.TaskInfo, 0, len(t.allTaskMap))
+	for _, task := range t.notCompleteTasks {
+		tasks = append(tasks, task)
+	}
+	return tasks
+}
+
+func (t *SchedulerTaskManager) GetAllTasksWaitng() []*structure.TaskInfo {

Review Comment:
   The function name has a typo: 'GetAllTasksWaitng' should be 'GetAllTasksWaiting'.
   ```suggestion
   func (t *SchedulerTaskManager) GetAllTasksWaiting() []*structure.TaskInfo {
   ```



##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +64,53 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {
+					return nil, fmt.Errorf("priority should be non-negative")
+				}
+				taskInfo.Priority = int32(p)
+			} else {
+				logrus.Warnf("priority convert to int32 error:%v", err)
+				return nil, err
+			}
+		}
+		if preorders, ok := params["preorders"]; ok {
+			preorderList := strings.Split(preorders, ",")
+			for _, preorder := range preorderList {
+				if pid, err := strconv.ParseInt(preorder, 10, 32); err == nil {
+					if taskMgr.GetTaskByID(int32(pid)) == nil {
+						return nil, fmt.Errorf("preorder task id %d not exists", pid)

Review Comment:
   The error message 'preorder task id %d not exists' has incorrect grammar. It should be 'preorder task with ID %d does not exist'.
   ```suggestion
   						return nil, fmt.Errorf("preorder task with ID %d does not exist", pid)
   ```



##########
vermeer/apps/master/schedules/scheduler_algorithm_manager.go:
##########
@@ -0,0 +1,507 @@
+package schedules
+
+import (
+	"slices"
+	"sort"
+	"strconv"
+	"time"
+	"vermeer/apps/common"
+	"vermeer/apps/structure"
+
+	"github.com/sirupsen/logrus"
+)
+
+/*
+* @Description: SchedulerAlgorithm is the interface for the scheduler algorithm.
+* @Note: This is the interface for the scheduler algorithm.
+ */
+type SchedulerAlgorithm interface {
+	// Name returns the name of the SchedulerAlgorithm
+	Name() string
+	// Init initializes the SchedulerAlgorithm
+	Init()
+	// FilterNextTasks filters the next tasks to be scheduled based on the provided parameters
+	FilterNextTasks(allTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error)
+	// ScheduleNextTasks schedules the next tasks based on the filtered tasks
+	ScheduleNextTasks(filteredTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error)
+}
+
+/*
+* @Description: SchedulerAlgorithmManager is the manager for the scheduler algorithm.
+* @Note: This is the manager for the scheduler algorithm.
+ */
+type SchedulerAlgorithmManager struct {
+	filteredSchedulerAlgorithms  map[string]SchedulerAlgorithm
+	scheduledSchedulerAlgorithms map[string]SchedulerAlgorithm
+	dispatchPaused               bool
+}
+
+/*
+* @Description: Init initializes the SchedulerAlgorithmManager.
+* @Note: This function will initialize the SchedulerAlgorithmManager.
+ */
+// Need to put DependsSchedulerAlgorithm before WaitingSchedulerAlgorithm
+func (am *SchedulerAlgorithmManager) Init() {
+	am.filteredSchedulerAlgorithms = make(map[string]SchedulerAlgorithm)
+	am.scheduledSchedulerAlgorithms = make(map[string]SchedulerAlgorithm)
+	am.dispatchPaused = false
+	// Register filter and schedule algorithms
+	am.RegisterFilterAlgorithm(&DependsSchedulerAlgorithm{})
+	am.RegisterFilterAlgorithm(&WaitingSchedulerAlgorithm{})
+	// Register default SchedulerAlgorithms
+	am.RegisterSchedulerAlgorithm(&PriorityElderSchedulerAlgorithm{})
+}
+
+/*
+* @Description: RegisterSchedulerAlgorithm registers the scheduler algorithm.
+* @Note: This function will register the scheduler algorithm.
+* @Param schedulerAlgorithm
+ */
+func (am *SchedulerAlgorithmManager) RegisterSchedulerAlgorithm(schedulerAlgorithm SchedulerAlgorithm) {
+	if schedulerAlgorithm == nil {
+		return
+	}
+	name := schedulerAlgorithm.Name()
+	if _, exists := am.scheduledSchedulerAlgorithms[name]; exists {
+		return // SchedulerAlgorithm already registered
+	}
+
+	// only support one scheduling algorithm for now
+	if len(am.scheduledSchedulerAlgorithms) > 0 {
+		return // Only one scheduling algorithm can be registered
+	}
+	schedulerAlgorithm.Init()
+	am.scheduledSchedulerAlgorithms[name] = schedulerAlgorithm
+}
+
+/*
+* @Description: RegisterFilterAlgorithm registers the filter algorithm.
+* @Note: This function will register the filter algorithm.
+* @Param filterAlgorithm
+ */
+func (am *SchedulerAlgorithmManager) RegisterFilterAlgorithm(filterAlgorithm SchedulerAlgorithm) {
+	if filterAlgorithm == nil {
+		return
+	}
+	name := filterAlgorithm.Name()
+	if _, exists := am.filteredSchedulerAlgorithms[name]; exists {
+		return // SchedulerAlgorithm already registered
+	}
+	filterAlgorithm.Init()
+	am.filteredSchedulerAlgorithms[name] = filterAlgorithm
+}
+
+/*
+* @Description: IsDispatchPaused checks if the dispatch is paused.
+* @Note: This function will check if the dispatch is paused.
+* @Return bool
+ */
+func (am *SchedulerAlgorithmManager) IsDispatchPaused() bool {
+	return am.dispatchPaused
+}
+
+/*
+* @Description: PauseDispatch pauses the dispatch.
+* @Note: This function will pause the dispatch.
+ */
+func (am *SchedulerAlgorithmManager) PauseDispatch() {
+	am.dispatchPaused = true
+}
+
+/*
+* @Description: ResumeDispatch resumes the dispatch.
+* @Note: This function will resume the dispatch.
+ */
+func (am *SchedulerAlgorithmManager) ResumeDispatch() {
+	am.dispatchPaused = false
+}
+
+/*
+* @Description: ScheduleNextTasks schedules the next tasks.
+* @Note: This function will schedule the next tasks.
+* @Param allTasks
+* @Param taskToWorkerGroupMap
+* @Param idleWorkerGroups
+* @Param concurrentWorkerGroups
+* @Param softSchedule
+* @Return []*structure.TaskInfo, error
+ */
+// For all tasks, filter and schedule them
+// Only one scheduling algorithm is supported for now
+func (am *SchedulerAlgorithmManager) ScheduleNextTasks(allTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error) {
+	if am.dispatchPaused {
+		return nil, nil // No tasks to schedule if dispatch is paused
+	}
+
+	filteredTasks := allTasks
+	for _, algorithm := range am.filteredSchedulerAlgorithms {
+		var err error
+		filteredTasks, err = algorithm.FilterNextTasks(filteredTasks, taskToWorkerGroupMap, idleWorkerGroups, concurrentWorkerGroups, softSchedule)
+		if err != nil {
+			return nil, err
+		}
+	}
+	if len(filteredTasks) == 0 {
+		return nil, nil // No tasks to schedule after filtering
+	}
+
+	// only support one scheduling algorithm for now
+	// get first algorithm
+	for _, algorithm := range am.scheduledSchedulerAlgorithms {
+		tasks, err := algorithm.ScheduleNextTasks(filteredTasks, taskToWorkerGroupMap, idleWorkerGroups, concurrentWorkerGroups, softSchedule)
+		if err != nil {
+			return nil, err
+		}
+		return tasks, nil // Return the scheduled tasks
+	}
+
+	return nil, nil // No tasks scheduled
+}
+
+type FIFOSchedulerAlgorithm struct{}
+
+func (f *FIFOSchedulerAlgorithm) Name() string {
+	return "FIFO"
+}
+
+func (f *FIFOSchedulerAlgorithm) Init() {
+	// No specific initialization needed for FIFO
+	logrus.Info("Initializing FIFOSchedulerAlgorithm")
+}
+
+func (f *FIFOSchedulerAlgorithm) FilterNextTasks(allTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error) {
+	// just return the waiting tasks as is for FIFO
+	return allTasks, nil
+}
+
+func (f *FIFOSchedulerAlgorithm) ScheduleNextTasks(allTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error) {
+	if len(allTasks) == 0 {
+		return nil, nil // No tasks to schedule
+	}
+
+	// For FIFO, we simply return the available tasks in the order they are provided
+	for _, task := range allTasks {
+		if task.State != structure.TaskStateWaiting {
+			continue // Only consider tasks that are in the waiting state
+		}
+		if group, exists := taskToWorkerGroupMap[task.ID]; exists && group != "" {
+			// only support idle worker groups for now
+			for _, idleGroup := range idleWorkerGroups {
+				if group == idleGroup {
+					logrus.Debugf("Task %d is assigned to worker group %s", task.ID, group)
+					return []*structure.TaskInfo{task}, nil // Return the first task that can be scheduled
+				}
+			}
+		}
+	}
+
+	return nil, nil
+}
+
+type PrioritySchedulerAlgorithm struct{}
+
+func (p *PrioritySchedulerAlgorithm) Name() string {
+	return "Priority"
+}
+
+func (p *PrioritySchedulerAlgorithm) Init() {
+	// No specific initialization needed for Priority
+	logrus.Info("Initializing PrioritySchedulerAlgorithm")
+}
+
+func (p *PrioritySchedulerAlgorithm) FilterNextTasks(allTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error) {
+	// just return the waiting tasks as is for Priority
+	return allTasks, nil
+}
+
+func (p *PrioritySchedulerAlgorithm) ScheduleNextTasks(allTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error) {
+	if len(allTasks) == 0 {
+		return nil, nil // No tasks to schedule
+	}
+
+	// Sort tasks by priority (higher priority first)
+	sort.Slice(allTasks, func(i, j int) bool {
+		return allTasks[i].Priority > allTasks[j].Priority
+	})
+
+	for _, task := range allTasks {
+		if task.State != structure.TaskStateWaiting {
+			continue // Only consider tasks that are in the waiting state
+		}
+		if group, exists := taskToWorkerGroupMap[task.ID]; exists && group != "" {
+			// only support idle worker groups for now
+			for _, idleGroup := range idleWorkerGroups {
+				if group == idleGroup {
+					logrus.Debugf("Task %d is assigned to worker group %s", task.ID, group)
+					return []*structure.TaskInfo{task}, nil // Return the first task that can be scheduled
+				}
+			}
+		}
+	}
+
+	return nil, nil
+}
+
+type PriorityElderSchedulerAlgorithm struct {
+	ageParam         int64
+	priorityParam    int64
+	resourceParam    int64
+	randomValueParam int64
+}
+
+func (p *PriorityElderSchedulerAlgorithm) Name() string {
+	return "PriorityElder"
+}
+
+func (p *PriorityElderSchedulerAlgorithm) Init() {
+	logrus.Info("Initializing PriorityElderSchedulerAlgorithm")
+
+	// Initialize parameters with default values
+	defaultAgeParam := "1"
+	defaultPriorityParam := "1"
+	defaultResourceParam := "10000000000"
+	defaultRandomValueParam := "1" // Placeholder for any random value logic
+
+	// Load parameters from configuration
+	ageParam := common.GetConfigDefault("priority_elder_age_param", defaultAgeParam).(string)
+	priorityParam := common.GetConfigDefault("priority_elder_priority_param", defaultPriorityParam).(string)
+	resourceParam := common.GetConfigDefault("priority_elder_resource_param", defaultResourceParam).(string)
+	randomValueParam := common.GetConfigDefault("priority_elder_random_value_param", defaultRandomValueParam).(string)
+
+	ageParamInt, err := strconv.Atoi(ageParam)
+	if err != nil {
+		logrus.Errorf("failed to convert priority_elder_age_param to int: %v", err)
+		logrus.Infof("using default priority_elder_age_param: %s", defaultAgeParam)
+		ageParamInt, _ = strconv.Atoi(defaultAgeParam)
+	}
+	p.ageParam = int64(ageParamInt)
+	priorityParamInt, err := strconv.Atoi(priorityParam)
+	if err != nil {
+		logrus.Errorf("failed to convert priority_elder_priority_param to int: %v", err)
+		logrus.Infof("using default priority_elder_priority_param: %s", defaultPriorityParam)
+		priorityParamInt, _ = strconv.Atoi(defaultPriorityParam)
+	}
+	p.priorityParam = int64(priorityParamInt)
+	resourceParamInt, err := strconv.Atoi(resourceParam)
+	if err != nil {
+		logrus.Errorf("failed to convert priority_elder_resource_param to int: %v", err)
+		logrus.Infof("using default priority_elder_resource_param: %s", defaultResourceParam)
+		resourceParamInt, _ = strconv.Atoi(defaultResourceParam)
+	}
+	p.resourceParam = int64(resourceParamInt)
+	randomValueParamInt, err := strconv.Atoi(randomValueParam)
+	if err != nil {
+		logrus.Errorf("failed to convert priority_elder_random_value_param to int: %v", err)
+		logrus.Infof("using default priority_elder_random_value_param: %s", defaultRandomValueParam)
+		randomValueParamInt, _ = strconv.Atoi(defaultRandomValueParam)
+	}
+	p.randomValueParam = int64(randomValueParamInt)
+
+	logrus.Infof("PriorityElderSchedulerAlgorithm initialized with parameters: ageParam=%d, priorityParam=%d, resourceParam=%d, randomValueParam=%d", p.ageParam, p.priorityParam, p.resourceParam, p.randomValueParam)
+}
+
+func (p *PriorityElderSchedulerAlgorithm) FilterNextTasks(allTasks []*structure.TaskInfo, taskToWorkerGroupMap map[int32]string, idleWorkerGroups []string, concurrentWorkerGroups []string, softSchedule bool) ([]*structure.TaskInfo, error) {
+	// just return the waiting tasks as is for PriorityElder
+	return allTasks, nil
+}
+
+func (p *PriorityElderSchedulerAlgorithm) CalculateTaskEmergency(task *structure.TaskInfo, taskToWorkerGroupMap map[int32]string, printValue bool) int64 {
+	// step 0: get params
+	ageParam := p.ageParam
+	priorityParam := p.priorityParam
+	resourceParam := p.resourceParam
+	randomValueParam := p.randomValueParam
+	// step 1: age
+	ageCost := ageParam * time.Since(task.CreateTime).Milliseconds() / 1000 // in seconds
+	// step 2: priority
+	priorityCost := priorityParam * int64(task.Priority)
+	// step 3: resource cost
+	graph := structure.GraphManager.GetGraphByName(task.SpaceName, task.GraphName)
+	resourceCost := int64(0)
+	if graph == nil {
+		resourceCost = resourceParam // if graph not found, use max resource cost
+	} else {
+		resourceCost = resourceParam / max(1, graph.VertexCount+graph.EdgeCount) // Avoid division by zero, ensure at least 1

Review Comment:
   The calculation logic for resourceCost is complex and should be extracted into a separate method with proper documentation explaining the algorithm.



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +225,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 
-	go s.dispatch()
+	if s.cronManager.CheckCronExpression(taskInfo.CronExpr) == nil {
+		if err := s.cronManager.AddCronTask(taskInfo); err != nil {
+			logrus.Errorf("failed to add cron task: %v", err)
+			return false, err
+		}
+		logrus.Infof("added cron task for task '%d' with expression '%s'", taskInfo.ID, taskInfo.CronExpr)
+	}
 
 	return ok, nil
 }
 
-func (s *ScheduleBl) CancelTask(taskInfo *structure.TaskInfo) error {
-	if taskInfo == nil {
-		return errors.New("the argument `taskInfo` is nil")
+/*
+* @Description: QueueTaskFromTemplate queues the task from the template.
+* @Note: This function will queue the task from the template. This function is used by cron tasks.
+* @Param template
+* @Return int32, error
+ */
+func (s *ScheduleBl) QueueTaskFromTemplate(template *structure.TaskInfo) (int32, error) {
+	if template == nil {
+		return -1, errors.New("the argument `template` is nil")
 	}
 
-	s.Lock()
-	isHeadTask := s.spaceQueue.IsHeadTask(taskInfo.ID)
-	task := s.spaceQueue.RemoveTask(taskInfo.ID)
-	s.Unlock(nil)
+	bc := &baseCreator{}
+	taskInfo, err := bc.CopyTaskInfo(template)
+	if err != nil {
+		logrus.Errorf("failed to copy task info from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
+	}
+	bc.saveTaskInfo(taskInfo)
 
-	isInQueue := false
-	if task != nil {
-		logrus.Infof("removed task '%d' from space queue", task.ID)
-		isInQueue = true
+	ok, err := s.QueueTask(taskInfo)
+	if err != nil || !ok {
+		logrus.Errorf("failed to queue task from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
 	}
 
-	if isInQueue && !isHeadTask {
-		if err := taskMgr.SetState(taskInfo, structure.TaskStateCanceled); err != nil {
-			return err
-		}
+	logrus.Infof("queued task '%d' from template '%d'", taskInfo.ID, template.ID)
 
-		logrus.Infof("set task '%d' to TaskStateCanceled", taskInfo.ID)
-	} else {
-		logrus.Infof("sending task '%d' to task canceler", taskInfo.ID)
-		return s.handleCancelTask(taskInfo)
+	return taskInfo.ID, nil
+}
+
+/*
+* @Description: BatchQueueTask batches the task.
+* @Note: This function will batch the task.
+* @Param taskInfos
+* @Return []bool, []error
+ */
+func (s *ScheduleBl) BatchQueueTask(taskInfos []*structure.TaskInfo) ([]bool, []error) {
+	if len(taskInfos) == 0 {
+		return []bool{}, []error{}
 	}
 
-	return nil
-}
+	s.PauseDispatch()
 
-func (s *ScheduleBl) IsDispatchPaused() bool {
-	return s.isDispatchPaused
-}
-func (s *ScheduleBl) PauseDispatch() {
-	s.isDispatchPaused = true
-}
+	defer s.ResumeDispatch()
+	// defer s.Unlock(s.Lock())
 
-func (s *ScheduleBl) ResumeDispatch() {
-	s.isDispatchPaused = false
+	errors := make([]error, len(taskInfos))
+	oks := make([]bool, len(taskInfos))
+
+	for _, taskInfo := range taskInfos {
+		ok, err := s.QueueTask(taskInfo)
+		if err != nil {
+			logrus.Errorf("failed to queue task '%d': %v", taskInfo.ID, err)
+		}
+		errors = append(errors, err)
+		oks = append(oks, ok)
+	}
+
+	return oks, errors
 }
 
-func (s *ScheduleBl) AllTasksInQueue() []*structure.TaskInfo {
-	return s.spaceQueue.AllTasks()
+// ******** CloseCurrent ********
+func (s *ScheduleBl) CloseCurrent(taskId int32, removeWorkerName ...string) error {
+	defer s.Unlock(s.Lock())
+
+	// trace tasks need these workers, check if these tasks are available
+	s.taskManager.RemoveTask(taskId)
+	// release the worker group
+	s.resourceManager.ReleaseByTaskID(taskId)
+
+	if len(removeWorkerName) > 0 {
+		// stop the cron job if exists when need remove worker, otherwise the task is just closed normally
+		s.cronManager.DeleteTask(taskId)
+		// remove the worker from resource manager
+		workerName := removeWorkerName[0]
+		if workerName == "" {
+			return errors.New("the argument `removeWorkerName` is empty")
+		}
+		logrus.Infof("removing worker '%s' from resource manager", workerName)
+		s.ChangeWorkerStatus(workerName, schedules.WorkerOngoingStatusDeleted)
+	}
+
+	logrus.Infof("invoke dispatch when task '%d' is closed", taskId)
+	s.TryScheduleNextTasks(true)
+	return nil
 }
 
-func (s *ScheduleBl) TasksInQueue(space string) []*structure.TaskInfo {
-	return s.spaceQueue.SpaceTasks(space)
+// This will be called when a worker is offline.
+// This will be called when a worker is online.
+func (s *ScheduleBl) ChangeWorkerStatus(workerName string, status schedules.WorkerOngoingStatus) (bool, error) {
+	defer s.Unlock(s.Lock())
+	s.resourceManager.ChangeWorkerStatus(workerName, status)
+
+	logrus.Infof("worker '%s' status changed to '%s'", workerName, status)
+	// After changing the worker status, we may need to reschedule tasks
+	s.TryScheduleNextTasks(true)
+
+	return true, nil
 }
 
-func (s *ScheduleBl) CloseCurrent(taskId int32) error {
-	logrus.Infof("invoke dispatch when task '%d' is closed", taskId)
-	s.dispatch()
+// ******** START TASK ********
+func (s *ScheduleBl) waitingStartedTask() {
+	for taskInfo := range s.startChan {
+		if taskInfo == nil {
+			logrus.Warnf("recieved a nil task from startChan")

Review Comment:
   Spelling error: 'recieved' should be 'received'.
   ```suggestion
   			logrus.Warnf("received a nil task from startChan")
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447135259


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -222,68 +441,115 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 
 	taskInfo.StartTime = time.Now()
 	err = taskStarter.StartTask()
+
+	// only for test or debug, record the task start sequence

Review Comment:
   It has been stated in the comments that it is only for testing purposes. And if it is not in debug mode, it will return an empty array.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422716174


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -28,16 +28,58 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+/*
+* @Description: ScheduleBl is the scheduler business logic.
+* @Note: This is the main scheduler business logic.
+ */
 type ScheduleBl struct {
 	structure.MutexLocker
-	dispatchLocker   structure.MutexLocker
-	spaceQueue       *schedules.SpaceQueue
-	broker           *schedules.Broker
-	startChan        chan *structure.TaskInfo
-	isDispatchPaused bool
+	// resource management
+	resourceManager *schedules.SchedulerResourceManager
+	// algorithm management
+	algorithmManager *schedules.SchedulerAlgorithmManager
+	// task management
+	taskManager *schedules.SchedulerTaskManager
+	// cron management
+	cronManager *schedules.SchedulerCronManager
+	// start channel for tasks to be started
+	startChan chan *structure.TaskInfo
+	// configurations
+	startChanSize  int
+	tickerInterval int
+	softSchedule   bool
 }
 
+/*
+* @Description: Init initializes the ScheduleBl.
+* @Note: This function will initialize the ScheduleBl.
+ */
 func (s *ScheduleBl) Init() {
+	logrus.Info("Initializing ScheduleBl...")
+	s.LoadConfig()
+	startChan := make(chan *structure.TaskInfo, s.startChanSize)
+	s.startChan = startChan
+
+	s.resourceManager = &schedules.SchedulerResourceManager{}
+	s.resourceManager.Init()
+	s.taskManager = &schedules.SchedulerTaskManager{}
+	s.taskManager.Init()
+	s.algorithmManager = &schedules.SchedulerAlgorithmManager{}

Review Comment:
   **Resource leak: goroutines started without context**
   
   The goroutines started in `Init()` run indefinitely without any shutdown mechanism. This prevents graceful shutdown and testing.
   
   **Suggested improvements:**
   ```go
   type ScheduleBl struct {
       // ... existing fields
       ctx context.Context
       cancel context.CancelFunc
   }
   
   func (s *ScheduleBl) Init() {
       logrus.Info("Initializing ScheduleBl...")
       s.LoadConfig()
       
       s.ctx, s.cancel = context.WithCancel(context.Background())
       startChan := make(chan *structure.TaskInfo, s.startChanSize)
       s.startChan = startChan
   
       // ... rest of initialization
   
       go s.startTicker(s.ctx)
       go s.waitingStartedTask(s.ctx)
   }
   
   func (s *ScheduleBl) Shutdown() {
       s.cancel()
       close(s.startChan)
   }
   
   func (s *ScheduleBl) startTicker(ctx context.Context) {
       ticker := time.NewTicker(time.Duration(s.tickerInterval) * time.Second)
       defer ticker.Stop()
       
       for {
           select {
           case <-ctx.Done():
               return
           case <-ticker.C:
               logrus.Debug("Ticker ticked")
               s.TryScheduleNextTasks()
           }
       }
   }
   ```



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -28,16 +28,58 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+/*
+* @Description: ScheduleBl is the scheduler business logic.
+* @Note: This is the main scheduler business logic.
+ */
 type ScheduleBl struct {
 	structure.MutexLocker
-	dispatchLocker   structure.MutexLocker
-	spaceQueue       *schedules.SpaceQueue
-	broker           *schedules.Broker
-	startChan        chan *structure.TaskInfo
-	isDispatchPaused bool
+	// resource management
+	resourceManager *schedules.SchedulerResourceManager
+	// algorithm management
+	algorithmManager *schedules.SchedulerAlgorithmManager
+	// task management
+	taskManager *schedules.SchedulerTaskManager
+	// cron management
+	cronManager *schedules.SchedulerCronManager
+	// start channel for tasks to be started
+	startChan chan *structure.TaskInfo
+	// configurations
+	startChanSize  int
+	tickerInterval int
+	softSchedule   bool
 }
 
+/*
+* @Description: Init initializes the ScheduleBl.
+* @Note: This function will initialize the ScheduleBl.
+ */
 func (s *ScheduleBl) Init() {
+	logrus.Info("Initializing ScheduleBl...")
+	s.LoadConfig()
+	startChan := make(chan *structure.TaskInfo, s.startChanSize)
+	s.startChan = startChan
+
+	s.resourceManager = &schedules.SchedulerResourceManager{}
+	s.resourceManager.Init()
+	s.taskManager = &schedules.SchedulerTaskManager{}
+	s.taskManager.Init()
+	s.algorithmManager = &schedules.SchedulerAlgorithmManager{}
+	s.algorithmManager.Init()
+	s.cronManager = &schedules.SchedulerCronManager{}
+	s.cronManager.Init(s.QueueTaskFromTemplate)
+	go s.startTicker()
+	go s.waitingStartedTask()
+}
+
+/*
+* @Description: LoadConfig loads the configuration from the common package.
+* @Note: This function will load the configuration from the common package.
+ */
+func (s *ScheduleBl) LoadConfig() {
+	// Load configuration from common package
+
+	// startChanSize
 	const defaultChanSizeConfig = "10"
 	chanSize := common.GetConfigDefault("start_chan_size", defaultChanSizeConfig).(string)

Review Comment:
   **Missing bounds validation for configuration values**
   
   The configuration loading doesn't validate that the parsed integer values are within reasonable bounds. Negative or extremely large values could cause issues.
   
   **Suggested fix:**
   ```go
   const (
       minStartChanSize = 1
       maxStartChanSize = 10000
       minTickerInterval = 1
       maxTickerInterval = 3600
   )
   
   chanSizeInt, err := strconv.Atoi(chanSize)
   if err != nil {
       logrus.Errorf("failed to convert start_chan_size to int: %v", err)
       logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
       chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
   } else if chanSizeInt < minStartChanSize || chanSizeInt > maxStartChanSize {
       logrus.Warnf("start_chan_size %d out of range [%d, %d], using default", 
           chanSizeInt, minStartChanSize, maxStartChanSize)
       chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
   }
   s.startChanSize = chanSizeInt
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422717426


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -28,16 +28,58 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+/*
+* @Description: ScheduleBl is the scheduler business logic.
+* @Note: This is the main scheduler business logic.
+ */
 type ScheduleBl struct {
 	structure.MutexLocker
-	dispatchLocker   structure.MutexLocker
-	spaceQueue       *schedules.SpaceQueue
-	broker           *schedules.Broker
-	startChan        chan *structure.TaskInfo
-	isDispatchPaused bool
+	// resource management
+	resourceManager *schedules.SchedulerResourceManager
+	// algorithm management
+	algorithmManager *schedules.SchedulerAlgorithmManager
+	// task management
+	taskManager *schedules.SchedulerTaskManager
+	// cron management
+	cronManager *schedules.SchedulerCronManager
+	// start channel for tasks to be started
+	startChan chan *structure.TaskInfo
+	// configurations
+	startChanSize  int
+	tickerInterval int
+	softSchedule   bool

Review Comment:
   **Missing test coverage for critical scheduling logic**
   
   The scheduler is the core component of this PR but there's no comprehensive test coverage shown for:
   1. Priority-based scheduling
   2. Dependency resolution
   3. Concurrent vs exclusive task handling
   4. Edge cases (circular dependencies, resource exhaustion, etc.)
   
   **Required test cases:**
   - Test priority ordering with multiple tasks
   - Test dependency chain execution order
   - Test circular dependency detection
   - Test concurrent task limits
   - Test resource allocation and release
   - Test cron task scheduling and cleanup
   - Test worker failure scenarios
   - Test race conditions in state transitions
   
   Please add integration tests in the `/test` folder that verify these scenarios.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422711697


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +225,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())

Review Comment:
   **Unconventional use of defer with unlock pattern**
   
   The pattern `defer s.Unlock(s.Lock())` is unusual and may be confusing. While it works, it evaluates `s.Lock()` immediately and schedules the unlock for later.
   
   **Suggested alternative:**
   ```go
   s.Lock()
   defer s.Unlock(nil)
   ```
   
   Or use a more conventional pattern:
   ```go
   s.locker.Lock()
   defer s.locker.Unlock()
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2441756101


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -202,6 +417,7 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 		}
 	}()
 
+	// TODO: Is here need a lock? TOCTTOU

Review Comment:
   It seems to need more time to do. It will need to change all APIs about task.setstate/taskmgr.setstate.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444649992


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 

Review Comment:
   **资源泄漏**: Cron 任务添加成功但后续 QueueTask 失败时,cron 任务不会被清理。
   
   **问题场景**:
   1. CheckCronExpression 返回 nil (表达式有效)
   2. AddCronTask 成功,cron 任务开始调度
   3. 后续代码失败,函数返回错误
   4. Cron 任务继续运行,但原始任务状态是失败的
   
   **建议 - 调整顺序**:
   ```go
   // 先完成任务队列操作
   ok, err := s.taskManager.QueueTask(taskInfo)
   if err != nil {
       taskMgr.SetError(taskInfo, err.Error())
       return ok, err
   }
   
   // 任务队列成功后再添加 cron
   if s.cronManager.CheckCronExpression(taskInfo.CronExpr) == nil {
       if err := s.cronManager.AddCronTask(taskInfo); err != nil {
           logrus.Errorf("failed to add cron task: %v", err)
           // 回滚:移除已队列的任务
           s.taskManager.RemoveTask(taskInfo.ID)
           return false, err
       }
   }
   ```
   



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)

Review Comment:
   **依赖验证不充分**: 只检查依赖任务是否存在,没有验证:
   
   **缺失的检查**:
   1. 依赖任务的状态(是否完成、失败、取消)
   2. 循环依赖检测
   3. 依赖链深度限制
   4. 依赖任务是否属于同一个 space
   
   **建议**:
   ```go
   for _, depTaskID := range taskInfo.Preorders {
       depTask := taskMgr.GetTaskByID(depTaskID)
       if depTask == nil {
           return false, fmt.Errorf("dependency task %d does not exist", depTaskID)
       }
   
       // 检查依赖任务状态
       if depTask.State == structure.TaskStateFailed {
           return false, fmt.Errorf("dependency task %d has failed", depTaskID)
       }
   
       // 检查循环依赖
       if hasCircularDependency(taskInfo.ID, depTaskID) {
           return false, errors.New("circular dependency detected")
       }
   
       // 检查 space 一致性
       if depTask.SpaceName != taskInfo.SpaceName {
           return false, errors.New("dependency task must be in same space")
       }
   }
   ```
   



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {

Review Comment:
   **Channel 满时的任务丢失风险**: 当 startChan 满时使用 select default 直接失败,可能导致任务丢失。
   
   **问题**:
   1. 如果 channel 满了,任务会被标记为错误但没有重试机制
   2. 这可能导致高优先级任务因为 channel 满而无法执行
   3. 没有背压机制来限制任务提交速率
   
   **建议**:
   1. 使用阻塞发送而不是 select default,或者实现重试机制
   2. 将任务放回队列而不是直接失败
   3. 添加监控指标跟踪 channel 满的频率
   4. 考虑使用有界队列配合背压机制
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444649810


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))

Review Comment:
   **严重的逻辑错误**: 使用 append 添加到预分配的 slice 会导致返回错误的结果。
   
   **当前问题**:
   ```go
   errors := make([]error, len(taskInfos))  // 创建 len 个 nil 元素
   oks := make([]bool, len(taskInfos))      // 创建 len 个 false 元素
   // ... 循环中
   errors = append(errors, err)  // 追加到末尾,导致 2*len 个元素
   oks = append(oks, ok)
   ```
   
   **结果**: 返回的 slice 长度是 2*len,前 len 个元素都是零值
   
   **修复方案 1 - 使用容量**:
   ```go
   errors := make([]error, 0, len(taskInfos))
   oks := make([]bool, 0, len(taskInfos))
   ```
   
   **修复方案 2 - 使用索引赋值**:
   ```go
   for i, taskInfo := range taskInfos {
       ok, err := s.QueueTask(taskInfo)
       errors[i] = err
       oks[i] = ok
   }
   ```
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447180770


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 

Review Comment:
   The previous steps have checked its correctness. And you can't join the queue directly here.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444649666


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -202,6 +420,7 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 		}
 	}()
 
+	// TODO: Is here need a lock? TOCTTOU

Review Comment:
   **潜在的并发安全问题**: 在第 423 行检查任务状态后到实际启动任务之间存在时间窗口,这是经典的 TOCTTOU (Time-of-check to Time-of-use) 问题。
   
   **问题**:
   - 在检查 taskInfo.State 和调用 taskStarter.StartTask() 之间,任务状态可能被其他 goroutine 修改
   - 这可能导致任务被重复启动或状态不一致
   
   **建议**:
   1. 在 taskStarter.StartTask() 内部进行原子性的状态检查和转换
   2. 使用互斥锁保护整个检查和启动过程
   3. 使用 atomic.CompareAndSwap 来原子地检查和更新状态
   



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {

Review Comment:
   **死锁风险**: BatchQueueTask 调用 PauseDispatch 后在 defer 中调用 ResumeDispatch,但被注释掉的锁代码表明这里可能存在锁的设计问题。
   
   **问题**:
   1. 注释掉的 defer s.Unlock(s.Lock()) 需要解释原因
   2. 如果在 PauseDispatch 和 ResumeDispatch 之间发生 panic,可能导致调度器永久暂停
   3. PauseDispatch 和 ResumeDispatch 的实现可能不是线程安全的
   
   **建议**:
   1. 添加注释说明为什么这行锁被注释掉
   2. 确保 PauseDispatch 和 ResumeDispatch 是线程安全的
   3. 使用更明确的锁管理策略
   



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))

Review Comment:
   **严重的逻辑错误**: 使用 append 添加到预分配的 slice 会导致返回错误的结果。
   
   **当前问题**:
   ```go
   errors := make([]error, len(taskInfos))  // 创建 len 个 nil 元素
   oks := make([]bool, len(taskInfos))      // 创建 len 个 false 元素
   // ... 循环中
   errors = append(errors, err)  // 追加到末尾,导致 2*len 个元素
   oks = append(oks, ok)
   ```
   
   **结果**: 返回的 slice 长度是 2*len,前 len 个元素都是零值
   
   **修复方案 1 - 使用容量**:
   ```go
   errors := make([]error, 0, len(taskInfos))
   oks := make([]bool, 0, len(taskInfos))
   ```
   
   **修复方案 2 - 使用索引赋值**:
   ```go
   for i, taskInfo := range taskInfos {
       ok, err := s.QueueTask(taskInfo)
       errors[i] = err
       oks[i] = ok
   }
   ```
   



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {

Review Comment:
   **Channel 满时的任务丢失风险**: 当 startChan 满时使用 select default 直接失败,可能导致任务丢失。
   
   **问题**:
   1. 如果 channel 满了,任务会被标记为错误但没有重试机制
   2. 这可能导致高优先级任务因为 channel 满而无法执行
   3. 没有背压机制来限制任务提交速率
   
   **建议**:
   1. 使用阻塞发送而不是 select default,或者实现重试机制
   2. 将任务放回队列而不是直接失败
   3. 添加监控指标跟踪 channel 满的频率
   4. 考虑使用有界队列配合背压机制
   



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)

Review Comment:
   **依赖验证不充分**: 只检查依赖任务是否存在,没有验证:
   
   **缺失的检查**:
   1. 依赖任务的状态(是否完成、失败、取消)
   2. 循环依赖检测
   3. 依赖链深度限制
   4. 依赖任务是否属于同一个 space
   
   **建议**:
   ```go
   for _, depTaskID := range taskInfo.Preorders {
       depTask := taskMgr.GetTaskByID(depTaskID)
       if depTask == nil {
           return false, fmt.Errorf("dependency task %d does not exist", depTaskID)
       }
   
       // 检查依赖任务状态
       if depTask.State == structure.TaskStateFailed {
           return false, fmt.Errorf("dependency task %d has failed", depTaskID)
       }
   
       // 检查循环依赖
       if hasCircularDependency(taskInfo.ID, depTaskID) {
           return false, errors.New("circular dependency detected")
       }
   
       // 检查 space 一致性
       if depTask.SpaceName != taskInfo.SpaceName {
           return false, errors.New("dependency task must be in same space")
       }
   }
   ```
   



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 

Review Comment:
   **资源泄漏**: Cron 任务添加成功但后续 QueueTask 失败时,cron 任务不会被清理。
   
   **问题场景**:
   1. CheckCronExpression 返回 nil (表达式有效)
   2. AddCronTask 成功,cron 任务开始调度
   3. 后续代码失败,函数返回错误
   4. Cron 任务继续运行,但原始任务状态是失败的
   
   **建议 - 调整顺序**:
   ```go
   // 先完成任务队列操作
   ok, err := s.taskManager.QueueTask(taskInfo)
   if err != nil {
       taskMgr.SetError(taskInfo, err.Error())
       return ok, err
   }
   
   // 任务队列成功后再添加 cron
   if s.cronManager.CheckCronExpression(taskInfo.CronExpr) == nil {
       if err := s.cronManager.AddCronTask(taskInfo); err != nil {
           logrus.Errorf("failed to add cron task: %v", err)
           // 回滚:移除已队列的任务
           s.taskManager.RemoveTask(taskInfo.ID)
           return false, err
       }
   }
   ```
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444643895


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -222,68 +441,115 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 
 	taskInfo.StartTime = time.Now()
 	err = taskStarter.StartTask()
+
+	// only for test or debug, record the task start sequence

Review Comment:
   **缺少错误处理**:  的错误只是被记录,但没有适当的处理。如果这是关键功能,应该考虑是否需要回滚或重试。
   
   **建议**: 
   - 如果  是仅用于调试/测试,应该在注释中明确说明失败不会影响正常功能
   - 考虑在测试/调试模式下才调用此方法



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}

Review Comment:
   **逻辑错误**: 这里使用  添加到预分配的 slice,会导致前面的  创建的空元素被保留。
   
   **当前代码问题**:
   ```go
   errors := make([]error, len(taskInfos))  // 创建了 len 个 nil 元素
   oks := make([]bool, len(taskInfos))      // 创建了 len 个 false 元素
   // ... 循环中
   errors = append(errors, err)  // 会追加到末尾,导致返回 2*len 个元素
   oks = append(oks, ok)
   ```
   
   **修复方案**:
   ```go
   errors := make([]error, 0, len(taskInfos))  // 使用容量而不是长度
   oks := make([]bool, 0, len(taskInfos))
   ```
   
   或者使用索引赋值:
   ```go
   for i, taskInfo := range taskInfos {
       ok, err := s.QueueTask(taskInfo)
       errors[i] = err
       oks[i] = ok
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444659706


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}

Review Comment:
   **严重的逻辑错误**: 使用 append 添加到预分配的 slice 会导致返回错误的结果。
   
   **当前问题**:
   ```go
   errors := make([]error, len(taskInfos))  // 创建 len 个 nil 元素
   oks := make([]bool, len(taskInfos))      // 创建 len 个 false 元素
   // ... 循环中
   errors = append(errors, err)  // 追加到末尾,导致 2*len 个元素
   oks = append(oks, ok)
   ```
   
   **结果**: 返回的 slice 长度是 2*len,前 len 个元素都是零值
   
   **修复方案 1 - 使用容量**:
   ```go
   errors := make([]error, 0, len(taskInfos))
   oks := make([]bool, 0, len(taskInfos))
   ```
   
   **修复方案 2 - 使用索引赋值**:
   ```go
   for i, taskInfo := range taskInfos {
       ok, err := s.QueueTask(taskInfo)
       errors[i] = err
       oks[i] = ok
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(vermeer): support task priority based scheduling [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2464778410


##########
vermeer/.gitignore:
##########
@@ -83,3 +83,7 @@ node_modules/
 /output/
 /bin/*
 !/bin/*.sh
+
+# 其他 #
+######################

Review Comment:
   ```suggestion
   # Others #
   ######################
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422717131


##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +64,53 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {
+					return nil, fmt.Errorf("priority should be non-negative")
+				}
+				taskInfo.Priority = int32(p)
+			} else {
+				logrus.Warnf("priority convert to int32 error:%v", err)
+				return nil, err
+			}
+		}
+		if preorders, ok := params["preorders"]; ok {
+			preorderList := strings.Split(preorders, ",")
+			for _, preorder := range preorderList {
+				if pid, err := strconv.ParseInt(preorder, 10, 32); err == nil {
+					if taskMgr.GetTaskByID(int32(pid)) == nil {
+						return nil, fmt.Errorf("preorder task id %d not exists", pid)
+					}
+					taskInfo.Preorders = append(taskInfo.Preorders, int32(pid))
+				} else {
+					logrus.Warnf("preorder convert to int32 error:%v", err)
+					return nil, err
+				}
+			}
+		}
+		if exclusive, ok := params["exclusive"]; ok {
+			if ex, err := strconv.ParseBool(exclusive); err == nil {
+				taskInfo.Exclusive = ex
+			} else {
+				logrus.Warnf("exclusive convert to bool error:%v", err)
+				return nil, err
+			}
+		}
+		if cronExpr, ok := params["cron_expr"]; ok {

Review Comment:
   **Security concern: Unrestricted cron expression execution**
   
   The cron expression is taken directly from user parameters without comprehensive validation or rate limiting. Malicious users could:
   1. Schedule tasks to run every second, causing resource exhaustion
   2. Use complex cron expressions to cause parsing overhead
   
   **Suggested mitigations:**
   ```go
   // Add validation
   func (s *ScheduleBl) validateCronExpression(cronExpr string) error {
       if err := s.cronManager.CheckCronExpression(cronExpr); err != nil {
           return err
       }
       
       // Parse and check minimum interval
       schedule, err := cron.ParseStandard(cronExpr)
       if err != nil {
           return err
       }
       
       // Ensure minimum 1-minute interval
       now := time.Now()
       next1 := schedule.Next(now)
       next2 := schedule.Next(next1)
       interval := next2.Sub(next1)
       
       if interval < time.Minute {
           return fmt.Errorf("cron interval too short: %v, minimum 1 minute required", interval)
       }
       
       return nil
   }
   ```
   
   Also add per-user rate limiting for cron task creation.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444661235


##########
vermeer/apps/master/master_main.go:
##########
@@ -56,6 +56,7 @@ func Main() {
 		services.SetUI(sen)
 		logrus.Info("token-auth was activated")
 	default:

Review Comment:
   **代码风格**: 这里有一个空的 default 分支并且后面有多余的空行。应该删除或添加注释说明为什么需要空的 default 分支。
   
   **建议**:
   ```go
   default:
       // No authentication required for other routes
   ```
   
   或者如果不需要这个 default 分支,可以直接删除。



##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {

Review Comment:
   **类型转换安全性**: 虽然已经检查了 p < 0,但还应该检查转换后的值是否会溢出。
   
   **建议**:
   ```go
   if p < 0 || p > math.MaxInt32 {
       return nil, fmt.Errorf("priority %d out of valid range [0, %d]", p, math.MaxInt32)
   }
   taskInfo.Priority = int32(p)
   ```
   
   这样可以防止在 32 位系统上出现意外的溢出问题。



##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {
+					return nil, fmt.Errorf("priority should be non-negative")
+				}
+				if p > math.MaxInt32 {
+					return nil, fmt.Errorf("priority exceeds maximum value: %d", math.MaxInt32)
+				}
+				taskInfo.Priority = int32(p)
+			} else {
+				logrus.Warnf("priority convert to int32 error:%v", err)
+				return nil, err
+			}
+		}
+		if preorders, ok := params["preorders"]; ok {
+			preorderList := strings.Split(preorders, ",")
+			for _, preorder := range preorderList {
+				if pid, err := strconv.ParseInt(preorder, 10, 32); err == nil {
+					if taskMgr.GetTaskByID(int32(pid)) == nil {

Review Comment:
   **错误消息国际化**: 错误消息使用了中文和英文混合。建议统一使用英文,或者实现完整的国际化支持。
   
   **示例**:
   ```go
   // 当前
   return nil, fmt.Errorf("preorder task id %d not exists", depTaskID)
   
   // 建议改为统一英文
   return nil, fmt.Errorf("preorder task id %d does not exist", depTaskID)
   ```
   
   或者实现国际化框架:
   ```go
   return nil, i18n.Errorf("error.task.dependency_not_found", depTaskID)
   ```



##########
vermeer/apps/master/bl/task_creator.go:
##########
@@ -99,6 +99,28 @@ func (bc *baseCreator) NewTaskInfo(graphName string, params map[string]string, t
 	return task, nil
 }
 
+func (bc *baseCreator) CopyTaskInfo(src *structure.TaskInfo) (*structure.TaskInfo, error) {

Review Comment:
   **测试覆盖率**: 这个函数复制了任务信息,但清除了 cron 表达式。这个行为需要单元测试来确保:
   1. 所有字段都被正确复制
   2. cron 表达式确实被清除
   3. 新任务 ID 是唯一的
   
   **建议添加测试**:
   ```go
   func TestCopyTaskInfo(t *testing.T) {
       original := &structure.TaskInfo{
           ID: 1,
           CronExpr: "0 0 * * *",
           // ... 其他字段
       }
   
       copied := CopyTaskInfo(original)
   
       assert.NotEqual(t, original.ID, copied.ID)
       assert.Empty(t, copied.CronExpr)
       assert.Equal(t, original.Name, copied.Name)
       // ... 验证其他字段
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444646140


##########
vermeer/apps/master/master_main.go:
##########
@@ -56,6 +56,7 @@ func Main() {
 		services.SetUI(sen)
 		logrus.Info("token-auth was activated")
 	default:

Review Comment:
   **代码风格**: 这里有一个空的  分支并且后面有多余的空行。应该删除或添加注释说明为什么需要空的 default 分支。
   
   **建议**: 
   ```go
   default:
       // No authentication required for other modes
   ```
   
   或者如果不需要 default 分支,直接删除。



##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {

Review Comment:
   **类型转换安全性**: 虽然已经检查了 ,但还应该检查转换后的值是否会溢出。
   
   **建议**:
   ```go
   if p < 0 || p > math.MaxInt32 {
       return nil, fmt.Errorf("priority must be between 0 and %d, got %d", math.MaxInt32, p)
   }
   ```
   
   另外,考虑直接使用  的第三个参数来限制位数,这样就不需要额外检查了。



##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {
+					return nil, fmt.Errorf("priority should be non-negative")
+				}
+				if p > math.MaxInt32 {
+					return nil, fmt.Errorf("priority exceeds maximum value: %d", math.MaxInt32)
+				}
+				taskInfo.Priority = int32(p)
+			} else {
+				logrus.Warnf("priority convert to int32 error:%v", err)
+				return nil, err
+			}
+		}
+		if preorders, ok := params["preorders"]; ok {
+			preorderList := strings.Split(preorders, ",")
+			for _, preorder := range preorderList {
+				if pid, err := strconv.ParseInt(preorder, 10, 32); err == nil {
+					if taskMgr.GetTaskByID(int32(pid)) == nil {

Review Comment:
   **错误消息国际化**: 错误消息使用了中文和英文混合。建议统一使用英文,或者实现完整的国际化支持。
   
   **示例**:
   ```go
   // 当前
   return nil, fmt.Errorf("preorder task with ID %d does not exist", pid)
   
   // 建议保持一致的风格
   ```
   
   这对于开源项目尤其重要,因为国际贡献者可能不懂中文。



##########
vermeer/apps/master/bl/task_creator.go:
##########
@@ -99,6 +99,28 @@ func (bc *baseCreator) NewTaskInfo(graphName string, params map[string]string, t
 	return task, nil
 }
 
+func (bc *baseCreator) CopyTaskInfo(src *structure.TaskInfo) (*structure.TaskInfo, error) {

Review Comment:
   **测试覆盖率**: 这个函数复制了任务信息,但清除了 cron 表达式。这个行为需要单元测试来确保:
   1. 所有字段都被正确复制
   2. cron 表达式确实被清除
   3. 新任务 ID 是唯一的
   
   **建议**: 添加单元测试:
   ```go
   func TestCopyTaskInfo(t *testing.T) {
       // Test case 1: nil input
       // Test case 2: all fields copied correctly
       // Test case 3: cron expression cleared
       // Test case 4: new task ID generated
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444660658


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -222,68 +441,115 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 
 	taskInfo.StartTime = time.Now()
 	err = taskStarter.StartTask()
+
+	// only for test or debug, record the task start sequence

Review Comment:
   **缺少错误处理**: StartSequence 的错误只是被记录,但没有适当的处理。如果这是关键功能,应该考虑是否需要回滚或重试。
   
   **建议**:
   - 如果 StartSequence 是仅用于调试/测试,应该在注释中明确说明失败不会影响正常功能
   - 如果是关键功能,应该考虑错误处理策略:
     ```go
     if s.startSequence != nil {
         if err := s.startSequence.StartSequence(taskInfo); err != nil {
             logrus.Errorf("start sequence failed: %v, rolling back task", err)
             s.taskManager.RemoveTask(taskInfo.ID)
             return err
         }
     }
     ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447308299


##########
vermeer/apps/master/master_main.go:
##########
@@ -56,6 +56,7 @@ func Main() {
 		services.SetUI(sen)
 		logrus.Info("token-auth was activated")
 	default:

Review Comment:
   This is not an empty branch.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447191011


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+
+	// TODO: NEED TO JUDGE IF THE TASK CAN CONCURRENTLY RUNNING
+	// NOT only by user setting, but also by scheduler setting

Review Comment:
   This can be shown by the error array.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#issuecomment-3393058130

   # Code Review: Priority/Elder/Depends Based Scheduling
   
   ## Summary
   This PR introduces a comprehensive scheduler redesign with priority-based scheduling, task dependencies, and cron support. The architecture is well-designed, but there are **critical
  concurrency issues** that must be addressed before merging.
   
   **Verdict: REQUEST CHANGES** ⚠️
   
   ---
   
   ## Architecture Changes
   
   ### Major Refactoring ✅
   The PR completely redesigns the scheduler from a simple queue-based system to a sophisticated multi-component architecture:
   
   **Old Architecture:**
   - Simple `SpaceQueue` (FIFO)
   - Basic `Broker` for worker assignment
   - Single-threaded dispatch loop
   
   **New Architecture:**
   - `SchedulerResourceManager` - Worker pool management
   - `SchedulerTaskManager` - Task lifecycle, dependencies, priorities
   - `SchedulerAlgorithmManager` - Scheduling algorithm logic
   - `SchedulerCronManager` - Recurring task support
   - Event-driven scheduling with ticker-based dispatch
   
   ### New Features:
   1. ✅ Priority-based scheduling with integer priorities
   2. ✅ Task dependency management (`preorders`)
   3. ✅ Cron expressions for recurring tasks
   4. ✅ Exclusive/concurrent execution modes
   5. ✅ Configurable soft scheduling
   
   ---
   
   ## Critical Issues 🚨
   
   ### 1. TOCTTOU Race Condition (scheduler_bl.go:421-423)
   **Severity: CRITICAL**
   
   ```go
   // TODO: Is here need a lock? TOCTTOU
   if taskInfo.State != structure.TaskStateWaiting {
       logrus.Errorf("task state is not in 'Waiting' state, taskID: %v", taskInfo)
       return
   }
   ```
   
   The TODO correctly identifies a Time-Of-Check-Time-Of-Use vulnerability. State check and operations are not atomic.
   
   **Fix:**
   ```go
   defer taskInfo.Unlock(taskInfo.Lock())
   if taskInfo.State != structure.TaskStateWaiting {
       // ...
   }
   ```
   
   ### 2. Resource Leak: Channel Not Closed (scheduler_bl.go:62)
   **Severity: HIGH**
   
   The `startChan` is never closed, causing potential goroutine leaks during shutdown.
   
   **Fix:** Add cleanup:
   ```go
   func (s *ScheduleBl) Shutdown() {
       close(s.startChan)
   }
   ```
   
   ### 3. Dependency State Validation Missing (scheduler_bl.go:234-244)
   **Severity: HIGH**
   
   No check if dependency tasks are completed. Tasks could depend on failed/canceled tasks.
   
   **Fix:**
   ```go
   if depTask.State != structure.TaskStateComplete {
       return false, fmt.Errorf("dependency task %d is not complete (state: %s)", depTaskID, depTask.State)
   }
   ```
   
   ### 4. Potential Deadlock Pattern (scheduler_bl.go:97-99)
   **Severity: HIGH**
   
   Multiple functions acquire locks then call `TryScheduleNextTasks`, which conditionally locks. This pattern is deadlock-prone.
   
   **Fix:** Use consistent locking strategy, consider `sync.RWMutex` for read-heavy operations.
   
   ---
   
   ## High Priority Issues ⚠️
   
   ### 5. Nil Pointer Risk (task_bl.go:87-92)
   If `Scheduler.cronManager` is nil (initialization order issue), this will panic:
   ```go
   if err := Scheduler.cronManager.CheckCronExpression(cronExpr); err != nil {
   ```
   Add nil checks.
   
   ### 6. API Breaking Change (scheduler_bl.go:502)
   `CloseCurrent` signature changed from `(taskId int32)` to `(taskId int32, removeWorkerName ...string)`. Document this breaking change and verify all call sites are updated.
   
   ### 7. Commented Out Critical Code (scheduler_bl.go:229)
   ```go
   //defer s.Unlock(s.Lock())
   ```
   Critical locking code commented without explanation - this is dangerous.
   
   ### 8. Incomplete Implementation (scheduler_bl.go:159)
   ```go
   // TODO: NEED TO JUDGE IF THE TASK CAN CONCURRENTLY RUNNING
   // NOT only by user setting, but also by scheduler setting
   ```
   Important concurrent execution logic deferred.
   
   ---
   
   ## Medium Priority Issues
   
   ### 9. Missing Input Validation (task_bl.go:71-79)
   No upper bound on priority values. Large priorities could cause unexpected behavior.
   
   ### 10. Silent Error Handling (scheduler_bl.go:441-443)
   ```go
   if err := s.taskManager.AddTaskStartSequence(taskInfo.ID); err != nil {
       logrus.Errorf("failed to add task '%d' to start sequence: %v", taskInfo.ID, err)
   }
   ```
   Errors logged but not handled.
   
   ---
   
   ## CI/CD Status
   
   ### Failed Checks ❌
   1. **check-license-header** - Missing Apache license headers in new files
   2. **dependency-review** - New dependencies need review
   
   ### Passed Checks ✅
   - CodeQL (Go & Java)
   - Dependency license check
   
   ### Test Coverage ⚠️
   - PR mentions tests in `/test` folder but not visible in diff
   - Need unit tests for:
     - Priority scheduling algorithm
     - Dependency resolution
     - Cron task scheduling
     - Concurrent execution
   
   ---
   
   ## Recommendations
   
   ### Must Fix Before Merge:
   1. Fix TOCTTOU race condition with atomic state checking
   2. Add resource cleanup (close channels on shutdown)
   3. Validate dependency task states
   4. Fix failing CI checks (license headers, dependency review)
   5. Review and fix all locking patterns
   6. Add nil pointer checks for managers
   
   ### Should Fix:
   7. Complete TODO implementations (concurrent execution logic)
   8. Document API breaking changes
   9. Add comprehensive test coverage
   10. Add input validation (priority bounds)
   11. Remove commented code with explanation
   
   ### Documentation Needed:
   12. Migration guide for scheduler API changes
   13. Configuration documentation (`start_chan_size`, `ticker_interval`, `soft_schedule`)
   14. Cron expression format specification
   
   ---
   
   ## Positive Aspects 👍
   
   1. ✅ Well-architected component separation
   2. ✅ Good configuration externalization
   3. ✅ Comprehensive feature set
   4. ✅ Improved observability with detailed logging
   5. ✅ Added Docker support documentation
   
   ---
   
   ## Risk Assessment
   
   **Risk Level:** MEDIUM-HIGH due to concurrency issues in core scheduling logic
   
   **Estimated Time to Fix:** 2-3 days for critical fixes, 1 week for comprehensive improvements
   
   The architecture and features are excellent, but the critical concurrency issues must be resolved before this can be safely merged to production.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422650799


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +89,133 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+

Review Comment:
   **Potential deadlock in tryScheduleInner**
   
   The `noLock` parameter pattern is error-prone and can lead to deadlocks. When calling `TryScheduleNextTasks(true)` from within locked sections (like in `CloseCurrent` and `ChangeWorkerStatus`), you're relying on manual coordination to prevent deadlock.
   
   **Suggested improvements:**
   1. Split this into two methods: one that handles locking (`TryScheduleNextTasks`) and one that contains the core logic (`tryScheduleNextTasksNoLock`)
   2. Document which methods expect the caller to hold locks
   3. Consider using recursive mutexes if the design requires nested locking
   
   Example:
   ```go
   func (s *ScheduleBl) TryScheduleNextTasks() {
       defer s.Unlock(s.Lock())
       s.tryScheduleNextTasksNoLock(s.softSchedule)
   }
   
   func (s *ScheduleBl) tryScheduleNextTasksNoLock(softSchedule bool) error {
       // Implementation without locking
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422641675


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -202,6 +417,7 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 		}
 	}()
 
+	// TODO: Is here need a lock? TOCTTOU

Review Comment:
   **Race condition risk in task state checking**
   
   There's a TOCTTOU (Time-of-check to time-of-use) issue here. The comment indicates this concern but it should be addressed. Between checking the task state and starting the task, another goroutine could modify the state.
   
   **Suggested fix:**
   ```go
   // Acquire lock before checking and starting
   s.Lock()
   defer s.Unlock(nil)
   
   if taskInfo.State != structure.TaskStateWaiting {
       logrus.Errorf("task state is not in 'Waiting' state, taskID: %v", taskInfo)
       return
   }
   
   taskStarter, err := NewTaskStarter(taskInfo, agent)
   if err != nil {
       logrus.Errorf("failed to create new TaskStarter err: %v", err)
       taskMgr.SetError(taskInfo, err.Error())
       return
   }
   
   taskInfo.StartTime = time.Now()
   err = taskStarter.StartTask()
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422717203


##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +64,53 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {

Review Comment:
   **Integer overflow risk with priority values**
   
   The priority is parsed as int64 but stored as int32, which could cause overflow if users provide large values.
   
   **Suggested fix:**
   ```go
   if priority, ok := params["priority"]; ok {
       p, err := strconv.ParseInt(priority, 10, 32)
       if err != nil {
           logrus.Warnf("priority parse error: %v", err)
           return nil, fmt.Errorf("invalid priority value: %v", err)
       }
       if p < 0 {
           return nil, fmt.Errorf("priority must be non-negative, got: %d", p)
       }
       if p > math.MaxInt32 {
           return nil, fmt.Errorf("priority exceeds maximum value: %d", math.MaxInt32)
       }
       taskInfo.Priority = int32(p)
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444643895


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -222,68 +441,115 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 
 	taskInfo.StartTime = time.Now()
 	err = taskStarter.StartTask()
+
+	// only for test or debug, record the task start sequence

Review Comment:
   **缺少错误处理**:  的错误只是被记录,但没有适当的处理。如果这是关键功能,应该考虑是否需要回滚或重试。
   
   **建议**: 
   - 如果  是仅用于调试/测试,应该在注释中明确说明失败不会影响正常功能
   - 考虑在测试/调试模式下才调用此方法



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}

Review Comment:
   **逻辑错误**: 这里使用  添加到预分配的 slice,会导致前面的  创建的空元素被保留。
   
   **当前代码问题**:
   ```go
   errors := make([]error, len(taskInfos))  // 创建了 len 个 nil 元素
   oks := make([]bool, len(taskInfos))      // 创建了 len 个 false 元素
   // ... 循环中
   errors = append(errors, err)  // 会追加到末尾,导致返回 2*len 个元素
   oks = append(oks, ok)
   ```
   
   **修复方案**:
   ```go
   errors := make([]error, 0, len(taskInfos))  // 使用容量而不是长度
   oks := make([]bool, 0, len(taskInfos))
   ```
   
   或者使用索引赋值:
   ```go
   for i, taskInfo := range taskInfos {
       ok, err := s.QueueTask(taskInfo)
       errors[i] = err
       oks[i] = ok
   }
   ```



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+
+	// TODO: NEED TO JUDGE IF THE TASK CAN CONCURRENTLY RUNNING
+	// NOT only by user setting, but also by scheduler setting

Review Comment:
   **错误处理不一致**: 当  失败时,错误被记录但函数继续执行。这可能导致部分任务入队成功,部分失败,但调用者无法知道具体哪些失败了。
   
   **建议**:
   1. 在遇到第一个错误时考虑是否应该停止并回滚
   2. 或者返回更详细的错误信息,包括失败的任务 ID 列表



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)

Review Comment:
   **验证不足**: 依赖任务的验证只检查了任务是否存在,但没有检查:
   1. 依赖任务的状态(是否已完成、是否失败等)
   2. 是否存在循环依赖
   3. 依赖链的深度是否合理
   
   **建议**:
   ```go
   // 检查依赖任务状态
   if depTask.State == structure.TaskStateFailed {
       return false, fmt.Errorf("dependency task %d has failed", depTaskID)
   }
   
   // 检查循环依赖
   if s.taskManager.HasCircularDependency(taskInfo.ID, depTaskID) {
       return false, fmt.Errorf("circular dependency detected")
   }
   ```



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 

Review Comment:
   **配置值硬编码**: 这里使用字符串字面量 "3" 作为默认值,建议定义为常量。
   
   **建议**:
   ```go
   const (
       defaultStartChanSize = "10"
       defaultTickerInterval = "3"  // 秒
       defaultSoftSchedule = "true"
   )
   ```
   
   这样更易于维护和文档化。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444646140


##########
vermeer/apps/master/master_main.go:
##########
@@ -56,6 +56,7 @@ func Main() {
 		services.SetUI(sen)
 		logrus.Info("token-auth was activated")
 	default:

Review Comment:
   **代码风格**: 这里有一个空的  分支并且后面有多余的空行。应该删除或添加注释说明为什么需要空的 default 分支。
   
   **建议**: 
   ```go
   default:
       // No authentication required for other modes
   ```
   
   或者如果不需要 default 分支,直接删除。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444649751


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {

Review Comment:
   **死锁风险**: BatchQueueTask 调用 PauseDispatch 后在 defer 中调用 ResumeDispatch,但被注释掉的锁代码表明这里可能存在锁的设计问题。
   
   **问题**:
   1. 注释掉的 defer s.Unlock(s.Lock()) 需要解释原因
   2. 如果在 PauseDispatch 和 ResumeDispatch 之间发生 panic,可能导致调度器永久暂停
   3. PauseDispatch 和 ResumeDispatch 的实现可能不是线程安全的
   
   **建议**:
   1. 添加注释说明为什么这行锁被注释掉
   2. 确保 PauseDispatch 和 ResumeDispatch 是线程安全的
   3. 使用更明确的锁管理策略
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444647199


##########
vermeer/apps/master/bl/task_creator.go:
##########
@@ -99,6 +99,28 @@ func (bc *baseCreator) NewTaskInfo(graphName string, params map[string]string, t
 	return task, nil
 }
 
+func (bc *baseCreator) CopyTaskInfo(src *structure.TaskInfo) (*structure.TaskInfo, error) {

Review Comment:
   **测试覆盖率**: 这个函数复制了任务信息,但清除了 cron 表达式。这个行为需要单元测试来确保:
   1. 所有字段都被正确复制
   2. cron 表达式确实被清除
   3. 新任务 ID 是唯一的
   
   **建议**: 添加单元测试:
   ```go
   func TestCopyTaskInfo(t *testing.T) {
       // Test case 1: nil input
       // Test case 2: all fields copied correctly
       // Test case 3: cron expression cleared
       // Test case 4: new task ID generated
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447144298


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 
-	go s.dispatch()
+	if s.cronManager.CheckCronExpression(taskInfo.CronExpr) == nil {
+		if err := s.cronManager.AddCronTask(taskInfo); err != nil {
+			logrus.Errorf("failed to add cron task: %v", err)
+			return false, err
+		}
+		logrus.Infof("added cron task for task '%d' with expression '%s'", taskInfo.ID, taskInfo.CronExpr)
+	}
 
 	return ok, nil
 }
 
-func (s *ScheduleBl) CancelTask(taskInfo *structure.TaskInfo) error {
-	if taskInfo == nil {
-		return errors.New("the argument `taskInfo` is nil")
+/*
+* @Description: QueueTaskFromTemplate queues the task from the template.
+* @Note: This function will queue the task from the template. This function is used by cron tasks.
+* @Param template
+* @Return int32, error
+ */
+func (s *ScheduleBl) QueueTaskFromTemplate(template *structure.TaskInfo) (int32, error) {
+	if template == nil {
+		return -1, errors.New("the argument `template` is nil")
 	}
 
-	s.Lock()
-	isHeadTask := s.spaceQueue.IsHeadTask(taskInfo.ID)
-	task := s.spaceQueue.RemoveTask(taskInfo.ID)
-	s.Unlock(nil)
+	bc := &baseCreator{}
+	taskInfo, err := bc.CopyTaskInfo(template)
+	if err != nil {
+		logrus.Errorf("failed to copy task info from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
+	}
+	bc.saveTaskInfo(taskInfo)
 
-	isInQueue := false
-	if task != nil {
-		logrus.Infof("removed task '%d' from space queue", task.ID)
-		isInQueue = true
+	ok, err := s.QueueTask(taskInfo)
+	if err != nil || !ok {
+		logrus.Errorf("failed to queue task from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
 	}
 
-	if isInQueue && !isHeadTask {
-		if err := taskMgr.SetState(taskInfo, structure.TaskStateCanceled); err != nil {
-			return err
-		}
+	logrus.Infof("queued task '%d' from template '%d'", taskInfo.ID, template.ID)
 
-		logrus.Infof("set task '%d' to TaskStateCanceled", taskInfo.ID)
-	} else {
-		logrus.Infof("sending task '%d' to task canceler", taskInfo.ID)
-		return s.handleCancelTask(taskInfo)
+	return taskInfo.ID, nil
+}
+
+/*
+* @Description: BatchQueueTask batches the task.
+* @Note: This function will batch the task.
+* @Param taskInfos
+* @Return []bool, []error
+ */
+func (s *ScheduleBl) BatchQueueTask(taskInfos []*structure.TaskInfo) ([]bool, []error) {
+	if len(taskInfos) == 0 {
+		return []bool{}, []error{}
 	}
 
-	return nil
-}
+	s.PauseDispatch()
 
-func (s *ScheduleBl) IsDispatchPaused() bool {
-	return s.isDispatchPaused
-}
-func (s *ScheduleBl) PauseDispatch() {
-	s.isDispatchPaused = true
-}
+	defer s.ResumeDispatch()
+	// defer s.Unlock(s.Lock())
 
-func (s *ScheduleBl) ResumeDispatch() {
-	s.isDispatchPaused = false
+	errors := make([]error, len(taskInfos))
+	oks := make([]bool, len(taskInfos))
+
+	for _, taskInfo := range taskInfos {
+		ok, err := s.QueueTask(taskInfo)
+		if err != nil {
+			logrus.Errorf("failed to queue task '%d': %v", taskInfo.ID, err)
+		}
+		errors = append(errors, err)
+		oks = append(oks, ok)
+	}
+
+	return oks, errors
 }
 
-func (s *ScheduleBl) AllTasksInQueue() []*structure.TaskInfo {
-	return s.spaceQueue.AllTasks()
+// ******** CloseCurrent ********
+func (s *ScheduleBl) CloseCurrent(taskId int32, removeWorkerName ...string) error {
+	defer s.Unlock(s.Lock())
+
+	// trace tasks need these workers, check if these tasks are available
+	s.taskManager.RemoveTask(taskId)
+	// release the worker group
+	s.resourceManager.ReleaseByTaskID(taskId)
+
+	if len(removeWorkerName) > 0 {
+		// stop the cron job if exists when need remove worker, otherwise the task is just closed normally
+		s.cronManager.DeleteTask(taskId)
+		// remove the worker from resource manager
+		workerName := removeWorkerName[0]
+		if workerName == "" {
+			return errors.New("the argument `removeWorkerName` is empty")
+		}
+		logrus.Infof("removing worker '%s' from resource manager", workerName)
+		s.ChangeWorkerStatus(workerName, schedules.WorkerOngoingStatusDeleted)
+	}
+
+	logrus.Infof("invoke dispatch when task '%d' is closed", taskId)
+	s.TryScheduleNextTasks(true)
+	return nil
 }
 
-func (s *ScheduleBl) TasksInQueue(space string) []*structure.TaskInfo {
-	return s.spaceQueue.SpaceTasks(space)
+// This will be called when a worker is offline.
+// This will be called when a worker is online.
+func (s *ScheduleBl) ChangeWorkerStatus(workerName string, status schedules.WorkerOngoingStatus) (bool, error) {
+	defer s.Unlock(s.Lock())
+	s.resourceManager.ChangeWorkerStatus(workerName, status)
+
+	logrus.Infof("worker '%s' status changed to '%s'", workerName, status)
+	// After changing the worker status, we may need to reschedule tasks
+	s.TryScheduleNextTasks(true)
+
+	return true, nil
 }

Review Comment:
   handleStartTask only checks and calls s.startWaitingTask(agent, taskInfo) using a goroutine. So there is no problem.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447321431


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 

Review Comment:
   I think it would be easier to understand to define within the function rather than globally



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422717534


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -222,68 +438,115 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 
 	taskInfo.StartTime = time.Now()
 	err = taskStarter.StartTask()
+
+	// only for test or debug, record the task start sequence
+	if err := s.taskManager.AddTaskStartSequence(taskInfo.ID); err != nil {
+		logrus.Errorf("failed to add task '%d' to start sequence: %v", taskInfo.ID, err)
+	}
+
 	if err != nil {
 		logrus.Errorf("failed to start a task, type: %s, taskID: %d, caused by: %v", taskInfo.Type, taskInfo.ID, err)
 		taskMgr.SetError(taskInfo, err.Error())
 	}
-
 }
 
-func (s *ScheduleBl) dispatch() {
-	defer func() {
-		if err := recover(); err != nil {
-			logrus.Errorln("dispatch() has been recovered:", err)
+// ********* CANCEL TASK ********
+// handle cancel task
+// need to cancel cron task
+func (s *ScheduleBl) CancelTask(taskInfo *structure.TaskInfo) error {
+	if taskInfo == nil {
+		return errors.New("the argument `taskInfo` is nil")
+	}
+
+	defer s.Unlock(s.Lock())
+
+	isHeadTask := s.taskManager.IsTaskOngoing(taskInfo.ID)
+	task := s.taskManager.RemoveTask(taskInfo.ID)
+	s.cronManager.DeleteTask(taskInfo.ID)
+	// err := s.taskManager.CancelTask(taskInfo)
+	isInQueue := false
+	if task != nil {
+		logrus.Infof("removed task '%d' from space queue", taskInfo.ID)
+		isInQueue = true
+	}
+
+	if isInQueue && !isHeadTask {
+		if err := taskMgr.SetState(taskInfo, structure.TaskStateCanceled); err != nil {
+			return err
 		}
-	}()
 
-	if err := s.doDispatch(); err != nil {
-		logrus.Errorf("do dispatching error:%v", err)
+		logrus.Infof("set task '%d' to TaskStateCanceled", taskInfo.ID)
+	} else {
+		logrus.Infof("sending task '%d' to task canceler", taskInfo.ID)
+		return s.handleCancelTask(taskInfo)
 	}
+
+	return nil
 }
 
-func (s *ScheduleBl) doDispatch() error {
-	if s.isDispatchPaused {
-		logrus.Warn("the dispatching was paused")
-		return nil
+func (s *ScheduleBl) handleCancelTask(taskInfo *structure.TaskInfo) error {
+	logrus.Infof("received task '%d' to cancel", taskInfo.ID)
+	canceler, err := NewTaskCanceler(taskInfo)
+	if err != nil {
+		logrus.Errorf("failed to create new TaskCanceler err: %v", err)
+		taskMgr.SetError(taskInfo, err.Error())
+		return err
 	}
 
-	defer s.dispatchLocker.Unlock(s.dispatchLocker.Lock())
-
-	buffer := s.spaceQueue.HeadTasks()
-	if len(buffer) == 0 {
-		return nil
+	if err := canceler.CancelTask(); err != nil {
+		logrus.Errorf("failed to cancel task '%d', caused by: %v", taskInfo.ID, err)
+		taskMgr.SetError(taskInfo, err.Error())
+		return err
 	}
 
-	for _, task := range buffer {
-		select {
-		case s.startChan <- task:
-		default:
-			logrus.Warnf("the start channel is full, dropped task: %d", task.ID)
-		}
+	// set worker state to idle or concurrent running
+	s.resourceManager.ReleaseByTaskID(taskInfo.ID)
 
+	return nil
+}
+
+func (s *ScheduleBl) CancelCronTask(taskInfo *structure.TaskInfo) error {
+	if taskInfo == nil {
+		return errors.New("the argument `taskInfo` is nil")
 	}
 
+	s.cronManager.DeleteTask(taskInfo.ID)
+
 	return nil
 }
 
-func (s *ScheduleBl) waitingTask() {
-	for taskInfo := range s.startChan {
-		if taskInfo == nil {
-			logrus.Warnf("recieved a nil task from startChan")
-			return
-		}
+// ** Other Methods **
 
-		logrus.Infof("chan received task '%d' to start", taskInfo.ID)
-		s.handleStartTask(taskInfo)
-	}
+func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
+	return s.taskManager.GetLastTask(space)
 }
 
-func (s *ScheduleBl) startTicker() {
-	// Create a ticker that triggers every 3 seconds
-	ticker := time.Tick(3 * time.Second)
+func (s *ScheduleBl) IsDispatchPaused() bool {
+	// Implement logic to check if dispatching is paused
+	return s.algorithmManager.IsDispatchPaused()
+}
 
-	for range ticker {
-		//logrus.Debug("Ticker ticked")
-		s.dispatch()
-	}
+func (s *ScheduleBl) PauseDispatch() {
+	// Implement logic to pause dispatching
+	s.algorithmManager.PauseDispatch()
+}
+
+func (s *ScheduleBl) ResumeDispatch() {
+	// Implement logic to resume dispatching
+	s.algorithmManager.ResumeDispatch()
+}
+
+func (s *ScheduleBl) AllTasksInQueue() []*structure.TaskInfo {
+	// Implement logic to get all tasks in the queue
+	return s.taskManager.GetAllTasks()
+}
+
+func (s *ScheduleBl) TasksInQueue(space string) []*structure.TaskInfo {
+	// Implement logic to get tasks in the queue for a specific space
+	return s.taskManager.GetTasksInQueue(space)
+}

Review Comment:
   **Debugging code left in production**
   
   The `TaskStartSequence` functionality appears to be for debugging/testing only but is included in the production code without proper conditional compilation or feature flags.
   
   **Suggested improvements:**
   1. Move to a separate debug build with build tags:
   ```go
   // +build debug
   
   func (s *ScheduleBl) TaskStartSequence(queryTasks []int32) []*structure.TaskInfo {
       return s.taskManager.GetTaskStartSequence(queryTasks)
   }
   ```
   
   2. Or add a runtime flag check:
   ```go
   func (s *ScheduleBl) TaskStartSequence(queryTasks []int32) []*structure.TaskInfo {
       if !common.GetConfigDefault("enable_debug_features", "false").(bool) {
           logrus.Warn("TaskStartSequence called but debug features are disabled")
           return nil
       }
       return s.taskManager.GetTaskStartSequence(queryTasks)
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422716580


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +89,133 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+

Review Comment:
   **Commented TODO needs resolution**
   
   This TODO indicates an important design decision that's unresolved. The scheduler needs to determine if tasks can run concurrently based on both user settings AND scheduler policies.
   
   **Action required:**
   1. Either implement the logic to check concurrent execution capability
   2. Create a tracking issue for this work and reference it in the comment
   3. Document the temporary behavior and its limitations
   
   Example:
   ```go
   // TODO(issue #XXX): Implement comprehensive concurrent execution check
   // Currently only checks user settings. Need to add:
   // 1. Resource availability check
   // 2. Conflict detection with running tasks
   // 3. Worker group capacity limits
   // Temporary behavior: Tasks marked as non-exclusive may still be blocked
   // if resources are insufficient
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422717374


##########
vermeer/apps/master/bl/task_creator.go:
##########
@@ -99,6 +99,28 @@ func (bc *baseCreator) NewTaskInfo(graphName string, params map[string]string, t
 	return task, nil
 }
 
+func (bc *baseCreator) CopyTaskInfo(src *structure.TaskInfo) (*structure.TaskInfo, error) {

Review Comment:
   **Poor error handling in copy operation**
   
   When copying task info, if any error occurs after task creation, the partially created task is left in the system without cleanup.
   
   **Suggested fix:**
   ```go
   func (bc *baseCreator) CopyTaskInfo(src *structure.TaskInfo) (*structure.TaskInfo, error) {
       if src == nil {
           return nil, fmt.Errorf("the argument 'src' should not be nil")
       }
   
       task, err := taskMgr.CreateTask(src.SpaceName, src.Type, 0)
       if err != nil {
           return nil, err
       }
   
       // Ensure cleanup on error
       success := false
       defer func() {
           if !success {
               if cleanupErr := taskMgr.RemoveTask(task.ID); cleanupErr != nil {
                   logrus.Errorf("failed to cleanup task %d after copy error: %v", task.ID, cleanupErr)
               }
           }
       }()
   
       task.CreateType = structure.TaskCreateAsync
       task.GraphName = src.GraphName
       task.CreateUser = src.CreateUser
       task.Params = src.Params
       task.CronExpr = "" // clear cron expression for the new task
       task.Priority = src.Priority
       task.Preorders = src.Preorders
       task.Exclusive = src.Exclusive
   
       success = true
       return task, nil
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#issuecomment-3393063945

   ## Summary
   This PR introduces a comprehensive scheduler redesign with priority-based scheduling, task dependencies, and cron support. The architecture is well-designed, but there are **critical concurrency issues** that must be addressed before merging.
   
   **Verdict: REQUEST CHANGES** ⚠️
   
   ---
   
   ## Architecture Changes
   
   ### Major Refactoring ✅
   The PR completely redesigns the scheduler from a simple queue-based system to a sophisticated multi-component architecture:
   
   **Old Architecture:**
   - Simple `SpaceQueue` (FIFO)
   - Basic `Broker` for worker assignment
   - Single-threaded dispatch loop
   
   **New Architecture:**
   - `SchedulerResourceManager` - Worker pool management
   - `SchedulerTaskManager` - Task lifecycle, dependencies, priorities
   - `SchedulerAlgorithmManager` - Scheduling algorithm logic
   - `SchedulerCronManager` - Recurring task support
   - Event-driven scheduling with ticker-based dispatch
   
   ### New Features:
   1. ✅ Priority-based scheduling with integer priorities
   2. ✅ Task dependency management (`preorders`)
   3. ✅ Cron expressions for recurring tasks
   4. ✅ Exclusive/concurrent execution modes
   5. ✅ Configurable soft scheduling
   
   ---
   
   ## Critical Issues 🚨
   
   ### 1. TOCTTOU Race Condition (scheduler_bl.go:421-423)
   **Severity: CRITICAL**
   
   ```go
   // TODO: Is here need a lock? TOCTTOU
   if taskInfo.State != structure.TaskStateWaiting {
       logrus.Errorf("task state is not in 'Waiting' state, taskID: %v", taskInfo)
       return
   }
   ```
   
   The TODO correctly identifies a Time-Of-Check-Time-Of-Use vulnerability. State check and operations are not atomic.
   
   **Fix:**
   ```go
   defer taskInfo.Unlock(taskInfo.Lock())
   if taskInfo.State != structure.TaskStateWaiting {
       // ...
   }
   ```
   
   ### 2. Resource Leak: Channel Not Closed (scheduler_bl.go:62)
   **Severity: HIGH**
   
   The `startChan` is never closed, causing potential goroutine leaks during shutdown.
   
   **Fix:** Add cleanup:
   ```go
   func (s *ScheduleBl) Shutdown() {
       close(s.startChan)
   }
   ```
   
   ### 3. Dependency State Validation Missing (scheduler_bl.go:234-244)
   **Severity: HIGH**
   
   No check if dependency tasks are completed. Tasks could depend on failed/canceled tasks.
   
   **Fix:**
   ```go
   if depTask.State != structure.TaskStateComplete {
       return false, fmt.Errorf("dependency task %d is not complete (state: %s)", depTaskID, depTask.State)
   }
   ```
   
   ### 4. Potential Deadlock Pattern (scheduler_bl.go:97-99)
   **Severity: HIGH**
   
   Multiple functions acquire locks then call `TryScheduleNextTasks`, which conditionally locks. This pattern is deadlock-prone.
   
   **Fix:** Use consistent locking strategy, consider `sync.RWMutex` for read-heavy operations.
   
   ---
   
   ## High Priority Issues ⚠️
   
   ### 5. Nil Pointer Risk (task_bl.go:87-92)
   If `Scheduler.cronManager` is nil (initialization order issue), this will panic:
   ```go
   if err := Scheduler.cronManager.CheckCronExpression(cronExpr); err != nil {
   ```
   Add nil checks.
   
   ### 6. API Breaking Change (scheduler_bl.go:502)
   `CloseCurrent` signature changed from `(taskId int32)` to `(taskId int32, removeWorkerName ...string)`. Document this breaking change and verify all call sites are updated.
   
   ### 7. Commented Out Critical Code (scheduler_bl.go:229)
   ```go
   //defer s.Unlock(s.Lock())
   ```
   Critical locking code commented without explanation - this is dangerous.
   
   ### 8. Incomplete Implementation (scheduler_bl.go:159)
   ```go
   // TODO: NEED TO JUDGE IF THE TASK CAN CONCURRENTLY RUNNING
   // NOT only by user setting, but also by scheduler setting
   ```
   Important concurrent execution logic deferred.
   
   ---
   
   ## Medium Priority Issues
   
   ### 9. Missing Input Validation (task_bl.go:71-79)
   No upper bound on priority values. Large priorities could cause unexpected behavior.
   
   ### 10. Silent Error Handling (scheduler_bl.go:441-443)
   ```go
   if err := s.taskManager.AddTaskStartSequence(taskInfo.ID); err != nil {
       logrus.Errorf("failed to add task '%d' to start sequence: %v", taskInfo.ID, err)
   }
   ```
   Errors logged but not handled.
   
   ---
   
   ## CI/CD Status
   
   ### Failed Checks ❌
   1. **check-license-header** - Missing Apache license headers in new files
   2. **dependency-review** - New dependencies need review
   
   ### Passed Checks ✅
   - CodeQL (Go & Java)
   - Dependency license check
   
   ### Test Coverage ⚠️
   - PR mentions tests in `/test` folder but not visible in diff
   - Need unit tests for:
     - Priority scheduling algorithm
     - Dependency resolution
     - Cron task scheduling
     - Concurrent execution
   
   ---
   
   ## Recommendations
   
   ### Must Fix Before Merge:
   1. Fix TOCTTOU race condition with atomic state checking
   2. Add resource cleanup (close channels on shutdown)
   3. Validate dependency task states
   4. Fix failing CI checks (license headers, dependency review)
   5. Review and fix all locking patterns
   6. Add nil pointer checks for managers
   
   ### Should Fix:
   7. Complete TODO implementations (concurrent execution logic)
   8. Document API breaking changes
   9. Add comprehensive test coverage
   10. Add input validation (priority bounds)
   11. Remove commented code with explanation
   
   ### Documentation Needed:
   12. Migration guide for scheduler API changes
   13. Configuration documentation (`start_chan_size`, `ticker_interval`, `soft_schedule`)
   14. Cron expression format specification
   
   ---
   
   ## Positive Aspects 👍
   
   1. ✅ Well-architected component separation
   2. ✅ Good configuration externalization
   3. ✅ Comprehensive feature set
   4. ✅ Improved observability with detailed logging
   5. ✅ Added Docker support documentation
   
   ---
   
   ## Risk Assessment
   
   **Risk Level:** MEDIUM-HIGH due to concurrency issues in core scheduling logic
   
   **Estimated Time to Fix:** 2-3 days for critical fixes, 1 week for comprehensive improvements
   
   The architecture and features are excellent, but the critical concurrency issues must be resolved before this can be safely merged to production.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422693337


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +225,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+

Review Comment:
   **Incomplete validation of preorder dependencies**
   
   The code checks if the dependency task exists, but doesn't verify:
   1. Whether the dependency task is in a valid state (completed, running, waiting)
   2. Circular dependency detection
   3. Whether dependencies are in the same space/worker group
   
   **Suggested improvements:**
   ```go
   if len(taskInfo.Preorders) > 0 {
       visited := make(map[int32]bool)
       for _, depTaskID := range taskInfo.Preorders {
           // Check existence
           depTask := taskMgr.GetTaskByID(depTaskID)
           if depTask == nil {
               err := fmt.Errorf("dependency task with ID %d does not exist", depTaskID)
               logrus.Error(err)
               taskMgr.SetError(taskInfo, err.Error())
               return false, err
           }
           
           // Check for circular dependencies
           if err := s.checkCircularDependency(taskInfo.ID, depTaskID, visited); err \!= nil {
               logrus.Error(err)
               taskMgr.SetError(taskInfo, err.Error())
               return false, err
           }
           
           // Validate dependency state
           if depTask.State == structure.TaskStateError || depTask.State == structure.TaskStateCanceled {
               err := fmt.Errorf("dependency task %d is in invalid state: %s", depTaskID, depTask.State)
               logrus.Error(err)
               taskMgr.SetError(taskInfo, err.Error())
               return false, err
           }
       }
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422715999


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +225,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)
+
+	// check dependency if exists
+	if len(taskInfo.Preorders) > 0 {
+		for _, depTaskID := range taskInfo.Preorders {
+			depTask := taskMgr.GetTaskByID(depTaskID)
+			if depTask == nil {
+				err := errors.New("the dependency task with ID " + strconv.Itoa(int(depTaskID)) + " does not exist")
+				logrus.Error(err)
+				taskMgr.SetError(taskInfo, err.Error())
+				return false, err
+			}
+		}
+	}
+
 	// Notice: Ensure successful invocation.
-	ok, err := s.spaceQueue.PushTask(taskInfo)
+	// make sure all tasks have alloc to a worker group
+	ok, err := s.taskManager.QueueTask(taskInfo)
 	if err != nil {
 		taskMgr.SetError(taskInfo, err.Error())
 		return ok, err
 	}
 
-	go s.dispatch()
+	if s.cronManager.CheckCronExpression(taskInfo.CronExpr) == nil {
+		if err := s.cronManager.AddCronTask(taskInfo); err != nil {
+			logrus.Errorf("failed to add cron task: %v", err)
+			return false, err
+		}
+		logrus.Infof("added cron task for task '%d' with expression '%s'", taskInfo.ID, taskInfo.CronExpr)
+	}
 
 	return ok, nil
 }
 
-func (s *ScheduleBl) CancelTask(taskInfo *structure.TaskInfo) error {
-	if taskInfo == nil {
-		return errors.New("the argument `taskInfo` is nil")
+/*
+* @Description: QueueTaskFromTemplate queues the task from the template.
+* @Note: This function will queue the task from the template. This function is used by cron tasks.
+* @Param template
+* @Return int32, error
+ */
+func (s *ScheduleBl) QueueTaskFromTemplate(template *structure.TaskInfo) (int32, error) {
+	if template == nil {
+		return -1, errors.New("the argument `template` is nil")
 	}
 
-	s.Lock()
-	isHeadTask := s.spaceQueue.IsHeadTask(taskInfo.ID)
-	task := s.spaceQueue.RemoveTask(taskInfo.ID)
-	s.Unlock(nil)
+	bc := &baseCreator{}
+	taskInfo, err := bc.CopyTaskInfo(template)
+	if err != nil {
+		logrus.Errorf("failed to copy task info from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
+	}
+	bc.saveTaskInfo(taskInfo)
 
-	isInQueue := false
-	if task != nil {
-		logrus.Infof("removed task '%d' from space queue", task.ID)
-		isInQueue = true
+	ok, err := s.QueueTask(taskInfo)
+	if err != nil || !ok {
+		logrus.Errorf("failed to queue task from template, template ID: %d, caused by: %v", template.ID, err)
+		return -1, err
 	}
 
-	if isInQueue && !isHeadTask {
-		if err := taskMgr.SetState(taskInfo, structure.TaskStateCanceled); err != nil {
-			return err
-		}
+	logrus.Infof("queued task '%d' from template '%d'", taskInfo.ID, template.ID)
 
-		logrus.Infof("set task '%d' to TaskStateCanceled", taskInfo.ID)
-	} else {
-		logrus.Infof("sending task '%d' to task canceler", taskInfo.ID)
-		return s.handleCancelTask(taskInfo)
+	return taskInfo.ID, nil
+}
+
+/*
+* @Description: BatchQueueTask batches the task.
+* @Note: This function will batch the task.
+* @Param taskInfos
+* @Return []bool, []error
+ */
+func (s *ScheduleBl) BatchQueueTask(taskInfos []*structure.TaskInfo) ([]bool, []error) {
+	if len(taskInfos) == 0 {
+		return []bool{}, []error{}
 	}
 
-	return nil
-}
+	s.PauseDispatch()
 

Review Comment:
   **Silent failures in batch operations**
   
   The batch operation returns arrays of results but there's no clear indication of which operations failed. This makes error handling difficult for callers.
   
   **Suggested improvement:**
   ```go
   type BatchQueueResult struct {
       TaskID int32
       Success bool
       Error error
   }
   
   func (s *ScheduleBl) BatchQueueTask(taskInfos []*structure.TaskInfo) []*BatchQueueResult {
       if len(taskInfos) == 0 {
           return []*BatchQueueResult{}
       }
   
       s.PauseDispatch()
       defer s.ResumeDispatch()
   
       results := make([]*BatchQueueResult, 0, len(taskInfos))
       for _, taskInfo := range taskInfos {
           ok, err := s.QueueTask(taskInfo)
           results = append(results, &BatchQueueResult{
               TaskID: taskInfo.ID,
               Success: ok,
               Error: err,
           })
           if err != nil {
               logrus.Errorf("failed to queue task '%d': %v", taskInfo.ID, err)
           }
       }
   
       return results
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2422717301


##########
vermeer/apps/master/master_main.go:
##########
@@ -56,6 +56,7 @@ func Main() {
 		services.SetUI(sen)
 		logrus.Info("token-auth was activated")
 	default:
+		services.SetAdminRouters(sen, auth.NoneAuthFilter)

Review Comment:
   **Admin routes exposed without authentication**
   
   Setting admin routes with `NoneAuthFilter` in the default case is a critical security issue. Admin functionality should never be exposed without authentication.
   
   **Suggested fix:**
   ```go
   default:
       // DO NOT expose admin routes without authentication
       // services.SetAdminRouters(sen, auth.NoneAuthFilter)  // REMOVED
       services.SetRouters(sen, auth.NoneAuthFilter)
       logrus.Warn("No authentication was activated. ADMIN ROUTES DISABLED for security.")
       logrus.Warn("Please configure proper authentication to enable admin functionality.")
   }
   ```
   
   If admin routes must be available for development:
   1. Add a separate dev-mode flag
   2. Log prominent warnings
   3. Bind only to localhost
   4. Document security implications clearly



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444644762


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+
+	// TODO: NEED TO JUDGE IF THE TASK CAN CONCURRENTLY RUNNING
+	// NOT only by user setting, but also by scheduler setting

Review Comment:
   **错误处理不一致**: 当  失败时,错误被记录但函数继续执行。这可能导致部分任务入队成功,部分失败,但调用者无法知道具体哪些失败了。
   
   **建议**:
   1. 在遇到第一个错误时考虑是否应该停止并回滚
   2. 或者返回更详细的错误信息,包括失败的任务 ID 列表



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444646515


##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {

Review Comment:
   **类型转换安全性**: 虽然已经检查了 ,但还应该检查转换后的值是否会溢出。
   
   **建议**:
   ```go
   if p < 0 || p > math.MaxInt32 {
       return nil, fmt.Errorf("priority must be between 0 and %d, got %d", math.MaxInt32, p)
   }
   ```
   
   另外,考虑直接使用  的第三个参数来限制位数,这样就不需要额外检查了。



##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {
+					return nil, fmt.Errorf("priority should be non-negative")
+				}
+				if p > math.MaxInt32 {
+					return nil, fmt.Errorf("priority exceeds maximum value: %d", math.MaxInt32)
+				}
+				taskInfo.Priority = int32(p)
+			} else {
+				logrus.Warnf("priority convert to int32 error:%v", err)
+				return nil, err
+			}
+		}
+		if preorders, ok := params["preorders"]; ok {
+			preorderList := strings.Split(preorders, ",")
+			for _, preorder := range preorderList {
+				if pid, err := strconv.ParseInt(preorder, 10, 32); err == nil {
+					if taskMgr.GetTaskByID(int32(pid)) == nil {

Review Comment:
   **错误消息国际化**: 错误消息使用了中文和英文混合。建议统一使用英文,或者实现完整的国际化支持。
   
   **示例**:
   ```go
   // 当前
   return nil, fmt.Errorf("preorder task with ID %d does not exist", pid)
   
   // 建议保持一致的风格
   ```
   
   这对于开源项目尤其重要,因为国际贡献者可能不懂中文。



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2444660211


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+
+	// TODO: NEED TO JUDGE IF THE TASK CAN CONCURRENTLY RUNNING
+	// NOT only by user setting, but also by scheduler setting

Review Comment:
   **错误处理不一致**: 当 QueueTask 失败时,错误被记录但函数继续执行。这可能导致部分任务入队成功,部分失败,但调用者无法知道具体哪些失败了。
   
   **建议**:
   1. 在遇到第一个错误时考虑是否应该停止并回滚已入队的任务
   2. 或者至少在返回值中明确标识哪些任务失败了
   3. 使用事务性的批量操作来保证原子性
   
   **示例改进**:
   ```go
   // 记录成功和失败的任务
   successTasks := make([]*structure.TaskInfo, 0, len(taskInfos))
   for _, taskInfo := range taskInfos {
       ok, err := s.QueueTask(taskInfo)
       if err != nil {
           // 回滚已成功的任务
           for _, t := range successTasks {
               s.taskManager.RemoveTask(t.ID)
           }
           return false, fmt.Errorf("failed to queue task %d: %w", taskInfo.ID, err)
       }
       successTasks = append(successTasks, taskInfo)
   }
   ```



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -72,81 +228,161 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 		return false, errors.New("the property `SpaceName` of taskInfo is empty")
 	}
 
-	//defer s.Unlock(s.Lock())
+	defer s.Unlock(s.Lock())
 	if err := taskMgr.SetState(taskInfo, structure.TaskStateWaiting); err != nil {
 		return false, err
 	}
 
+	logrus.Debugf("queuing task %d with parameters: %+v", taskInfo.ID, taskInfo)

Review Comment:
   **验证不足**: 依赖任务的验证只检查了任务是否存在,但没有检查:
   1. 依赖任务的状态(是否已完成、是否失败等)
   2. 是否存在循环依赖
   3. 依赖链的深度是否合理
   
   **建议**:
   ```go
   for _, depTaskID := range taskInfo.Preorders {
       depTask := taskMgr.GetTaskByID(depTaskID)
       if depTask == nil {
           return false, fmt.Errorf("dependency task %d does not exist", depTaskID)
       }
   
       // 检查依赖任务状态
       if depTask.State == structure.TaskStateFailed {
           return false, fmt.Errorf("dependency task %d has failed", depTaskID)
       }
   
       // 检查循环依赖
       if hasCircularDependency(taskInfo.ID, depTaskID) {
           return false, errors.New("circular dependency detected")
       }
   
       // 检查 space 一致性
       if depTask.SpaceName != taskInfo.SpaceName {
           return false, errors.New("dependency task must be in same space")
       }
   }
   ```



##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +90,135 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 

Review Comment:
   **配置值硬编码**: 这里使用字符串字面量 "3" 作为默认值,建议定义为常量。
   
   **建议**:
   ```go
   const (
       defaultStartChanSize = "10"
       defaultMaxRunningTasks = "3"
       defaultCheckWaitingInterval = "5"
   )
   
   // 然后在使用时
   maxRunningTasks := options.GetInt(s.Name+".max_running_tasks", defaultMaxRunningTasks)
   ```
   
   这样可以:
   1. 提高代码可维护性
   2. 避免魔法数字
   3. 便于统一修改默认值



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447325475


##########
vermeer/apps/master/bl/scheduler_bl.go:
##########
@@ -47,22 +89,133 @@ func (s *ScheduleBl) Init() {
 		logrus.Infof("using default start_chan_size: %s", defaultChanSizeConfig)
 		chanSizeInt, _ = strconv.Atoi(defaultChanSizeConfig)
 	}
-	startChan := make(chan *structure.TaskInfo, chanSizeInt)
-	s.startChan = startChan
-	s.spaceQueue = (&schedules.SpaceQueue{}).Init()
-	s.broker = (&schedules.Broker{}).Init()
+	s.startChanSize = chanSizeInt
 
-	go s.waitingTask()
-	go s.startTicker()
+	// tickerInterval
+	const defaultTickerInterval = "3"
+	tickerInterval := common.GetConfigDefault("ticker_interval", defaultTickerInterval).(string)
+	tickerIntervalInt, err := strconv.Atoi(tickerInterval)
+	if err != nil {
+		logrus.Errorf("failed to convert ticker_interval to int: %v", err)
+		logrus.Infof("using default ticker_interval: %s", defaultTickerInterval)
+		tickerIntervalInt, _ = strconv.Atoi(defaultTickerInterval)
+	}
+	s.tickerInterval = tickerIntervalInt
+
+	// softSchedule
+	softSchedule := common.GetConfigDefault("soft_schedule", "true").(string)
+	if softSchedule == "true" {
+		s.softSchedule = true
+	} else {
+		s.softSchedule = false
+	}
+
+	logrus.Infof("ScheduleBl configuration: startChanSize=%d, tickerInterval=%d, softSchedule=%v",
+		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
-func (s *ScheduleBl) PeekSpaceTail(space string) *structure.TaskInfo {
-	return s.spaceQueue.PeekTailTask(space)
+/*
+* @Description: startTicker starts the ticker.
+* @Note: This function will start the ticker.
+ */
+func (s *ScheduleBl) startTicker() {
+	// Create a ticker with the specified interval
+	ticker := time.Tick(time.Duration(s.tickerInterval) * time.Second)
+
+	for range ticker {
+		logrus.Debug("Ticker ticked")
+		s.TryScheduleNextTasks()
+	}
+}
+
+// this make scheduler manager try to schedule next tasks
+/*
+* @Description: TryScheduleNextTasks tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param noLock
+ */
+func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorln("TryScheduleNextTasks() has been recovered:", err)
+		}
+	}()
+
+	if err := s.tryScheduleInner(s.softSchedule, noLock...); err != nil {
+		logrus.Errorf("do scheduling error:%v", err)
+	}
+}
+
+// Main routine to schedule tasks
+/*
+* @Description: tryScheduleInner tries to schedule the next tasks.
+* @Note: This function will try to schedule the next tasks.
+* @Param softSchedule
+* @Param noLock
+ */
+func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
+	// Implement logic to get the next task in the queue for the given space
+	if !(len(noLock) > 0 && noLock[0]) {
+		defer s.Unlock(s.Lock())
+	}
+
+	// step 1: make sure all tasks have alloc to a worker group
+	// This is done by the TaskManager, which assigns a worker group to each task
+	s.taskManager.RefreshTaskToWorkerGroupMap()
+
+	// step 2: get available resources and tasks
+	logrus.Debugf("scheduling next tasks, softSchedule: %v", softSchedule)
+	idleWorkerGroups := s.resourceManager.GetIdleWorkerGroups()
+	concurrentWorkerGroups := s.resourceManager.GetConcurrentWorkerGroups()
+	allTasks := s.taskManager.GetAllTasksNotComplete()
+	if len(allTasks) == 0 || (len(idleWorkerGroups) == 0 && len(concurrentWorkerGroups) == 0) {
+		logrus.Debugf("no available tasks or workerGroups, allTasks: %d, workerGroups: %d/%d",
+			len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+		return nil
+	}
+	logrus.Debugf("all tasks: %d, workerGroups: %d/%d", len(allTasks), len(idleWorkerGroups), len(concurrentWorkerGroups))
+

Review Comment:
   Perhaps we need to raise a new issue to solve this problem



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447128157


##########
vermeer/apps/master/bl/task_creator.go:
##########
@@ -99,6 +99,28 @@ func (bc *baseCreator) NewTaskInfo(graphName string, params map[string]string, t
 	return task, nil
 }
 
+func (bc *baseCreator) CopyTaskInfo(src *structure.TaskInfo) (*structure.TaskInfo, error) {

Review Comment:
   The functional test has been written in the test/ folder. The copied task will not have a new cron expression and will behave as expected.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org

Re: [PR] feat(Scheduler): Support priority/elder/depends based scheduling. [incubator-hugegraph-computer]

["Ethereal-O (via GitHub)" <gi...@apache.org>]

Ethereal-O commented on code in PR #336:
URL: https://github.com/apache/incubator-hugegraph-computer/pull/336#discussion_r2447129671


##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -62,6 +65,56 @@ func (tb *TaskBl) CreateTaskInfo(
 		return nil, err
 	}
 
+	// for scheduler
+	taskInfo.Priority = 0
+	taskInfo.Preorders = make([]int32, 0)
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
+	if params != nil {
+		if priority, ok := params["priority"]; ok {
+			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {

Review Comment:
   Have solved.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org