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

fix task updatedtime bugs 

Former-commit-id: 226201747e9a5aabd47cb93a57d66448df3248b5
This commit is contained in:
tzwang 2024-05-19 22:52:41 +08:00
parent ed1929d356
commit 6c724b4ab9
3 changed files with 117 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/internal/logic/ai/getcenteroverviewlogic.go
View File

@ -76,7 +76,7 @@ func (l *GetCenterOverviewLogic) GetCenterOverview() (resp *types.CenterOverview
select {
case _ = <-ch:
return resp, nil
case <-time.After(2 * time.Second):
case <-time.After(1 * time.Second):
return resp, nil
}

2
api/internal/logic/ai/getcentertasklistlogic.go
View File

@ -101,7 +101,7 @@ func (l *GetCenterTaskListLogic) updateAiTaskStatus(mu *sync.RWMutex, ch chan<-
}
for _, task := range taskList {
t := task
if t.Status == constants.Completed || task.Status == constants.Failed {
if t.Status == constants.Completed || task.Status == constants.Failed || task.Status == constants.Stopped {
continue
}
wg.Add(1)

214
api/internal/logic/core/pagelisttasklogic.go
View File

@ -86,104 +86,8 @@ func (l *PageListTaskLogic) PageListTask(req *types.PageTaskReq) (resp *types.Pa
}
func (l *PageListTaskLogic) updateTaskStatus(tasks []*types.TaskModel, ch chan<- struct{}) {
for _, task := range tasks {
if task.AdapterTypeDict != 1 {
continue
}
if task.Status == constants.Succeeded || task.Status == constants.Failed {
continue
}
var aiTask []*models.TaskAi
tx := l.svcCtx.DbEngin.Raw("select * from task_ai where `task_id` = ? ", task.Id).Scan(&aiTask)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
return
}
if len(aiTask) == 0 {
continue
}
if len(aiTask) == 1 {
task.Status = aiTask[0].Status
task.StartTime = aiTask[0].StartTime
task.EndTime = aiTask[0].EndTime
tx = l.svcCtx.DbEngin.Model(task).Table("task").Where("deleted_at is null").Updates(task)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
return
}
continue
}
for i := len(aiTask) - 1; i >= 0; i-- {
if aiTask[i].StartTime == "" {
aiTask = append(aiTask[:i], aiTask[i+1:]...)
}
}
start, _ := time.ParseInLocation(constants.Layout, aiTask[0].StartTime, time.Local)
end, _ := time.ParseInLocation(constants.Layout, aiTask[0].EndTime, time.Local)
var status string
var count int
for _, a := range aiTask {
s, _ := time.ParseInLocation(constants.Layout, a.StartTime, time.Local)
e, _ := time.ParseInLocation(constants.Layout, a.EndTime, time.Local)
if s.Before(start) {
start = s
}
if e.After(end) {
end = e
}
if a.Status == constants.Failed {
status = a.Status
break
}
if a.Status == constants.Pending {
status = a.Status
continue
}
if a.Status == constants.Running {
status = a.Status
continue
}
if a.Status == constants.Completed {
count++
continue
}
}
if count == len(aiTask) {
status = constants.Succeeded
}
if status != "" {
task.Status = status
task.StartTime = start.Format(constants.Layout)
task.EndTime = end.Format(constants.Layout)
}
task.UpdatedTime = time.Now().Format(constants.Layout)
tx = l.svcCtx.DbEngin.Table("task").Model(task).Updates(task)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
return
}
}
ch <- struct{}{}
}
func (l *PageListTaskLogic) updateAiTaskStatus(tasks []*types.TaskModel, ch chan<- struct{}) {
for i := len(tasks) - 1; i >= 0; i-- {
if tasks[i].AdapterTypeDict == 0 || tasks[i].Status == constants.Succeeded || tasks[i].Status == constants.Failed {
if tasks[i].AdapterTypeDict != 1 || tasks[i].Status == constants.Succeeded || tasks[i].Status == constants.Failed {
tasks = append(tasks[:i], tasks[i+1:]...)
}
}
@ -197,7 +101,118 @@ func (l *PageListTaskLogic) updateAiTaskStatus(tasks []*types.TaskModel, ch chan
for i, _ := range tasks {
earliest, _ := time.Parse(constants.Layout, task.UpdatedTime)
latest, _ := time.Parse(constants.Layout, tasks[i].UpdatedTime)
if earliest.Before(latest) {
if latest.Before(earliest) {
task = tasks[i]
}
}
var aiTask []*models.TaskAi
tx := l.svcCtx.DbEngin.Raw("select * from task_ai where `task_id` = ? ", task.Id).Scan(&aiTask)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
ch <- struct{}{}
return
}
if len(aiTask) == 0 {
ch <- struct{}{}
return
}
if len(aiTask) == 1 {
task.Status = aiTask[0].Status
task.StartTime = aiTask[0].StartTime
task.EndTime = aiTask[0].EndTime
tx = l.svcCtx.DbEngin.Model(task).Table("task").Where("deleted_at is null").Updates(task)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
return
}
ch <- struct{}{}
return
}
for i := len(aiTask) - 1; i >= 0; i-- {
if aiTask[i].StartTime == "" {
aiTask = append(aiTask[:i], aiTask[i+1:]...)
}
}
start, _ := time.ParseInLocation(constants.Layout, aiTask[0].StartTime, time.Local)
end, _ := time.ParseInLocation(constants.Layout, aiTask[0].EndTime, time.Local)
var status string
var count int
for _, a := range aiTask {
s, _ := time.ParseInLocation(constants.Layout, a.StartTime, time.Local)
e, _ := time.ParseInLocation(constants.Layout, a.EndTime, time.Local)
if s.Before(start) {
start = s
}
if e.After(end) {
end = e
}
if a.Status == constants.Failed {
status = a.Status
break
}
if a.Status == constants.Pending {
status = a.Status
continue
}
if a.Status == constants.Running {
status = a.Status
continue
}
if a.Status == constants.Completed {
count++
continue
}
}
if count == len(aiTask) {
status = constants.Succeeded
}
if status != "" {
task.Status = status
task.StartTime = start.Format(constants.Layout)
task.EndTime = end.Format(constants.Layout)
}
task.UpdatedTime = time.Now().Format(constants.Layout)
tx = l.svcCtx.DbEngin.Table("task").Model(task).Updates(task)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
ch <- struct{}{}
return
}
ch <- struct{}{}
}
func (l *PageListTaskLogic) updateAiTaskStatus(tasks []*types.TaskModel, ch chan<- struct{}) {
for i := len(tasks) - 1; i >= 0; i-- {
if tasks[i].AdapterTypeDict != 1 || tasks[i].Status == constants.Succeeded || tasks[i].Status == constants.Failed {
tasks = append(tasks[:i], tasks[i+1:]...)
}
}
if len(tasks) == 0 {
ch <- struct{}{}
return
}
task := tasks[0]
for i, _ := range tasks {
earliest, _ := time.Parse(constants.Layout, task.UpdatedTime)
latest, _ := time.Parse(constants.Layout, tasks[i].UpdatedTime)
if latest.Before(earliest) {
task = tasks[i]
}
}
@ -206,6 +221,7 @@ func (l *PageListTaskLogic) updateAiTaskStatus(tasks []*types.TaskModel, ch chan
tx := l.svcCtx.DbEngin.Raw("select * from task_ai where `task_id` = ? ", task.Id).Scan(&aiTaskList)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
ch <- struct{}{}
return
}
@ -217,7 +233,7 @@ func (l *PageListTaskLogic) updateAiTaskStatus(tasks []*types.TaskModel, ch chan
var wg sync.WaitGroup
for _, aitask := range aiTaskList {
t := aitask
if t.Status == constants.Completed {
if t.Status == constants.Completed || t.Status == constants.Failed {
continue
}
wg.Add(1)