add mutex

nm/monitor_seed.go

@@ -42,6 +42,7 @@ func monitorNodeManagerSeed() {
 			}
 			for _, node := range list.GetManagers() {
 				if isExistNodeManager(node) {
+					log.Warn("Node manager is already exist and updated")
 					continue
 				}
 				nodeManagerArr = append(nodeManagerArr, &NodeManager{Info: node, IsUsed: false, IsExist: true})

nm/start.go

@@ -285,7 +285,9 @@ func handlerMsg(nodeManager *models.NodeManagerClient,
 						taskMsgWorker.Wg.Add(1)
 						taskMsgWorker.TaskMsg <- taskMsg
 						taskMsgWorker.Wg.Wait()
+						taskMsgWorker.Mutex.Lock()
 						taskExecResInterface, _ := taskMsgWorker.LruCache.Get(taskMsg.TaskId)
+						taskMsgWorker.Mutex.Unlock()
 						//log.WithField("result", taskExecResInterface).Info("lru cache get task result")
 						taskExecRes := &models.TaskResult{
 							TaskHttpStatusCode: 200,
@@ -305,11 +307,13 @@ func handlerMsg(nodeManager *models.NodeManagerClient,
 						}
 						reqHash, respHash, minerSign := taskMsgWorker.GetMinerSign(taskMsg, taskExecRes.TaskRespBody)
 						params := buildParams(taskMsg.TaskId, containerSign, minerSign, taskExecRes, isSuccess)
+						taskMsgWorker.Mutex.Lock()
 						taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.TaskType, taskMsg.TaskType)
 						taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.ContainerSign, containerSign)
 						taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.MinerSign, minerSign)
 						taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.ReqHash, reqHash)
 						taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.RespHash, respHash)
+						taskMsgWorker.Mutex.Unlock()
 						msgRespWorker.RegisterMsgResp(nodeManager, worker, SubmitResultResp, params)
 						log.Info("--------------taskMsg--------------:", taskMsg)
 					}(msgRespWorker, taskMsgWorker, taskMsg)

nm/task_msg.go

@@ -27,6 +27,7 @@ import (
 
 type TaskHandler struct {
 	Wg                 *sync.WaitGroup
+	Mutex              *sync.Mutex
 	LruCache           *lru.Cache
 	DockerOp           *operate.DockerOp
 	CmdOp              *operate.Command
@@ -41,6 +42,7 @@ var oldTaskImageName string
 func NewTaskWorker(op *operate.DockerOp) *TaskHandler {
 	return &TaskHandler{
 		Wg:           &sync.WaitGroup{},
+		Mutex:        &sync.Mutex{},
 		LruCache:     lru.New(100),
 		DockerOp:     op,
 		TaskMsg:      make(chan *nodeManagerV1.PushTaskMessage, 0),
@@ -288,7 +290,9 @@ func (t *TaskHandler) ComputeTaskHandler(taskMsg *nodeManagerV1.PushTaskMessage)
 	} else if taskMsg.TaskKind == baseV1.TaskKind_StandardTask {
 		t.IsExecStandardTask = false
 	}
+	t.Mutex.Lock()
 	t.LruCache.Add(taskMsg.TaskId, taskExecResult)
+	t.Mutex.Unlock()
 	//log.WithField("result", taskExecResult).Info("lru cache storage task result")
 	log.Info("received computeTask--------------------------------")
 }