mirror of
https://github.com/go-gitea/gitea.git
synced 2026-10-22 01:43:20 +02:00
perf(actions): batch per-row queries in run, job and cancellation paths (#39723)
Several Actions code paths issued one or more queries per listed row. They now load in batch, with unchanged output. - API job list: tasks, steps and runners load in batch - API run list: latest attempts and their trigger users load in batch - Repository runs page: the jobs of all active runs load in one query - Cancelling previous runs and replacing concurrency group entries fetch their jobs in one query Per call with 50 rows on SQLite, base vs head: | Path | Queries | Time | | --- | --- | --- | | API job list | 254 → 6 | 12.9 → 1.9 ms | | API run list | 305 → 207 | 11.7 → 6.2 ms | | Runs page job loading | 100 → 1 | 6.5 → 0.47 ms | | `CancelPreviousJobs` | 546 → 496 | ~120 ms, per-job writes dominate | The remaining run list queries come from per-run repository conversion, which this PR doesn't change. --------- Co-authored-by: Zettat123 <zettat123@gmail.com> Co-authored-by: silverwind <me@silverwind.io>
This commit is contained in:
10 files changed
+241
-121
No files matched your search
+16
-14
@@ -15,6 +15,7 @@ import (
|
||||
"gitea.dev/models/db"
|
||||
repo_model "gitea.dev/models/repo"
|
||||
user_model "gitea.dev/models/user"
|
||||
"gitea.dev/modules/container"
|
||||
"gitea.dev/modules/git"
|
||||
"gitea.dev/modules/json"
|
||||
"gitea.dev/modules/log"
|
||||
@@ -408,26 +409,27 @@ func CancelPreviousJobsByRunConcurrency(ctx context.Context, attempt *ActionRunA
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var jobsToCancel []*ActionRunJob
|
||||
|
||||
attempts, jobs, err := getConcurrencyEntriesToReplace(ctx, attempt.RepoID, attempt.ConcurrencyGroup, attempt.ConcurrencyCancel)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("find concurrent runs and jobs: %w", err)
|
||||
}
|
||||
jobs = slices.DeleteFunc(jobs, func(job *ActionRunJob) bool { return job.RunID == attempt.RunID })
|
||||
jobsToCancel = append(jobsToCancel, jobs...)
|
||||
|
||||
// cancel runs in the same concurrency group
|
||||
for _, concurrentAttempt := range attempts {
|
||||
if concurrentAttempt.RunID == attempt.RunID {
|
||||
continue
|
||||
}
|
||||
jobs, err := GetRunJobsByRunAndAttemptID(ctx, concurrentAttempt.RunID, concurrentAttempt.ID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("find run %d attempt %d jobs: %w", concurrentAttempt.RunID, concurrentAttempt.ID, err)
|
||||
}
|
||||
jobsToCancel = append(jobsToCancel, jobs...)
|
||||
attempts = slices.DeleteFunc(attempts, func(concurrentAttempt *ActionRunAttempt) bool { return concurrentAttempt.RunID == attempt.RunID })
|
||||
attemptJobs, err := getRunJobsByAttempts(ctx, attempts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("find concurrent attempt jobs: %w", err)
|
||||
}
|
||||
|
||||
return CancelJobs(ctx, jobsToCancel, false)
|
||||
return CancelJobs(ctx, append(jobs, attemptJobs...), false)
|
||||
}
|
||||
|
||||
func getRunJobsByAttempts(ctx context.Context, attempts []*ActionRunAttempt) (jobs []*ActionRunJob, _ error) {
|
||||
for chunk := range slices.Chunk(attempts, db.DefaultMaxInSize) {
|
||||
attemptIDs := container.FilterSlice(chunk, func(attempt *ActionRunAttempt) (int64, bool) { return attempt.ID, true })
|
||||
if err := db.GetEngine(ctx).In("run_attempt_id", attemptIDs).Find(&jobs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return jobs, nil
|
||||
}
|
||||
+29
-43
@@ -271,6 +271,24 @@ func GetLatestAttemptJobsByRun(ctx context.Context, run *ActionRun) (ActionJobLi
|
||||
return jobs, nil
|
||||
}
|
||||
|
||||
func GetLatestAttemptJobsByRuns(ctx context.Context, runs []*ActionRun) (map[int64]ActionJobList, error) {
|
||||
result := make(map[int64]ActionJobList, len(runs))
|
||||
if len(runs) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
attemptIDs := container.FilterSlice(runs, func(run *ActionRun) (int64, bool) { return run.LatestAttemptID, run.LatestAttemptID != 0 })
|
||||
legacyRunIDs := container.FilterSlice(runs, func(run *ActionRun) (int64, bool) { return run.ID, run.LatestAttemptID == 0 })
|
||||
cond := builder.In("run_attempt_id", attemptIDs).Or(builder.Eq{"run_attempt_id": 0}.And(builder.In("run_id", legacyRunIDs)))
|
||||
var jobs []*ActionRunJob
|
||||
if err := db.GetEngine(ctx).Where(cond).OrderBy("id").Find(&jobs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, job := range jobs {
|
||||
result[job.RunID] = append(result[job.RunID], job)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// GetAllRunJobsByRepoAndRunID returns all jobs for a run across all attempts.
|
||||
func GetAllRunJobsByRepoAndRunID(ctx context.Context, repoID, runID int64) (ActionJobList, error) {
|
||||
var jobs []*ActionRunJob
|
||||
@@ -685,43 +703,18 @@ func hasFailFastMatrixFailure(jobs []*ActionRunJob) bool {
|
||||
// It's useful when a new run is triggered, and all previous runs needn't be continued anymore.
|
||||
func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) ([]*ActionRunJob, error) {
|
||||
// Find all runs in the specified repository, reference, and workflow with non-final status
|
||||
runs, total, err := db.FindAndCount[ActionRun](ctx, FindRunOptions{
|
||||
runIDs := builder.Select("id").From("action_run").Where(FindRunOptions{
|
||||
RepoID: repoID,
|
||||
Ref: ref,
|
||||
WorkflowID: workflowID,
|
||||
TriggerEvent: event,
|
||||
Status: []Status{StatusRunning, StatusWaiting, StatusBlocked, StatusCancelling},
|
||||
})
|
||||
if err != nil {
|
||||
}.ToConds())
|
||||
var jobs []*ActionRunJob
|
||||
if err := db.GetEngine(ctx).Where(builder.In("run_id", runIDs)).Find(&jobs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// If there are no runs found, there's no need to proceed with cancellation, so return nil.
|
||||
if total == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
cancelledJobs := make([]*ActionRunJob, 0, total)
|
||||
|
||||
// Iterate over each found run and cancel its associated jobs.
|
||||
for _, run := range runs {
|
||||
// Find all jobs associated with the current run.
|
||||
jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
|
||||
RunID: run.ID,
|
||||
})
|
||||
if err != nil {
|
||||
return cancelledJobs, err
|
||||
}
|
||||
|
||||
cjs, err := CancelJobs(ctx, jobs, false)
|
||||
if err != nil {
|
||||
return cancelledJobs, err
|
||||
}
|
||||
cancelledJobs = append(cancelledJobs, cjs...)
|
||||
}
|
||||
|
||||
// Return nil to indicate successful cancellation of all running and waiting jobs.
|
||||
return cancelledJobs, nil
|
||||
return CancelJobs(ctx, jobs, false)
|
||||
}
|
||||
|
||||
// GetAncestorCallerIDs returns the IDs of the reusable workflow callers the job is nested in.
|
||||
@@ -738,7 +731,7 @@ func GetAncestorCallerIDs(ctx context.Context, job *ActionRunJob) (container.Set
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) (jobsToCancel []*ActionRunJob, _ error) {
|
||||
func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) ([]*ActionRunJob, error) {
|
||||
if job.RawConcurrency == "" {
|
||||
return nil, nil
|
||||
}
|
||||
@@ -758,21 +751,14 @@ func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob)
|
||||
return nil, err
|
||||
}
|
||||
jobs = slices.DeleteFunc(jobs, func(j *ActionRunJob) bool { return j.ID == job.ID || callerIDs.Contains(j.ID) })
|
||||
jobsToCancel = append(jobsToCancel, jobs...)
|
||||
|
||||
// cancel runs in the same concurrency group
|
||||
for _, attempt := range attempts {
|
||||
if attempt.ID == job.RunAttemptID {
|
||||
continue
|
||||
}
|
||||
jobs, err := GetRunJobsByRunAndAttemptID(ctx, attempt.RunID, attempt.ID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("find run %d attempt %d jobs: %w", attempt.RunID, attempt.ID, err)
|
||||
}
|
||||
jobsToCancel = append(jobsToCancel, jobs...)
|
||||
attempts = slices.DeleteFunc(attempts, func(attempt *ActionRunAttempt) bool { return attempt.ID == job.RunAttemptID })
|
||||
attemptJobs, err := getRunJobsByAttempts(ctx, attempts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("find concurrent attempt jobs: %w", err)
|
||||
}
|
||||
|
||||
return CancelJobs(ctx, jobsToCancel, false)
|
||||
return CancelJobs(ctx, append(jobs, attemptJobs...), false)
|
||||
}
|
||||
|
||||
// CancelJobs cancels every cancellable job it is given, force skipping the graceful cancelling
|
||||
|
||||
@@ -467,3 +467,48 @@ func TestCancelJobs_CallerWaitsForCancellingChild(t *testing.T) {
|
||||
}
|
||||
assert.Equal(t, StatusCancelled, unittest.AssertExistsAndLoadBean(t, &ActionRun{ID: run.ID}).Status)
|
||||
}
|
||||
|
||||
func TestBatchLoadAndCancelRunJobs(t *testing.T) {
|
||||
require.NoError(t, unittest.PrepareTestDatabase())
|
||||
ctx := t.Context()
|
||||
|
||||
legacyRun := &ActionRun{RepoID: 4, OwnerID: 1, Index: 9600, TriggerUserID: 1, Ref: "refs/heads/batch", Status: StatusRunning}
|
||||
run := &ActionRun{RepoID: 4, OwnerID: 1, Index: 9601, TriggerUserID: 1, Ref: "refs/heads/batch", Status: StatusRunning}
|
||||
require.NoError(t, db.Insert(ctx, legacyRun, run))
|
||||
require.NoError(t, db.Insert(ctx, &ActionRunJob{RunID: legacyRun.ID, RepoID: legacyRun.RepoID, OwnerID: legacyRun.OwnerID}))
|
||||
oldAttempt := &ActionRunAttempt{RepoID: run.RepoID, RunID: run.ID, Attempt: 1, TriggerUserID: 1}
|
||||
latestAttempt := &ActionRunAttempt{RepoID: run.RepoID, RunID: run.ID, Attempt: 2, TriggerUserID: 2}
|
||||
require.NoError(t, db.Insert(ctx, oldAttempt, latestAttempt))
|
||||
for _, attemptID := range []int64{oldAttempt.ID, latestAttempt.ID, latestAttempt.ID} {
|
||||
require.NoError(t, db.Insert(ctx, &ActionRunJob{RunID: run.ID, RunAttemptID: attemptID, RepoID: run.RepoID, OwnerID: run.OwnerID}))
|
||||
}
|
||||
run.LatestAttemptID = latestAttempt.ID
|
||||
require.NoError(t, UpdateRun(ctx, run, "latest_attempt_id"))
|
||||
|
||||
runs := RunList{legacyRun, run}
|
||||
jobsByRun, err := GetLatestAttemptJobsByRuns(ctx, runs)
|
||||
require.NoError(t, err)
|
||||
for _, listedRun := range runs {
|
||||
expected, err := GetLatestAttemptJobsByRun(ctx, listedRun)
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, expected)
|
||||
assert.Equal(t, expected, jobsByRun[listedRun.ID])
|
||||
}
|
||||
|
||||
attempts, err := runs.GetLatestAttempts(ctx)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, attempts, 1)
|
||||
assert.EqualValues(t, 2, attempts[latestAttempt.ID].TriggerUser.ID)
|
||||
|
||||
chunkedAttempts := []*ActionRunAttempt{oldAttempt}
|
||||
for i := range db.DefaultMaxInSize {
|
||||
chunkedAttempts = append(chunkedAttempts, &ActionRunAttempt{ID: -int64(i + 1)})
|
||||
}
|
||||
attemptJobs, err := getRunJobsByAttempts(ctx, append(chunkedAttempts, latestAttempt))
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, attemptJobs, 3)
|
||||
|
||||
cancelled, err := CancelPreviousJobs(ctx, run.RepoID, run.Ref, "", "")
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, cancelled, 4)
|
||||
}
|
||||
@@ -5,6 +5,8 @@ package actions
|
||||
|
||||
import (
|
||||
"context"
|
||||
"maps"
|
||||
"slices"
|
||||
|
||||
"gitea.dev/models/db"
|
||||
repo_model "gitea.dev/models/repo"
|
||||
@@ -42,6 +44,21 @@ func (runs RunList) LoadTriggerUser(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetLatestAttempts returns the latest attempts of the runs keyed by attempt ID
|
||||
func (runs RunList) GetLatestAttempts(ctx context.Context) (map[int64]*ActionRunAttempt, error) {
|
||||
attemptIDs := container.FilterSlice(runs, func(run *ActionRun) (int64, bool) {
|
||||
return run.LatestAttemptID, run.LatestAttemptID != 0
|
||||
})
|
||||
attempts := make(map[int64]*ActionRunAttempt, len(attemptIDs))
|
||||
if len(attemptIDs) == 0 {
|
||||
return attempts, nil
|
||||
}
|
||||
if err := db.GetEngine(ctx).In("id", attemptIDs).Find(&attempts); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return attempts, ActionRunAttemptList(slices.Collect(maps.Values(attempts))).LoadTriggerUser(ctx)
|
||||
}
|
||||
|
||||
func (runs RunList) LoadRepos(ctx context.Context) error {
|
||||
repoIDs := container.FilterSlice(runs, func(run *ActionRun) (int64, bool) {
|
||||
return run.RepoID, run.Repo == nil
|
||||
|
||||
@@ -207,6 +207,14 @@ func init() {
|
||||
db.RegisterModel(&ActionRunner{})
|
||||
}
|
||||
|
||||
func GetRunnersMapByIDs(ctx context.Context, ids []int64) (map[int64]*ActionRunner, error) {
|
||||
runners := make(map[int64]*ActionRunner, len(ids))
|
||||
if len(ids) == 0 {
|
||||
return runners, nil
|
||||
}
|
||||
return runners, db.GetEngine(ctx).In("id", ids).Find(&runners)
|
||||
}
|
||||
|
||||
// FindRunnerOptions
|
||||
// ownerID == 0 and repoID == 0 means any runner including global runners
|
||||
// repoID != 0 and WithAvailable == false means any runner for the given repo
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"gitea.dev/models/db"
|
||||
"gitea.dev/modules/container"
|
||||
"gitea.dev/modules/timeutil"
|
||||
"gitea.dev/modules/util"
|
||||
|
||||
"xorm.io/builder"
|
||||
)
|
||||
@@ -41,6 +42,29 @@ func (tasks TaskList) LoadJobs(ctx context.Context) error {
|
||||
return jobsList.LoadAttributes(ctx, true)
|
||||
}
|
||||
|
||||
func (tasks TaskList) LoadSteps(ctx context.Context) error {
|
||||
taskIDs := container.FilterSlice(tasks, func(task *ActionTask) (int64, bool) {
|
||||
return task.ID, task.Steps == nil
|
||||
})
|
||||
if len(taskIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
var steps []*ActionTaskStep
|
||||
if err := db.GetEngine(ctx).In("task_id", taskIDs).OrderBy("task_id, `index`").Find(&steps); err != nil {
|
||||
return err
|
||||
}
|
||||
stepsByTask := make(map[int64][]*ActionTaskStep, len(taskIDs))
|
||||
for _, step := range steps {
|
||||
stepsByTask[step.TaskID] = append(stepsByTask[step.TaskID], step)
|
||||
}
|
||||
for _, task := range tasks {
|
||||
if task.Steps == nil {
|
||||
task.Steps = util.SliceNilAsEmpty(stepsByTask[task.ID])
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (tasks TaskList) LoadAttributes(ctx context.Context) error {
|
||||
return tasks.LoadJobs(ctx)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user