fix bug

nmregistry/registry.go

@@ -46,7 +46,7 @@ func NewRegistryService(conf *config.Config, rdb *redis.Client, public ecdsa.Pub
 }
 
 func (s *RegistryService) Start() {
-	ticker := time.NewTicker(time.Second * 20)
+	ticker := time.NewTicker(time.Second * 1)
 	defer ticker.Stop()
 	refresh := time.NewTicker(time.Second * 5)
 	defer refresh.Stop()
@@ -58,6 +58,8 @@ func (s *RegistryService) Start() {
 		case <-ticker.C:
 			if err := s.registry(s.rdb); err != nil {
 				log.WithError(err).Error("registry failed")
+			} else {
+				ticker.Reset(time.Second * 10)
 			}
 		case <-refresh.C:
 			if nodes, err := s.allNodeManager(s.rdb); err != nil {
@@ -113,7 +115,7 @@ func (s *RegistryService) GetNodeManagerList(filter ManagerFilter) []NodeManager
 
 func (s *RegistryService) allNodeManager(rdb *redis.Client) ([]RegistryInfo, error) {
 	var ret []RegistryInfo
-	var tsExpired = 30
+	var tsExpired = 100
 	keys, err := rdb.Keys(context.Background(), config.NODE_MANAGER_SET+"*").Result()
 	if err != nil {
 		return nil, err

server/workerManager.go

@@ -97,6 +97,7 @@ func (wm *WorkerManager) AddNewWorker(uuid int64, worker omanager.NodeManagerSer
 		resultCh: make(chan *omanager.SubmitTaskResult),
 		uuid:     uuid,
 		stream:   worker,
+		quit:     make(chan interface{}),
 	}
 	taskCache, err := lru.New(100)
 	if err != nil {
@@ -134,6 +135,8 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 	deviceUsageTicker := time.NewTicker(time.Second * 10)
 	defer deviceUsageTicker.Stop()
 
+	defer wm.InActiveWorker(worker)
+
 	for {
 		var msg = new(omanager.ManagerMessage)
 		var callback = Callback(func(err error) bool {
@@ -146,11 +149,13 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 			gb := new(omanager.ManagerMessage_GoodbyeMessage)
 			gb.GoodbyeMessage = &omanager.GoodbyeMessage{}
 			msg.Message = gb
+		case <-worker.quit:
+			return nil
+
 		case <-workerCheckTicker.C:
 			if time.Now().Unix()-wm.GetHeartBeat(worker.uuid) > int64(workerCheckDuration.Seconds()) {
 				wm.InActiveWorker(worker)
-				// remove worker
-				close(worker.quit)
+				// todo: remove worker
 				return ErrHeartBeatExpired
 			}