floraachy
/
pcm-coordinator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix staticweight bug 

Former-commit-id: fe3017c8e0ca945d333009e9447c7b07ac8c1f4c
This commit is contained in:
tzwang 2024-06-27 15:44:58 +08:00
parent ee87491df4
commit e5582bbbcb
2 changed files with 19 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
api/internal/scheduler/algorithm/weightDistributing/weightDistributing.go
View File

@ -11,6 +11,7 @@ type Weight struct {
} }
func DistributeReplicas(weights []*Weight, replicas int32) { func DistributeReplicas(weights []*Weight, replicas int32) {
var weightSum int32 var weightSum int32
weightSum = 0 weightSum = 0
for _, w := range weights { for _, w := range weights {
@ -59,6 +60,10 @@ func DistributeReplicas(weights []*Weight, replicas int32) {
weightRatio[maxIdx]-- weightRatio[maxIdx]--
rest-- rest--
} else { } else {
if weights[minIdx].Replica == 0 {
weightRatio[minIdx]++
continue
}
weights[minIdx].Replica-- weights[minIdx].Replica--
weightRatio[minIdx]++ weightRatio[minIdx]++
rest++ rest++

17
api/internal/scheduler/service/inference/imageInfer.go
View File

@ -63,18 +63,23 @@ func Infer(opt *option.InferOption, id int64, adapterName string, clusters []*st
} }
} }
var mutex sync.Mutex
errMap := make(map[string]string)
for _, cluster := range clusters { for _, cluster := range clusters {
wg.Add(1) wg.Add(1)
c := cluster c := cluster
go func() { go func() {
imageUrls, err := collectorMap[c.ClusterId].GetInferUrl(ctx, opt) imageUrls, err := collectorMap[c.ClusterId].GetInferUrl(ctx, opt)
for i, _ := range imageUrls {
imageUrls[i].Url = imageUrls[i].Url + storeLink.FORWARD_SLASH + "image"
}
if err != nil { if err != nil {
mutex.Lock()
errMap[c.ClusterId] = err.Error()
mutex.Unlock()
wg.Done() wg.Done()
return return
} }
for i, _ := range imageUrls {
imageUrls[i].Url = imageUrls[i].Url + storeLink.FORWARD_SLASH + "image"
}
clusterName, _ := storage.GetClusterNameById(c.ClusterId) clusterName, _ := storage.GetClusterNameById(c.ClusterId)
s := struct { s := struct {
@ -111,6 +116,9 @@ func Infer(opt *option.InferOption, id int64, adapterName string, clusters []*st
for _, t := range aiTaskList { for _, t := range aiTaskList {
t.Status = constants.Failed t.Status = constants.Failed
t.EndTime = time.Now().Format(time.RFC3339) t.EndTime = time.Now().Format(time.RFC3339)
if _, ok := errMap[strconv.Itoa(int(t.ClusterId))]; ok {
t.Msg = errMap[strconv.Itoa(int(t.ClusterId))]
}
err := storage.UpdateAiTask(t) err := storage.UpdateAiTask(t)
if err != nil { if err != nil {
logx.Errorf(err.Error()) logx.Errorf(err.Error())
@ -142,6 +150,9 @@ func Infer(opt *option.InferOption, id int64, adapterName string, clusters []*st
if ac.ClusterId == strconv.Itoa(int(t.ClusterId)) { if ac.ClusterId == strconv.Itoa(int(t.ClusterId)) {
t.Status = constants.Failed t.Status = constants.Failed
t.EndTime = time.Now().Format(time.RFC3339) t.EndTime = time.Now().Format(time.RFC3339)
if _, ok := errMap[strconv.Itoa(int(t.ClusterId))]; ok {
t.Msg = errMap[strconv.Itoa(int(t.ClusterId))]
}
err := storage.UpdateAiTask(t) err := storage.UpdateAiTask(t)
if err != nil { if err != nil {
logx.Errorf(err.Error()) logx.Errorf(err.Error())