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

modified executor implementations 

Former-commit-id: 4cf3a7804d5668bb0e1c3bd09891225b310598b9
This commit is contained in:
tzwang 2024-02-01 16:29:55 +08:00
parent 3cc16100fa
commit c18e5a484c
9 changed files with 52 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/internal/logic/storelink/submitlinktasklogic.go
View File

@ -67,7 +67,7 @@ func (l *SubmitLinkTaskLogic) SubmitLinkTask(req *types.SubmitLinkTaskReq) (resp
envs = append(envs, env)
}
}
task, err := storelink.ILinkage.SubmitTask(req.ImageId, req.Cmd, envs, params, req.ResourceId)
task, err := storelink.ILinkage.SubmitTask(req.ImageId, req.Cmd, envs, params, req.ResourceId, "")
if err != nil {
return nil, err
}

4
api/internal/scheduler/schedulers/aiScheduler.go
View File

@ -30,7 +30,7 @@ type AiScheduler struct {
yamlString string
task *response.TaskInfo
*scheduler.Scheduler
option option.AiOption
option *option.AiOption
}
func NewAiScheduler(val string, scheduler *scheduler.Scheduler) (*AiScheduler, error) {
@ -74,7 +74,7 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) error {
executorMap := *as.AiExecutor
for _, cluster := range clusters {
_, err := executorMap[cluster.Name].Execute(option.AiOption{})
_, err := executorMap[cluster.Name].Execute(as.option)
if err != nil {
// TODO: database operation
}

23
api/internal/scheduler/schedulers/option/aiOption.go
View File

@ -1,17 +1,20 @@
package option
type AiOption struct {
aiType string // shuguangAi/octopus
resourceType string // cpu/gpu/compute card
taskType string // pytorch/tensorflow
AiType string // shuguangAi/octopus
ResourceType string // cpu/gpu/compute card
TaskType string // pytorch/tensorflow
imageId string
specId string
datasetsId string
codeId string
ImageId string
SpecId string
DatasetsId string
CodeId string
ResourceId string
cmd string
Cmd string
Envs []string
Params []string
datasets string
code string
Datasets string
Code string
}

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

@ -6,6 +6,6 @@ import (
)
type AiExecutor interface {
Execute(option option.AiOption) (interface{}, error)
Execute(option *option.AiOption) (interface{}, error)
storeLink.Linkage
}

10
api/internal/storeLink/modelarts.go
View File

@ -63,7 +63,7 @@ func (o *ModelArtsLink) QueryImageList() (interface{}, error) {
return resp, nil
}
func (o *ModelArtsLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string) (interface{}, error) {
func (o *ModelArtsLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) {
// modelArts提交任务
environments := make(map[string]string)
parameters := make([]*modelarts.ParametersTrainJob, 0)
@ -153,6 +153,10 @@ func (o *ModelArtsLink) GetResourceSpecs() (*collector.ResourceSpecs, error) {
return nil, nil
}
func (o *ModelArtsLink) Execute(option option.AiOption) (interface{}, error) {
return nil, nil
func (o *ModelArtsLink) Execute(option *option.AiOption) (interface{}, error) {
task, err := o.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.AiType)
if err != nil {
return nil, err
}
return task, nil
}

10
api/internal/storeLink/octopus.go
View File

@ -107,7 +107,7 @@ func (o *OctopusLink) QueryImageList() (interface{}, error) {
return resp, nil
}
func (o *OctopusLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string) (interface{}, error) {
func (o *OctopusLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) {
// octopus提交任务
// python参数
@ -200,6 +200,10 @@ func (o *OctopusLink) GetResourceSpecs() (*collector.ResourceSpecs, error) {
return nil, nil
}
func (o *OctopusLink) Execute(option option.AiOption) (interface{}, error) {
return nil, nil
func (o *OctopusLink) Execute(option *option.AiOption) (interface{}, error) {
task, err := o.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.AiType)
if err != nil {
return nil, err
}
return task, nil
}

2
api/internal/storeLink/shuguangHpc.go
View File

@ -144,7 +144,7 @@ func (s ShuguangHpc) QueryImageList() (interface{}, error) {
return nil, nil
}
func (s ShuguangHpc) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string) (interface{}, error) {
func (s ShuguangHpc) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) {
// shuguangHpc提交任务
//判断是否resourceId匹配自定义资源Id

24
api/internal/storeLink/shuguangai.go
View File

@ -76,9 +76,7 @@ func (s *ShuguangAi) QueryImageList() (interface{}, error) {
return resp, nil
}
func (s *ShuguangAi) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string) (interface{}, error) {
// shuguangAi提交任务
func (s *ShuguangAi) SubmitPytorchTask(imageId string, cmd string, envs []string, params []string, resourceId string) (interface{}, error) {
//判断是否resourceId匹配自定义资源Id
if resourceId != SHUGUANGAI_CUSTOM_RESOURCE_ID {
return nil, errors.New("shuguangAi资源Id不存在")
@ -133,6 +131,18 @@ func (s *ShuguangAi) SubmitTask(imageId string, cmd string, envs []string, param
return resp, nil
}
func (s *ShuguangAi) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) {
// shuguangAi提交任务
if aiType == PYTORCH {
task, err := s.SubmitPytorchTask(imageId, cmd, envs, params, resourceId)
if err != nil {
return nil, err
}
return task, nil
}
return nil, errors.New("shuguangAi不支持的任务类型")
}
func (s *ShuguangAi) QueryTask(taskId string) (interface{}, error) {
// shuguangAi获取任务
req := &hpcAC.GetPytorchTaskReq{
@ -199,6 +209,10 @@ func (o *ShuguangAi) GetResourceSpecs() (*collector.ResourceSpecs, error) {
return nil, nil
}
func (o *ShuguangAi) Execute(option option.AiOption) (interface{}, error) {
return nil, nil
func (o *ShuguangAi) Execute(option *option.AiOption) (interface{}, error) {
task, err := o.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.AiType)
if err != nil {
return nil, err
}
return task, nil
}

2
api/internal/storeLink/storeLink.go
View File

@ -31,7 +31,7 @@ type Linkage interface {
UploadImage(path string) (interface{}, error)
DeleteImage(imageId string) (interface{}, error)
QueryImageList() (interface{}, error)
SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string) (interface{}, error)
SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error)
QueryTask(taskId string) (interface{}, error)
QuerySpecs() (interface{}, error)
DeleteTask(taskId string) (interface{}, error)