studio/flamenco
studio/flamenco
11
29
Fork 38
Code Issues 62 Pull Requests 8 Releases Wiki Activity

WIP: Allow jobs to be submitted in paused status #104318

Closed
David Zhang wants to merge 32 commits from David-Zhang-10/flamenco:submit-as-paused into main
pull from: David-Zhang-10/flamenco:submit-as-paused
merge into: studio:main

When changing the target branch, be careful to rebase the branch in your fork to match. See documentation.
Conversation 0 Commits 32 Files Changed 23 +3 -14
1 changed files with 3 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit d510eb390c - Show all commits

17
internal/manager/task_state_machine/task_state_machine.go
View File

@ -181,9 +181,7 @@ func (sm *StateMachine) updateJobOnTaskStatusCanceled(ctx context.Context, logge
} }
// Deal with the special case when the job is in pause-requested status. // Deal with the special case when the job is in pause-requested status.
if job.Status != api.JobStatusPauseRequested { if job.Status == api.JobStatusPauseRequested {
return nil
} else {
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive) numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive)
if err != nil { if err != nil {
return err return err
@ -220,10 +218,7 @@ func (sm *StateMachine) updateJobOnTaskStatusFailed(ctx context.Context, logger
// If the job didn't fail, this failure indicates that at least the job is active. // If the job didn't fail, this failure indicates that at least the job is active.
failLogger.Info().Msg("task failed, but not enough to fail the job") failLogger.Info().Msg("task failed, but not enough to fail the job")
if job.Status != api.JobStatusPauseRequested { if job.Status == api.JobStatusPauseRequested {
return sm.jobStatusIfAThenB(ctx, logger, job, api.JobStatusQueued, api.JobStatusActive,
"task failed, but not enough to fail the job")
} else {
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive) numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive)
if err != nil { if err != nil {
return err return err
@ -250,13 +245,7 @@ func (sm *StateMachine) updateJobOnTaskStatusCompleted(ctx context.Context, logg
return sm.JobStatusChange(ctx, job, api.JobStatusCompleted, "all tasks completed") return sm.JobStatusChange(ctx, job, api.JobStatusCompleted, "all tasks completed")
} }
if job.Status != api.JobStatusPauseRequested { if job.Status == api.JobStatusPauseRequested {
logger.Info().
Int("taskNumTotal", numTotal).
Int("taskNumComplete", numComplete).
Msg("task completed; there are more tasks to do")
return sm.jobStatusIfAThenB(ctx, logger, job, api.JobStatusQueued, api.JobStatusActive, "no more tasks to do")
} else {
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive) numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive)
if err != nil { if err != nil {
return err return err