chore: add some comments

Author: Ethereal-O

vermeer/apps/master/bl/scheduler_bl.go

@@ -28,6 +28,10 @@ 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
 	// resource management
@@ -46,6 +50,10 @@ type ScheduleBl struct {
 	softSchedule   bool
 }
 
+/*
+* @Description: Init initializes the ScheduleBl.
+* @Note: This function will initialize the ScheduleBl.
+ */
 func (s *ScheduleBl) Init() {
 	logrus.Info("Initializing ScheduleBl...")
 	s.LoadConfig()
@@ -64,6 +72,10 @@ func (s *ScheduleBl) Init() {
 	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
 
@@ -102,6 +114,10 @@ func (s *ScheduleBl) LoadConfig() {
 		s.startChanSize, s.tickerInterval, s.softSchedule)
 }
 
+/*
+* @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)
@@ -113,6 +129,11 @@ func (s *ScheduleBl) startTicker() {
 }
 
 // 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 {
@@ -126,6 +147,12 @@ func (s *ScheduleBl) TryScheduleNextTasks(noLock ...bool) {
 }
 
 // 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]) {
@@ -183,6 +210,12 @@ func (s *ScheduleBl) tryScheduleInner(softSchedule bool, noLock ...bool) error {
 
 // QueueTask Add the task to the inner queue.
 // If the task exists, return false.
+/*
+* @Description: QueueTask queues the task.
+* @Note: This function will queue the task.
+* @Param taskInfo
+* @Return bool, error
+ */
 func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 	if taskInfo == nil {
 		return false, errors.New("the argument `taskInfo` is nil")
@@ -231,6 +264,12 @@ func (s *ScheduleBl) QueueTask(taskInfo *structure.TaskInfo) (bool, error) {
 	return ok, 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")
@@ -255,6 +294,12 @@ func (s *ScheduleBl) QueueTaskFromTemplate(template *structure.TaskInfo) (int32,
 	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{}
@@ -281,7 +326,6 @@ func (s *ScheduleBl) BatchQueueTask(taskInfos []*structure.TaskInfo) ([]bool, []
 }
 
 // ******** CloseCurrent ********
-
 func (s *ScheduleBl) CloseCurrent(taskId int32, removeWorkerName ...string) error {
 	defer s.Unlock(s.Lock())
 
@@ -307,6 +351,8 @@ func (s *ScheduleBl) CloseCurrent(taskId int32, removeWorkerName ...string) erro
 	return nil
 }
 
+// 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)
@@ -406,7 +452,7 @@ func (s *ScheduleBl) startWaitingTask(agent *schedules.Agent, taskInfo *structur
 
 // ********* 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")

vermeer/apps/master/schedules/scheduler_algorithm_manager.go

@@ -11,6 +11,10 @@ import (
 	"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
@@ -22,12 +26,21 @@ type SchedulerAlgorithm interface {
 	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)
@@ -39,6 +52,11 @@ func (am *SchedulerAlgorithmManager) Init() {
 	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
@@ -56,6 +74,11 @@ func (am *SchedulerAlgorithmManager) RegisterSchedulerAlgorithm(schedulerAlgorit
 	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
@@ -68,18 +91,43 @@ func (am *SchedulerAlgorithmManager) RegisterFilterAlgorithm(filterAlgorithm Sch
 	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

vermeer/apps/master/schedules/scheduler_cron_manager.go

@@ -8,20 +8,36 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+/*
+* @Description: SchedulerCronManager is the manager for the scheduler cron.
+* @Note: This is the manager for the scheduler cron.
+ */
 type SchedulerCronManager struct {
 	cronTasks map[int32][]*structure.TaskInfo // cron expression to TaskInfo. Origin task ID to copied tasks
 	crons     map[int32][]*cron.Cron          // cron expression to cron jobs
 	// queueTemplateHandler is a function that handles the task queue
 	queueTemplateHandler func(*structure.TaskInfo) (int32, error)
 }
 
+/*
+* @Description: Init initializes the SchedulerCronManager.
+* @Note: This function will initialize the SchedulerCronManager.
+* @Param queueTemplateHandler
+* @Return *SchedulerCronManager
+ */
 func (t *SchedulerCronManager) Init(queueTemplateHandler func(*structure.TaskInfo) (int32, error)) *SchedulerCronManager {
 	t.cronTasks = make(map[int32][]*structure.TaskInfo)
 	t.crons = make(map[int32][]*cron.Cron)
 	t.queueTemplateHandler = queueTemplateHandler
 	return t
 }
 
+/*
+* @Description: CheckCronExpression checks the cron expression.
+* @Note: This function will check the cron expression.
+* @Param cronExpr
+* @Return error
+ */
 func (t *SchedulerCronManager) CheckCronExpression(cronExpr string) error {
 	if cronExpr == "" {
 		return errors.New("cron expression is empty")
@@ -33,6 +49,12 @@ func (t *SchedulerCronManager) CheckCronExpression(cronExpr string) error {
 	return nil
 }
 
+/*
+* @Description: AddCronTask adds the cron task.
+* @Note: This function will add the cron task.
+* @Param taskInfo
+* @Return error
+ */
 func (t *SchedulerCronManager) AddCronTask(taskInfo *structure.TaskInfo) error {
 	if taskInfo == nil {
 		return errors.New("the argument `taskInfo` is nil")
@@ -69,6 +91,12 @@ func (t *SchedulerCronManager) AddCronTask(taskInfo *structure.TaskInfo) error {
 	return nil
 }
 
+/*
+* @Description: DeleteTask deletes the cron task.
+* @Note: This function will delete the cron task.
+* @Param taskID
+* @Return error
+ */
 func (t *SchedulerCronManager) DeleteTask(taskID int32) error {
 	if _, exists := t.cronTasks[taskID]; !exists {
 		return errors.New("task not found in cron tasks")
@@ -83,6 +111,13 @@ func (t *SchedulerCronManager) DeleteTask(taskID int32) error {
 	return nil
 }
 
+/*
+* @Description: DeleteTaskByGraph deletes the cron task by graph.
+* @Note: This function will delete the cron task by graph.
+* @Param spaceName
+* @Param graphName
+* @Return error
+ */
 func (t *SchedulerCronManager) DeleteTaskByGraph(spaceName, graphName string) error {
 	if spaceName == "" || graphName == "" {
 		return errors.New("the argument `spaceName` or `graphName` is empty")