Merge pull request 'updated acquire resources concurrently' (#86) from tzwang/pcm-coordinator:master into master

Former-commit-id: 2559e61a8deed328a47f5e640d2096ce11e8be9f

api/internal/scheduler/schedulers/aiScheduler.go

@@ -24,6 +24,7 @@ import (
 	"gitlink.org.cn/JointCloud/pcm-coordinator/api/pkg/response"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils"
+	"sync"
 )
 
 type AiScheduler struct {
@@ -98,25 +99,60 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) error {
 
 	executorMap := *as.AiExecutor
 	for _, cluster := range clusters {
-		_, err := executorMap[cluster.Name].Execute(as.option)
+		c := cluster
-		if err != nil {
+		if cluster.Replicas == 0 {
-			// TODO: database operation
+			continue
 		}
-		// TODO: database operation
+		go func() {
+			_, err := executorMap[c.Name].Execute(as.option)
+			if err != nil {
+				// TODO: database operation
+			}
+			// TODO: database operation
+		}()
 	}
 
 	return nil
 }
 
 func (as *AiScheduler) findClustersWithResources() ([]*collector.ResourceStats, error) {
+	var wg sync.WaitGroup
+	var ch = make(chan *collector.ResourceStats, len(*as.ResourceCollector))
+	var errCh = make(chan error, len(*as.ResourceCollector))
+
 	var resourceSpecs []*collector.ResourceStats
+	var errs []error
+
 	for _, resourceCollector := range *as.ResourceCollector {
-		spec, err := resourceCollector.GetResourceStats()
+		wg.Add(1)
-		if err != nil {
+		rc := resourceCollector
-			continue
+		go func() {
-		}
+			spec, err := rc.GetResourceStats()
-		resourceSpecs = append(resourceSpecs, spec)
+			if err != nil {
+				errCh <- err
+				wg.Done()
+				return
+			}
+			ch <- spec
+			wg.Done()
+		}()
 	}
+	wg.Wait()
+	close(ch)
+	close(errCh)
+
+	for s := range ch {
+		resourceSpecs = append(resourceSpecs, s)
+	}
+
+	for e := range errCh {
+		errs = append(errs, e)
+	}
+
+	if len(errs) != 0 {
+		return nil, errors.New("get resources failed")
+	}
+
 	if len(resourceSpecs) == 0 {
 		return nil, errors.New("no resource found")
 	}

api/internal/storeLink/storeLink.go

@@ -28,6 +28,7 @@ import (
 	"gitlink.org.cn/jcce-pcm/pcm-participant-octopus/octopus"
 	"gorm.io/gorm"
 	"strings"
+	"sync"
 )
 
 type Linkage interface {
@@ -124,73 +125,122 @@ func GetResourceTypes() []string {
 }
 
 func GetDatasetsNames(collectorMap *map[string]collector.AiCollector) ([]string, error) {
+	var wg sync.WaitGroup
+	var errCh = make(chan error, len(*collectorMap))
+	var errs []error
 	var names []string
-	//errCount := 0
+	var mu sync.Mutex
 	colMap := *collectorMap
 	for _, col := range colMap {
-		var ns []string
+		wg.Add(1)
-		specs, err := col.GetDatasetsSpecs()
+		c := col
-		if err != nil {
+		go func() {
-			return nil, errors.New("failed to acquire datasets list")
+			var ns []string
-		}
+			specs, err := c.GetDatasetsSpecs()
-		for _, spec := range specs {
+			if err != nil {
-			ns = append(ns, spec.Name)
+				errCh <- err
-		}
+				wg.Done()
-		if len(ns) == 0 {
+				return
-			continue
+			}
-		}
+			for _, spec := range specs {
-		if len(names) == 0 {
+				ns = append(ns, spec.Name)
-			names = ns
+			}
-			continue
+			if len(ns) == 0 {
-		}
+				wg.Done()
-
+				return
-		names = common.IntersectString(names, ns)
+			}
+			mu.Lock()
+			if len(names) == 0 {
+				names = ns
+				wg.Done()
+				mu.Unlock()
+				return
+			}
+			names = common.IntersectString(names, ns)
+			wg.Done()
+			mu.Unlock()
+		}()
 	}
-	//if (len(*collectorMap) - errCount) < 2 {
+	wg.Wait()
-	//
+	close(errCh)
-	//}
+
+	for e := range errCh {
+		errs = append(errs, e)
+	}
+
+	if len(errs) != 0 {
+		return nil, errors.New("get DatasetsNames failed")
+	}
+
 	names = common.RemoveDuplicates(names)
 	return names, nil
 }
 
 func GetAlgorithms(collectorMap *map[string]collector.AiCollector, resourceType string, taskType string, dataset string) ([]string, error) {
 	var names []string
+	var wg sync.WaitGroup
+	var errCh = make(chan error, len(*collectorMap))
+	var errs []error
+	var mu sync.Mutex
+
 	colMap := *collectorMap
 	for _, col := range colMap {
-		var ns []string
+		wg.Add(1)
-		algorithms, err := col.GetAlgorithms()
+		c := col
-		if err != nil {
+		go func() {
-			return nil, err
+			var ns []string
-		}
+			algorithms, err := c.GetAlgorithms()
-		for _, algorithm := range algorithms {
+			if err != nil {
-			if algorithm.TaskType != taskType {
+				errCh <- err
-				continue
+				wg.Done()
+				return
 			}
-			switch algorithm.Platform {
+			for _, algorithm := range algorithms {
-			case OCTOPUS:
+				if algorithm.TaskType != taskType {
-				splitns := strings.Split(algorithm.Name, UNDERSCORE)
-				if dataset != splitns[0] || len(splitns) == 1 {
 					continue
 				}
-				ns = append(ns, splitns[1])
+				switch algorithm.Platform {
-			case SHUGUANGAI:
+				case OCTOPUS:
-				splitns := strings.Split(algorithm.Name, DASH)
+					splitns := strings.Split(algorithm.Name, UNDERSCORE)
-				if dataset != splitns[0] || len(splitns) == 1 {
+					if dataset != splitns[0] || len(splitns) == 1 {
-					continue
+						continue
+					}
+					ns = append(ns, splitns[1])
+				case SHUGUANGAI:
+					splitns := strings.Split(algorithm.Name, DASH)
+					if dataset != splitns[0] || len(splitns) == 1 {
+						continue
+					}
+					ns = append(ns, splitns[1])
 				}
-				ns = append(ns, splitns[1])
 			}
-		}
+			if len(ns) == 0 {
-		if len(ns) == 0 {
+				wg.Done()
-			continue
+				return
-		}
+			}
-		if len(names) == 0 {
+			mu.Lock()
-			names = ns
+			if len(names) == 0 {
-			continue
+				names = ns
-		}
+				wg.Done()
-
+				mu.Unlock()
-		names = common.IntersectString(names, ns)
+				return
+			}
+			names = common.IntersectString(names, ns)
+			wg.Done()
+			mu.Unlock()
+		}()
 	}
+	wg.Wait()
+	close(errCh)
+
+	for e := range errCh {
+		errs = append(errs, e)
+	}
+
+	if len(errs) != 0 {
+		return nil, errors.New("get Algorithms failed")
+	}
+
 	names = common.RemoveDuplicates(names)
 	return names, nil
 }