fix bug

server/node.go

@@ -228,8 +228,9 @@ func (n *Node) Loop(idx int) {
 				go func(ctx context.Context, task *odysseus.TaskContent) {
 					l := log.WithField("task-id", task.TaskId)
 					l.WithField("task", task).Info("get task")
+					checkedWorker := make(map[string]bool)
 					for {
-						worker, err := n.PopWorker(ctx, n.rdb, t)
+						worker, err := n.PopWorker(ctx, n.rdb, t, checkedWorker)
 						if err == ErrNoWorker || err == ErrTimeout {
 							result := &odysseus.TaskResponse{
 								TaskId:        task.TaskId,
@@ -357,8 +358,7 @@ func (n *Node) addWorkerBack(w Worker) {
 	log.WithField("worker", w.workerid).Debug("add worker back to queue")
 }
 
-func (n *Node) PopWorker(ctx context.Context, rdb *redis.Client, task *odysseus.TaskContent) (Worker, error) {
-	var checkedWorker = make(map[string]bool)
+func (n *Node) PopWorker(ctx context.Context, rdb *redis.Client, task *odysseus.TaskContent, checkedWorker map[string]bool) (Worker, error) {
 
 	for i := 0; i < maxPriority; i++ {
 		for {
@@ -396,7 +396,7 @@ func (n *Node) PopWorker(ctx context.Context, rdb *redis.Client, task *odysseus.
 				priority: i,
 				managers: managerList,
 			}
-			if false {
+			if true {
 				if !checkWorkerHasResource(rdb, worker.addr, task.TaskType) {
 					n.addWorkerBack(worker)
 					if checked := checkedWorker[worker.workerid]; checked {

server/worker.go

@@ -26,6 +26,7 @@ var (
 	ErrNoWorker       = errors.New("no worker")
 	ErrTimeout        = errors.New("schedule timeout")
 	ErrDispatchFailed = errors.New("dispatch to nodemanager failed")
+	ErrCannotExecute  = errors.New("worker can't execute task")
 )
 
 type Worker struct {
@@ -144,6 +145,10 @@ func (n *Node) DispatchTask(ctx context.Context, worker Worker, task *odysseus.T
 				shouldAddBack = true
 				return ErrTimeout
 			}
+			if strings.HasSuffix(err.Error(), "worker can't execute task") {
+				shouldAddBack = true
+				return ErrCannotExecute
+			}
 			continue
 		}
 		return nil