Shut down cluster at the end of execution

Author: salvacorts

mlp-ea-decentralized/common/ga/cluster.go

@@ -21,6 +21,7 @@ type Cluster struct {
 	ReceiveBestIndividual   chan Individual
 	list                    *memberlist.Memberlist
 	delegate                *nodeDelegate
+	stop                    chan bool
 }
 
 type eventHandler struct {
@@ -35,6 +36,7 @@ func MakeCluster(listenPort int, buffSize int, newBestIndividualChan chan Indivi
 		BoostrapPeers:           boostrapPeers,
 		BroadcastBestIndividual: make(chan Individual, buffSize),
 		ReceiveBestIndividual:   newBestIndividualChan,
+		stop:                    make(chan bool),
 	}
 
 	// If a logger is provided, set it instead of the logrus standard one
@@ -48,19 +50,32 @@ func MakeCluster(listenPort int, buffSize int, newBestIndividualChan chan Indivi
 // PrintMembers of the p2p cluster
 func (c *Cluster) PrintMembers() {
 	for {
-		time.Sleep(5 * time.Second)
+		select {
+		default:
+			time.Sleep(5 * time.Second)
 
-		nodes := c.list.Members()
-		str := fmt.Sprintf("Nodes - %d:\n", len(nodes))
+			nodes := c.list.Members()
+			str := fmt.Sprintf("Nodes - %d:\n", len(nodes))
 
-		for _, node := range nodes {
-			str += fmt.Sprintf("%s:%d - %s\n", node.Addr.String(), node.Port, node.Name)
+			for _, node := range nodes {
+				str += fmt.Sprintf("%s:%d - %s\n", node.Addr.String(), node.Port, node.Name)
+			}
+
+			c.Logger.Info(str)
+
+		case <-c.stop:
+			return
 		}
 
-		c.Logger.Info(str)
 	}
 }
 
+// Shutdown the cluster node
+func (c *Cluster) Shutdown() {
+	c.Logger.Info("Shutting down cluster")
+	close(c.stop)
+}
+
 // Start creates a new node with metadata for this process and joins a existing cluster
 // by connecting to peers in cluster.boostrapPeers array
 func (c *Cluster) Start(metadata NodeMetadata) {
@@ -100,10 +115,16 @@ func (c *Cluster) Start(metadata NodeMetadata) {
 	//go c.updateMetadataHandler()
 	go c.PrintMembers()
 
-	// TODO: Add stop channel and select
 	for {
-		indiv := <-c.BroadcastBestIndividual
-		c.delegate.Broadcast(indiv)
+		select {
+		default:
+			indiv := <-c.BroadcastBestIndividual
+			c.delegate.Broadcast(indiv)
+
+		case <-c.stop:
+			return
+		}
+
 	}
 }
 

mlp-ea-decentralized/common/ga/poolModel.go

@@ -128,6 +128,7 @@ func (pool *PoolModel) Minimize() {
 
 	// Launch cluster membership service
 	go pool.cluster.Start(pool.getCurrentNodeMetadata())
+	defer pool.cluster.Shutdown()
 
 	// Append all individuals in the population to the evaluation channel
 	pool.population.Fold(func(indiv eaopt.Individual) bool {
@@ -390,6 +391,7 @@ func (pool *PoolModel) handleBroadcastedIndividual() {
 			} else {
 				Log.Warn("Got broadcasted solution worse than currest best one. Ignoring...")
 			}
+
 		case <-pool.stop:
 			return
 		}
@@ -447,31 +449,38 @@ func (pool *PoolModel) populationGrowControl() {
 	Log.Infoln("populationGrowControl goroutine started")
 
 	for {
-		time.Sleep(1 * time.Second)
+		select {
+		default:
+			time.Sleep(1 * time.Second)
 
-		snap := pool.population.Snapshot()
-		indivArr := make([]eaopt.Individual, len(snap))
-		var bestCopy eaopt.Individual
-		i := 0
+			snap := pool.population.Snapshot()
+			indivArr := make([]eaopt.Individual, len(snap))
+			var bestCopy eaopt.Individual
+			i := 0
 
-		for _, item := range snap {
-			indivArr[i] = item.Object.(eaopt.Individual)
-			i++
-		}
+			for _, item := range snap {
+				indivArr[i] = item.Object.(eaopt.Individual)
+				i++
+			}
 
-		indivArr = pool.SortFunc(indivArr, pool.SortPrecission)
+			indivArr = pool.SortFunc(indivArr, pool.SortPrecission)
 
-		for i := pool.PopSize; i < len(indivArr); i++ {
-			pool.population.Remove(indivArr[i])
-		}
+			for i := pool.PopSize; i < len(indivArr); i++ {
+				pool.population.Remove(indivArr[i])
+			}
 
-		// Replace the remaining worst one by the best one so far
-		if len(indivArr) >= pool.PopSize {
-			pool.population.Remove(indivArr[pool.PopSize-1])
+			// Replace the remaining worst one by the best one so far
+			if len(indivArr) >= pool.PopSize {
+				pool.population.Remove(indivArr[pool.PopSize-1])
 
-			bestCopy = pool.BestSolution.Clone(pool.Rnd)
-			pool.population.Add(bestCopy)
+				bestCopy = pool.BestSolution.Clone(pool.Rnd)
+				pool.population.Add(bestCopy)
+			}
+
+		case <-pool.stop:
+			return
 		}
+
 	}
 }
 
@@ -483,61 +492,68 @@ func (pool *PoolModel) migrationScheduler() {
 	Log.Infoln("migrationScheduler goroutine started")
 
 	for {
-		time.Sleep(5 * time.Second)
-		var conn *grpc.ClientConn
-		dialed := false
+		select {
+		default:
+			time.Sleep(5 * time.Second)
+			var conn *grpc.ClientConn
+			dialed := false
 
-		snap := pool.population.Snapshot()
-		indivArr := make([]eaopt.Individual, 0, len(snap))
-		migrate := make([]Individual, pool.NMigrate)
+			snap := pool.population.Snapshot()
+			indivArr := make([]eaopt.Individual, 0, len(snap))
+			migrate := make([]Individual, pool.NMigrate)
 
-		for _, item := range snap {
-			indivArr = append(indivArr, item.Object.(eaopt.Individual).Clone(pool.Rnd))
-		}
+			for _, item := range snap {
+				indivArr = append(indivArr, item.Object.(eaopt.Individual).Clone(pool.Rnd))
+			}
 
-		// Sort population and get NMigrate best
-		indivArr = pool.SortFunc(indivArr, pool.SortPrecission)
-		for i := range migrate {
-			migrate[i] = Individual{
-				IndividualID: indivArr[i].ID,
-				Fitness:      indivArr[i].Fitness,
-				Evaluated:    indivArr[i].Evaluated,
-				Genome:       pool.Delegate.SerializeGenome(indivArr[i].Genome),
+			// Sort population and get NMigrate best
+			indivArr = pool.SortFunc(indivArr, pool.SortPrecission)
+			for i := range migrate {
+				migrate[i] = Individual{
+					IndividualID: indivArr[i].ID,
+					Fitness:      indivArr[i].Fitness,
+					Evaluated:    indivArr[i].Evaluated,
+					Genome:       pool.Delegate.SerializeGenome(indivArr[i].Genome),
+				}
 			}
-		}
 
-		// Pick a random node from the cluster and dial it
-		members := pool.cluster.GetMembers()
-		for !dialed {
-			remotePeer := members[pool.Rnd.Intn(len(members))]
-			remoteMetadata := NodeMetadata{}
+			// Pick a random node from the cluster and dial it
+			members := pool.cluster.GetMembers()
+			for !dialed {
+				remotePeer := members[pool.Rnd.Intn(len(members))]
+				remoteMetadata := NodeMetadata{}
 
-			err := remoteMetadata.Unmarshal(remotePeer.Meta)
-			if err != nil {
-				Log.Errorf("Could not parse Node Metadata on migrationScheduler. %s", err.Error())
+				err := remoteMetadata.Unmarshal(remotePeer.Meta)
+				if err != nil {
+					Log.Errorf("Could not parse Node Metadata on migrationScheduler. %s", err.Error())
+
+				}
+
+				remoteAddr := fmt.Sprintf("%s:%d",
+					remotePeer.Addr.String(), remoteMetadata.GrpcPort)
 
+				conn, err = grpc.Dial(remoteAddr, grpc.WithInsecure())
+				if err != nil {
+					Log.Warnf("Could not dial %s. %s", remoteAddr, err.Error())
+				}
+
+				dialed = true
 			}
 
-			remoteAddr := fmt.Sprintf("%s:%d",
-				remotePeer.Addr.String(), remoteMetadata.GrpcPort)
+			client := NewDistributedEAClient(conn)
+			_, err := client.MigrateIndividuals(context.Background(), &IndividualsBatch{
+				Individuals: migrate,
+			})
 
-			conn, err = grpc.Dial(remoteAddr, grpc.WithInsecure())
 			if err != nil {
-				Log.Warnf("Could not dial %s. %s", remoteAddr, err.Error())
+				Log.Errorf("could not MigrateIndividuals to %s. %s", conn.Target(), err.Error())
 			}
 
-			dialed = true
-		}
+			conn.Close()
 
-		client := NewDistributedEAClient(conn)
-		_, err := client.MigrateIndividuals(context.Background(), &IndividualsBatch{
-			Individuals: migrate,
-		})
-
-		if err != nil {
-			Log.Errorf("could not MigrateIndividuals to %s. %s", conn.Target(), err.Error())
+		case <-pool.stop:
+			return
 		}
 
-		conn.Close()
 	}
 }