update submit taskid indexes resoursemap

nm/model_handler.go

@@ -49,6 +49,8 @@ func monitorModelInfo(dockerOp *operate.DockerOp) {
 				log.Error("Docker op ps images failed:", err)
 				continue
 			}
+			reportTaskIds := make([]uint64, 0)
+			reportTaskIds = append(reportTaskIds, uint64(len(modelInfosResp)))
 			for _, modelInfo := range modelInfosResp {
 				if modelInfo.ImageName == "" {
 					continue
@@ -64,16 +66,16 @@ func monitorModelInfo(dockerOp *operate.DockerOp) {
 					if isPull {
 						go dockerOp.PullImage(modelInfo)
 						modelInfo.IsImageExist = true
-						dockerOp.ModelTaskIdChan <- modelInfo.TaskId
-						dockerOp.IsReportModelTaskId[modelInfo.TaskId] = true
+						// todo: 是否立马上报数据
+						// reportTaskIds = append(reportTaskIds, modelInfo.TaskId)
 					}
-				} else if !dockerOp.IsReportModelTaskId[modelInfo.TaskId] {
-					dockerOp.ModelTaskIdChan <- modelInfo.TaskId
-					dockerOp.IsReportModelTaskId[modelInfo.TaskId] = true
+				} else {
+					reportTaskIds = append(reportTaskIds, modelInfo.TaskId)
 				}
 				dockerOp.SignApi[modelInfo.ImageName] = modelInfo.SignUrl
 				dockerOp.ModelsInfo = append(dockerOp.ModelsInfo, modelInfo)
 			}
+			dockerOp.ModelTaskIdIndexesChan <- reportTaskIds
 			ticker = time.NewTicker(time.Minute * 10)
 		}
 	}
@@ -84,7 +86,7 @@ func reportModelInfo(nodeManager *models.NodeManagerClient,
 	msgRespWorker *RespMsgWorker, dockerOp *operate.DockerOp) {
 	for {
 		select {
-		case taskId := <-dockerOp.ModelTaskIdChan:
+		case taskId := <-dockerOp.ModelTaskIdIndexesChan:
 			if !nodeManager.Status {
 				log.WithField("endpoint", nodeManager.Endpoint).Error("Node manager is down , stop report model info")
 				return

nm/msg_resp.go

@@ -44,6 +44,10 @@ func (o *RespMsgWorker) SendMsg() {
 		case pool := <-o.MsgPool:
 			{
 				workerMsg := pool.handler(pool.params...)
+				if workerMsg == nil {
+					log.Warn("Send to node manager workerMsg is nil")
+					continue
+				}
 				err := pool.workerClient.SendMsg(workerMsg)
 				if err != nil {
 					log.Error("Send msg to nm client failed:", err)
@@ -71,10 +75,19 @@ func HeartbeatResp(params ...interface{}) *nodemanagerV1.WorkerMessage {
 
 func SubmitResourceMapRes(params ...interface{}) *nodemanagerV1.WorkerMessage {
 	log.Info("Heartbeat response received params: ", params)
-	taskIdIndex := params[0].(uint64)
-	b := bitmap.New(taskIdIndex)
+	taskIdLength := params[0].(uint64)
+	b := bitmap.New(taskIdLength)
+	for i := 1; i < len(params); i++ {
+		taskIdIndex := params[i].(uint64)
+		err := b.Set(taskIdIndex)
+		if err != nil {
+			log.WithField("taskId index", i).Error("Error setting task id index")
+			return nil
+		}
+	}
 	binary, err := b.MarshalBinary()
 	if err != nil {
+		log.Error("bitmap marshal binary failed with error: ", err)
 		return nil
 	}
 	heartRes := &nodemanagerV1.WorkerMessage{

operate/docker.go

@@ -29,8 +29,7 @@ type DockerOp struct {
 	UsedExternalPort       map[int64]bool
 	SignApi                map[string]string
 	ModelsInfo             []*models.ModelInfo
-	IsReportModelTaskId map[uint64]bool
-	ModelTaskIdChan     chan uint64
+	ModelTaskIdIndexesChan chan []uint64
 }
 
 func init() {
@@ -51,9 +50,8 @@ func NewDockerOp() *DockerOp {
 		dockerClient:           dockerClient,
 		SignApi:                make(map[string]string, 0),
 		ModelsInfo:             make([]*models.ModelInfo, 0),
-		IsReportModelTaskId: make(map[uint64]bool, 0),
 		UsedExternalPort:       make(map[int64]bool, 0),
-		ModelTaskIdChan:     make(chan uint64, 0),
+		ModelTaskIdIndexesChan: make(chan []uint64, 0),
 	}
 }