JointCloud
/
pcm-coordinator
24
0
Fork 8
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update aischeduler

This commit is contained in:
tzwang 2024-12-24 18:03:37 +08:00
parent 8f4bec0db1
commit 0e656f11f4
16 changed files with 33 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
internal/logic/cloud/commitgeneraltasklogic.go
View File

@ -5,6 +5,7 @@ import (
"context" "context"
"github.com/pkg/errors" "github.com/pkg/errors"
clientCore "gitlink.org.cn/JointCloud/pcm-coordinator/client" clientCore "gitlink.org.cn/JointCloud/pcm-coordinator/client"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option"
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants"
@ -69,7 +70,7 @@ func (l *CommitGeneralTaskLogic) CommitGeneralTask(req *types.GeneralTaskReq) er
utils.Convert(&req, &opt) utils.Convert(&req, &opt)
sc, _ := schedulers.NewCloudScheduler(l.ctx, "", l.svcCtx.Scheduler, opt, tx, l.svcCtx.PromClient) sc, _ := schedulers.NewCloudScheduler(l.ctx, "", l.svcCtx.Scheduler, opt, tx, l.svcCtx.PromClient)
results, err := l.svcCtx.Scheduler.AssignAndSchedule(sc) results, err := l.svcCtx.Scheduler.AssignAndSchedule(sc, scheduler.JOINT_CLOUD_MODE)
if err != nil { if err != nil {
logx.Errorf("AssignAndSchedule() => execution error: %v", err) logx.Errorf("AssignAndSchedule() => execution error: %v", err)
return err return err

3
internal/logic/core/commitvmtasklogic.go
View File

@ -4,6 +4,7 @@ import (
"context" "context"
"fmt" "fmt"
clientCore "gitlink.org.cn/JointCloud/pcm-coordinator/client" clientCore "gitlink.org.cn/JointCloud/pcm-coordinator/client"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
@ -62,7 +63,7 @@ func (l *CommitVmTaskLogic) CommitVmTask(req *types.CommitVmTaskReq) (resp *type
return nil, err return nil, err
} }
// 3、Return scheduling results // 3、Return scheduling results
results, err := l.svcCtx.Scheduler.AssignAndSchedule(vmSchdl) results, err := l.svcCtx.Scheduler.AssignAndSchedule(vmSchdl, scheduler.JOINT_CLOUD_MODE)
if err != nil { if err != nil {
logx.Errorf("AssignAndSchedule() => execution error: %v", err) logx.Errorf("AssignAndSchedule() => execution error: %v", err)
return nil, err return nil, err

5
internal/logic/schedule/scheduleruntasklogic.go
View File

@ -42,6 +42,11 @@ func (l *ScheduleRunTaskLogic) ScheduleRunTask(req *types.RunTaskReq) (resp *typ
return nil, err return nil, err
} }
//schedule, err := l.svcCtx.Scheduler.AssignAndSchedule()
//if err != nil {
// return nil, err
//}
adapterName, err := l.svcCtx.Scheduler.AiStorages.GetAdapterNameById(ADAPTERID) adapterName, err := l.svcCtx.Scheduler.AiStorages.GetAdapterNameById(ADAPTERID)
if err != nil { if err != nil {
return nil, err return nil, err

3
internal/logic/schedule/schedulesubmitlogic.go
View File

@ -2,6 +2,7 @@ package schedule
import ( import (
"context" "context"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
@ -51,7 +52,7 @@ func (l *ScheduleSubmitLogic) ScheduleSubmit(req *types.ScheduleReq) (resp *type
return nil, err return nil, err
} }
results, err := l.svcCtx.Scheduler.AssignAndSchedule(aiSchdl) results, err := l.svcCtx.Scheduler.AssignAndSchedule(aiSchdl, scheduler.JOINT_CLOUD_MODE)
if err != nil { if err != nil {
return nil, err return nil, err
} }

3
internal/mqs/ScheduleAi.go
View File

@ -16,6 +16,7 @@ package mqs
import ( import (
"context" "context"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
) )
@ -40,7 +41,7 @@ func (l *AiQueue) Consume(val string) error {
aiSchdl, _ := schedulers.NewAiScheduler(l.ctx, val, l.svcCtx.Scheduler, nil) aiSchdl, _ := schedulers.NewAiScheduler(l.ctx, val, l.svcCtx.Scheduler, nil)
// 调度算法 // 调度算法
_, err := l.svcCtx.Scheduler.AssignAndSchedule(aiSchdl) _, err := l.svcCtx.Scheduler.AssignAndSchedule(aiSchdl, scheduler.JOINT_CLOUD_MODE)
if err != nil { if err != nil {
return err return err
} }

11
internal/scheduler/scheduler.go
View File

@ -29,6 +29,11 @@ import (
"sync" "sync"
) )
const (
JOINT_CLOUD_MODE = iota + 1
STORAGE_SCHEDULE_MODE
)
type Scheduler struct { type Scheduler struct {
task *response.TaskInfo task *response.TaskInfo
participantIds []int64 participantIds []int64
@ -43,7 +48,7 @@ type Scheduler struct {
type SubSchedule interface { type SubSchedule interface {
GetNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) GetNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error)
PickOptimalStrategy() (strategy.Strategy, error) PickOptimalStrategy() (strategy.Strategy, error)
AssignTask(clusters []*strategy.AssignedCluster) (interface{}, error) AssignTask(clusters []*strategy.AssignedCluster, mode int) (interface{}, error)
} }
func NewScheduler(subSchedule SubSchedule, val string, dbEngin *gorm.DB) (*Scheduler, error) { func NewScheduler(subSchedule SubSchedule, val string, dbEngin *gorm.DB) (*Scheduler, error) {
@ -126,7 +131,7 @@ func (s *Scheduler) TempAssign() error {
return nil return nil
} }
func (s *Scheduler) AssignAndSchedule(ss SubSchedule) (interface{}, error) { func (s *Scheduler) AssignAndSchedule(ss SubSchedule, mode int) (interface{}, error) {
//choose strategy //choose strategy
strategy, err := ss.PickOptimalStrategy() strategy, err := ss.PickOptimalStrategy()
if err != nil { if err != nil {
@ -140,7 +145,7 @@ func (s *Scheduler) AssignAndSchedule(ss SubSchedule) (interface{}, error) {
} }
//assign tasks to clusters //assign tasks to clusters
resp, err := ss.AssignTask(clusters) resp, err := ss.AssignTask(clusters, mode)
if err != nil { if err != nil {
return nil, err return nil, err
} }

5
internal/scheduler/schedulers/aiScheduler.go
View File

@ -140,7 +140,7 @@ func (as *AiScheduler) PickOptimalStrategy() (strategy.Strategy, error) {
return nil, errors.New("no strategy has been chosen") return nil, errors.New("no strategy has been chosen")
} }
func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interface{}, error) { func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster, mode int) (interface{}, error) {
if clusters == nil { if clusters == nil {
return nil, errors.New("clusters is nil") return nil, errors.New("clusters is nil")
} }
@ -173,7 +173,7 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interfa
wg.Add(1) wg.Add(1)
go func() { go func() {
opt, _ := cloneAiOption(as.option) opt, _ := cloneAiOption(as.option)
resp, err := executorMap[c.ClusterId].Execute(as.ctx, opt) resp, err := executorMap[c.ClusterId].Execute(as.ctx, opt, mode)
if err != nil { if err != nil {
e := struct { e := struct {
err error err error
@ -227,6 +227,7 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interfa
return nil, err return nil, err
} }
// aiTasks
adapterName, err := as.AiStorages.GetAdapterNameById(as.option.AdapterId) adapterName, err := as.AiStorages.GetAdapterNameById(as.option.AdapterId)
if err != nil { if err != nil {
return nil, err return nil, err

2
internal/scheduler/schedulers/cloudScheduler.go
View File

@ -114,7 +114,7 @@ func (as *CloudScheduler) PickOptimalStrategy() (strategy.Strategy, error) {
return nil, errors.New("no strategy has been chosen") return nil, errors.New("no strategy has been chosen")
} }
func (as *CloudScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interface{}, error) { func (as *CloudScheduler) AssignTask(clusters []*strategy.AssignedCluster, mode int) (interface{}, error) {
if clusters == nil { if clusters == nil {
return nil, errors.New("clusters is nil") return nil, errors.New("clusters is nil")
} }

2
internal/scheduler/schedulers/hpcScheduler.go
View File

@ -50,6 +50,6 @@ func (h *HpcScheduler) genTaskAndProviders(task *response.TaskInfo) (*providerPr
return nil, nil return nil, nil
} }
func (h *HpcScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interface{}, error) { func (h *HpcScheduler) AssignTask(clusters []*strategy.AssignedCluster, mode int) (interface{}, error) {
return nil, nil return nil, nil
} }

2
internal/scheduler/schedulers/vmScheduler.go
View File

@ -114,7 +114,7 @@ func (vm *VmScheduler) genTaskAndProviders() (*providerPricing.Task, []*provider
return nil, providerList, nil return nil, providerList, nil
} }
func (as *VmScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interface{}, error) { func (as *VmScheduler) AssignTask(clusters []*strategy.AssignedCluster, mode int) (interface{}, error) {
//TODO implement me //TODO implement me
if clusters == nil { if clusters == nil {
return nil, errors.New("clusters is nil") return nil, errors.New("clusters is nil")

2
internal/scheduler/service/executor/aiExecutor.go
View File

@ -6,5 +6,5 @@ import (
) )
type AiExecutor interface { type AiExecutor interface {
Execute(ctx context.Context, option *option.AiOption) (interface{}, error) Execute(ctx context.Context, option *option.AiOption, mode int) (interface{}, error)
} }

2
internal/storeLink/modelarts.go
View File

@ -570,7 +570,7 @@ func (m *ModelArtsLink) GetTrainingTask(ctx context.Context, taskId string) (*co
return &task, nil return &task, nil
} }
func (m *ModelArtsLink) Execute(ctx context.Context, option *option.AiOption) (interface{}, error) { func (m *ModelArtsLink) Execute(ctx context.Context, option *option.AiOption, mode int) (interface{}, error) {
err := m.GenerateSubmitParams(ctx, option) err := m.GenerateSubmitParams(ctx, option)
if err != nil { if err != nil {
return nil, err return nil, err

2
internal/storeLink/octopus.go
View File

@ -555,7 +555,7 @@ func (o *OctopusLink) GetTrainingTask(ctx context.Context, taskId string) (*coll
return &task, nil return &task, nil
} }
func (o *OctopusLink) Execute(ctx context.Context, option *option.AiOption) (interface{}, error) { func (o *OctopusLink) Execute(ctx context.Context, option *option.AiOption, mode int) (interface{}, error) {
err := o.GenerateSubmitParams(ctx, option) err := o.GenerateSubmitParams(ctx, option)
if err != nil { if err != nil {
return nil, err return nil, err

2
internal/storeLink/openi.go
View File

@ -47,7 +47,7 @@ func NewOpenI(host string, id int64, name string, token string) *OpenI {
} }
} }
func (o OpenI) Execute(ctx context.Context, option *option.AiOption) (interface{}, error) { func (o OpenI) Execute(ctx context.Context, option *option.AiOption, mode int) (interface{}, error) {
return nil, errors.New("failed to implement") return nil, errors.New("failed to implement")
} }

2
internal/storeLink/shuguangai.go
View File

@ -587,7 +587,7 @@ func (s *ShuguangAi) GetTrainingTask(ctx context.Context, taskId string) (*colle
return &task, nil return &task, nil
} }
func (s *ShuguangAi) Execute(ctx context.Context, option *option.AiOption) (interface{}, error) { func (s *ShuguangAi) Execute(ctx context.Context, option *option.AiOption, mode int) (interface{}, error) {
err := s.GenerateSubmitParams(ctx, option) err := s.GenerateSubmitParams(ctx, option)
if err != nil { if err != nil {
return nil, err return nil, err

2
internal/storeLink/template.go
View File

@ -28,7 +28,7 @@ func NewTemplate(host string, id int64, name string, token string) *Template {
} }
// Execute 执行任务 // Execute 执行任务
func (o Template) Execute(ctx context.Context, option *option.AiOption) (interface{}, error) { func (o Template) Execute(ctx context.Context, option *option.AiOption, mode int) (interface{}, error) {
return nil, errors.New("failed to implement") return nil, errors.New("failed to implement")
} }