update protocol

server/dispatchTask.go

@@ -115,7 +115,7 @@ func (d *dispatchTask) finalize(wm *WorkerManager) {
 
 	task := d.task
 	if task.TaskKind != odysseus.TaskKind_StandardTask && d.worker.online == true {
-		_ = wm.AddWorkerSingle(d.worker)
+
 	}
 
 	_, err := wm.taskResult(d.worker, task, result)

server/service.go

@@ -6,7 +6,6 @@ import (
 	"github.com/odysseus/nodemanager/utils"
 	omanager "github.com/odysseus/odysseus-protocol/gen/proto/go/nodemanager/v2"
 	log "github.com/sirupsen/logrus"
-	"strconv"
 	"strings"
 )
 
@@ -64,13 +63,6 @@ func (n *NodeManagerService) DispatchTask(ctx context.Context, request *omanager
 	if worker.online == false {
 		return nil, errors.New("worker offline")
 	}
-	{
-		nonceds := strings.Split(mids[1], ":")
-		nonce, _ := strconv.ParseInt(nonceds[0], 10, 64)
-		if nonce < int64(worker.nonce) {
-			return nil, errors.New("expired worker nonce")
-		}
-	}
 
 	dtask := newDispatchTask(worker, request.TaskData)
 

server/workerManager.go

@@ -3,7 +3,6 @@ package server
 import (
 	"bytes"
 	"encoding/hex"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"github.com/golang/protobuf/proto"
@@ -285,15 +284,17 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 
 		case <-workerCheckTicker.C:
 			if worker.info.nodeInfo != nil {
-				//nodeinfoTicker.Reset(time.Hour * 24)
+				nodeinfoTicker.Reset(time.Minute * 30)
 			}
 
 			if worker.usage.hwUsage != nil {
 				deviceUsageTicker.Reset(time.Second * time.Duration(tickerConf.DeviceUsageTicker))
 			}
 
-			if worker.registed && worker.addFirstSucceed == false && len(worker.deviceInfoHash) > 0 {
-				wm.AddWorkerToQueue(worker)
+			if worker.registed && worker.addFirstSucceed == false {
+				if err := wm.AddWorker(worker); err == nil {
+					worker.addFirstSucceed = true
+				}
 			}
 
 			wm.UpdateWorkerActive(worker)
@@ -477,6 +478,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"heartBeat":   time.Now().Unix() - int64(msg.HeartbeatResponse.Timestamp),
 				}).Debug("receive worker heartbeat")
 			case *omanager.WorkerMessage_NodeInfo:
+				// todo: remove this message.
 				nodeinfo := msg.NodeInfo
 				log.WithField("worker-addr", worker.workerAddr).Debugf("receive worker node info:%v", nodeinfo)
 				if nodeinfo.Hardware != nil && nodeinfo.Hardware.NET != nil {
@@ -566,29 +568,15 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					// ignore the info.
 					continue
 				}
+				// todo: verify signature
 				{
 					var infoHash [32]byte
-					infoData, err := json.Marshal(msg.DeviceInfo)
-					if err != nil {
-						l.WithFields(log.Fields{
-							"worker-addr": worker.workerAddr,
-							"error":       err,
-						}).Error("marshal device info failed")
-
-					}
-					if len(infoData) == 0 {
-						continue
-					}
-
-					infoHash = sha3.Sum256(infoData)
+					infoHash = sha3.Sum256([]byte(msg.DeviceInfo.String()))
+					// update local cache.
 					worker.info.nodeInfo.Hardware = msg.DeviceInfo.Hardware
-
-					if worker.registed && worker.addFirstSucceed == false {
-						wm.AddWorkerToQueue(worker)
-					}
 					// check device info changed, and update to cache.
 					if bytes.Compare(infoHash[:], worker.deviceInfoHash) != 0 {
-						wm.UpdateWorkerDeviceInfo(worker, string(infoData))
+						wm.UpdateWorkerDeviceInfo(worker, msg.DeviceInfo)
 					}
 					worker.deviceInfoHash = infoHash[:]
 				}
@@ -597,8 +585,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 				if !worker.registed {
 					continue
 				}
-				usageData, _ := json.Marshal(msg.DeviceUsage)
-				wm.UpdateWorkerUsageInfo(worker, string(usageData))
+				wm.UpdateWorkerUsageInfo(worker, msg.DeviceUsage)
 
 				worker.usage.hwUsage = msg.DeviceUsage.Usage
 				l.WithFields(log.Fields{
@@ -613,6 +600,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"worker-addr": worker.workerAddr,
 					"model count": len(msg.AddModelRunning.Models),
 				}).Debugf("receive worker add model running:%v", msg.AddModelRunning.Models)
+				// todo: add worker running model.
 
 			case *omanager.WorkerMessage_DelModeRunning:
 				if !worker.registed {
@@ -622,6 +610,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"worker-addr": worker.workerAddr,
 					"model count": len(msg.DelModeRunning.ModelIds),
 				}).Debugf("receive worker del model running:%v", msg.DelModeRunning.ModelIds)
+				// todo: del worker running model with model_id.
 
 			case *omanager.WorkerMessage_AddModelInstalled:
 				if !worker.registed {
@@ -631,6 +620,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"worker-addr": worker.workerAddr,
 					"model count": len(msg.AddModelInstalled.Models),
 				}).Debugf("receive worker add model installed:%v", msg.AddModelInstalled.Models)
+				// todo: add worker installed model with model_id.
 
 			case *omanager.WorkerMessage_DelModelInstalled:
 				if !worker.registed {
@@ -640,6 +630,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"worker-addr": worker.workerAddr,
 					"model count": len(msg.DelModelInstalled.ModelIds),
 				}).Debugf("receive worker del model installed:%v", msg.DelModelInstalled.ModelIds)
+				// todo: del worker installed model with model_id.
 
 			case *omanager.WorkerMessage_InstalledModelStatus:
 				if !worker.registed {
@@ -650,6 +641,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"model":       len(msg.InstalledModelStatus.ModelId),
 					"type":        "status",
 				}).Debugf("receive worker installed model status:%v", msg.InstalledModelStatus)
+				// todo: update worker installed model status.
 
 			case *omanager.WorkerMessage_RunningModelStatus:
 				if !worker.registed {
@@ -660,6 +652,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"model":       len(msg.RunningModelStatus.ModelId),
 					"type":        "status",
 				}).Debugf("receive worker running model status:%v", msg.RunningModelStatus)
+				// todo: update worker running model status.
 
 			case *omanager.WorkerMessage_GpuUsage:
 				if !worker.registed {
@@ -669,15 +662,17 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					"worker-addr": worker.workerAddr,
 					"usage count": len(msg.GpuUsage.Usages),
 				}).Debugf("receive worker gpu usage:%v", msg.GpuUsage.Usages)
+				// todo: update worker gpu usage info.
 
 			case *omanager.WorkerMessage_RegisteMessage:
+				// 1. do some verify.
 				if worker.registed {
 					continue
 				}
 				l.WithFields(log.Fields{
 					"worker-addr": worker.workerAddr,
 				}).Debug("receive registed message")
-				// todo: verify signature
+				// 2. check signature.
 				info := msg.RegisteMessage.Info
 				{
 
@@ -695,6 +690,9 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 						worker.quit <- ErrInvalidMessageValue
 						return
 					}
+
+				}
+				// 3. check timestamp  not expired.
 				if time.Now().Unix()-int64(msg.RegisteMessage.Timestamp) > config.GetConfig().GetWorkerSignatureExpiredTime() {
 					l.WithFields(log.Fields{
 						"worker-addr": worker.workerAddr,
@@ -702,13 +700,17 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					worker.quit <- ErrExpiredMsgSignature
 					return
 				}
-				}
+
+				// 4. replace old connection.
 
 				if pubkey, err := utils.HexToPubkey(info.MinerPubkey); err != nil {
 					l.WithFields(log.Fields{
 						"worker-addr": worker.workerAddr,
 						"error":       err,
 					}).Error("parse pubkey failed")
+					worker.quit <- ErrInvalidMsgSignature
+					return
+
 				} else {
 					addr := utils.PubkeyToAddress(pubkey)
 					if old := wm.GetWorkerByAddr(addr); old != nil {
@@ -721,6 +723,7 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					worker.workerAddr = addr
 				}
 				worker.registed = true
+				// 5. check ip address.
 				matched, err := regexp.MatchString("((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(\\.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}",
 					msg.RegisteMessage.Hardware.NET.Ip)
 				if err != nil {
@@ -735,14 +738,21 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 					Models:   msg.RegisteMessage.Models,
 				}
 				wm.SetWorkerAddr(worker, worker.workerAddr)
-
+				// check white list.
 				if err := wm.checkWhiteList(worker, info.BenefitAddress); err != nil {
 					worker.quit <- err
 					return
 				} else {
-					wm.addWorkerToSets(worker, info.BenefitAddress)
+					wm.addWorkerToWhiteListSet(worker, info.BenefitAddress)
+				}
+
+				// add worker to mogo.
+				if err := wm.AddWorker(worker); err == nil {
+					worker.addFirstSucceed = true
+					wm.UpdateWorkerActive(worker)
 				}
 
+				// start manage worker.
 				wreg := workerRegistry{
 					worker: worker,
 					wm:     wm,

server/worker_registry.go

@@ -87,7 +87,7 @@ func (w workerRegistry) DetailInfo() (json.RawMessage, error) {
 	}
 	info.HearBeat = w.wm.GetHeartBeat(w.worker.uuid) * 1000 // to ms
 	info.MinerAddress = w.worker.workerAddr
-	info.Nonce = int64(w.worker.nonce)
+	info.Nonce = 0
 	if w.worker.info.nodeInfo != nil {
 		info.CpuModel = w.worker.info.nodeInfo.Hardware.CPU.Model
 		info.CpuCore = int(w.worker.info.nodeInfo.Hardware.CPU.Cores)