|
|
|
@@ -130,39 +130,44 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interfa |
|
|
|
var wg sync.WaitGroup |
|
|
|
var results []*AiResult |
|
|
|
var errs []interface{} |
|
|
|
var ch = make(chan *AiResult, len(clusters)) |
|
|
|
var errCh = make(chan interface{}, len(clusters)) |
|
|
|
var taskNum int32 |
|
|
|
for _, cluster := range clusters { |
|
|
|
taskNum += cluster.Replicas |
|
|
|
} |
|
|
|
var ch = make(chan *AiResult, taskNum) |
|
|
|
var errCh = make(chan interface{}, taskNum) |
|
|
|
|
|
|
|
executorMap := as.AiService.AiExecutorAdapterMap[as.option.AdapterId] |
|
|
|
for _, cluster := range clusters { |
|
|
|
c := cluster |
|
|
|
wg.Add(1) |
|
|
|
go func() { |
|
|
|
opt, _ := cloneAiOption(as.option) |
|
|
|
resp, err := executorMap[c.ClusterId].Execute(as.ctx, opt) |
|
|
|
|
|
|
|
if err != nil { |
|
|
|
e := struct { |
|
|
|
err error |
|
|
|
clusterId string |
|
|
|
}{ |
|
|
|
err: err, |
|
|
|
clusterId: c.ClusterId, |
|
|
|
for i := 0; i < int(c.Replicas); i++ { |
|
|
|
wg.Add(1) |
|
|
|
go func() { |
|
|
|
opt, _ := cloneAiOption(as.option) |
|
|
|
resp, err := executorMap[c.ClusterId].Execute(as.ctx, opt) |
|
|
|
if err != nil { |
|
|
|
e := struct { |
|
|
|
err error |
|
|
|
clusterId string |
|
|
|
}{ |
|
|
|
err: err, |
|
|
|
clusterId: c.ClusterId, |
|
|
|
} |
|
|
|
errCh <- e |
|
|
|
wg.Done() |
|
|
|
return |
|
|
|
} |
|
|
|
errCh <- e |
|
|
|
wg.Done() |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
result, _ := convertType(resp) |
|
|
|
result.Replica = c.Replicas |
|
|
|
result.ClusterId = c.ClusterId |
|
|
|
result.Strategy = as.option.StrategyName |
|
|
|
result.Card = opt.ComputeCard |
|
|
|
result, _ := convertType(resp) |
|
|
|
result.Replica = c.Replicas |
|
|
|
result.ClusterId = c.ClusterId |
|
|
|
result.Strategy = as.option.StrategyName |
|
|
|
result.Card = opt.ComputeCard |
|
|
|
|
|
|
|
ch <- result |
|
|
|
wg.Done() |
|
|
|
}() |
|
|
|
ch <- result |
|
|
|
wg.Done() |
|
|
|
}() |
|
|
|
} |
|
|
|
} |
|
|
|
wg.Wait() |
|
|
|
close(ch) |
|
|
|
|