adjust queue

Author: xitongsys

config/config.json

@@ -4,9 +4,8 @@
 	"Schema": "test",
 	"S3Region" : "us-east-1",
 	"ParallelNumber": 4,
-	"MaxExecutorNumber":30,
 	"MaxConcurrentTaskNumber": 4,
-	"QueueSize": 100
+	"MaxQueueSize": 100
     },
 
     "FileConnectorConfigs":{

scheduler/queue.go

@@ -75,17 +75,32 @@ func (self *Queue) Delete(task *Task) error {
 	return nil
 }
 
-func (self *Queue) Add(task *Task) {
+func (self *Queue) AddForce(task *Task) error {
 	self.Lock()
 	defer self.Unlock()
 
 	ln := len(self.Tasks)
-	if ln >= int(self.MaxQueueSize) && self.Tasks.Len() > 0 {
+	if ln >= int(self.MaxQueueSize) {
 		self.Tasks[0] = task
 	} else {
 		self.Tasks = append(self.Tasks, task)
 	}
 	sort.Sort(self.Tasks)
+	return nil
+}
+
+func (self *Queue) Add(task *Task) error {
+	self.Lock()
+	defer self.Unlock()
+
+	ln := len(self.Tasks)
+	if ln >= int(self.MaxQueueSize) {
+		return fmt.Errorf("queue is full")
+	} else {
+		self.Tasks = append(self.Tasks, task)
+	}
+	sort.Sort(self.Tasks)
+	return nil
 }
 
 func (self *Queue) HasTask(taskId string) bool {

scheduler/scheduler.go

@@ -74,13 +74,13 @@ func (self *Scheduler) AddTask(task *Task) (err error) {
 	defer self.Unlock()
 
 	if task.Status == pb.TaskStatus_TODO {
-		self.TodoQueue.Add(task)
+		return self.TodoQueue.Add(task)
 	} else if task.Status == pb.TaskStatus_ERROR {
-		self.ErrorQueue.Add(task)
+		return self.ErrorQueue.AddForce(task)
 	} else if task.Status == pb.TaskStatus_SUCCEED {
-		self.SucceedQueue.Add(task)
+		return self.SucceedQueue.AddForce(task)
 	} else if task.Status == pb.TaskStatus_RUNNING {
-		self.RunningQueue.Add(task)
+		return self.RunningQueue.Add(task)
 	} else {
 		return fmt.Errorf("unknown task status")
 	}