Reduce unnecessary DB queries for Actions tasks (#25199)

Close #24544

Changes:

- Create `action_tasks_version` table to store the latest version of
each scope (global, org and repo).
- When a job with the status of `waiting` is created, the tasks version
of the scopes it belongs to will increase.
- When the status of a job already in the database is updated to
`waiting`, the tasks version of the scopes it belongs to will increase.
- On Gitea side, in `FeatchTask()`, will try to query the
`action_tasks_version` record of the scope of the runner that call
`FetchTask()`. If the record does not exist, will insert a row. Then,
Gitea will compare the version passed from runner to Gitea with the
version in database, if inconsistent, try pick task. Gitea always
returns the latest version from database to the runner.

Related:

- Protocol: https://gitea.com/gitea/actions-proto-def/pulls/10
- Runner: https://gitea.com/gitea/act_runner/pulls/219
This commit is contained in:
sillyguodong 2023-07-24 14:11:27 +08:00 committed by GitHub
parent 674df05b16
commit f5c7d4cfdd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 174 additions and 9 deletions

View file

@ -195,6 +195,7 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
} }
runJobs := make([]*ActionRunJob, 0, len(jobs)) runJobs := make([]*ActionRunJob, 0, len(jobs))
var hasWaiting bool
for _, v := range jobs { for _, v := range jobs {
id, job := v.Job() id, job := v.Job()
needs := job.Needs() needs := job.Needs()
@ -205,6 +206,8 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
status := StatusWaiting status := StatusWaiting
if len(needs) > 0 || run.NeedApproval { if len(needs) > 0 || run.NeedApproval {
status = StatusBlocked status = StatusBlocked
} else {
hasWaiting = true
} }
job.Name, _ = util.SplitStringAtByteN(job.Name, 255) job.Name, _ = util.SplitStringAtByteN(job.Name, 255)
runJobs = append(runJobs, &ActionRunJob{ runJobs = append(runJobs, &ActionRunJob{
@ -225,6 +228,13 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
return err return err
} }
// if there is a job in the waiting status, increase tasks version.
if hasWaiting {
if err := IncreaseTaskVersion(ctx, run.OwnerID, run.RepoID); err != nil {
return err
}
}
return commiter.Commit() return commiter.Commit()
} }

View file

@ -111,6 +111,13 @@ func UpdateRunJob(ctx context.Context, job *ActionRunJob, cond builder.Cond, col
return affected, nil return affected, nil
} }
if affected != 0 && util.SliceContains(cols, "status") && job.Status.IsWaiting() {
// if the status of job changes to waiting again, increase tasks version.
if err := IncreaseTaskVersion(ctx, job.OwnerID, job.RepoID); err != nil {
return affected, err
}
}
if job.RunID == 0 { if job.RunID == 0 {
var err error var err error
if job, err = GetRunJobByID(ctx, job.ID); err != nil { if job, err = GetRunJobByID(ctx, job.ID); err != nil {

View file

@ -215,12 +215,11 @@ func GetRunningTaskByToken(ctx context.Context, token string) (*ActionTask, erro
} }
func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (*ActionTask, bool, error) { func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (*ActionTask, bool, error) {
dbCtx, commiter, err := db.TxContext(ctx) ctx, commiter, err := db.TxContext(ctx)
if err != nil { if err != nil {
return nil, false, err return nil, false, err
} }
defer commiter.Close() defer commiter.Close()
ctx = dbCtx.WithContext(ctx)
e := db.GetEngine(ctx) e := db.GetEngine(ctx)

View file

@ -0,0 +1,105 @@
// Copyright 2023 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package actions
import (
"context"
"code.gitea.io/gitea/models/db"
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/timeutil"
)
// ActionTasksVersion
// If both ownerID and repoID is zero, its scope is global.
// If ownerID is not zero and repoID is zero, its scope is org (there is no user-level runner currrently).
// If ownerID is zero and repoID is not zero, its scope is repo.
type ActionTasksVersion struct {
ID int64 `xorm:"pk autoincr"`
OwnerID int64 `xorm:"UNIQUE(owner_repo)"`
RepoID int64 `xorm:"INDEX UNIQUE(owner_repo)"`
Version int64
CreatedUnix timeutil.TimeStamp `xorm:"created"`
UpdatedUnix timeutil.TimeStamp `xorm:"updated"`
}
func init() {
db.RegisterModel(new(ActionTasksVersion))
}
func GetTasksVersionByScope(ctx context.Context, ownerID, repoID int64) (int64, error) {
var tasksVersion ActionTasksVersion
has, err := db.GetEngine(ctx).Where("owner_id = ? AND repo_id = ?", ownerID, repoID).Get(&tasksVersion)
if err != nil {
return 0, err
} else if !has {
return 0, nil
}
return tasksVersion.Version, err
}
func insertTasksVersion(ctx context.Context, ownerID, repoID int64) (*ActionTasksVersion, error) {
tasksVersion := &ActionTasksVersion{
OwnerID: ownerID,
RepoID: repoID,
Version: 1,
}
if _, err := db.GetEngine(ctx).Insert(tasksVersion); err != nil {
return nil, err
}
return tasksVersion, nil
}
func increaseTasksVersionByScope(ctx context.Context, ownerID, repoID int64) error {
result, err := db.GetEngine(ctx).Exec("UPDATE action_tasks_version SET version = version + 1 WHERE owner_id = ? AND repo_id = ?", ownerID, repoID)
if err != nil {
return err
}
affected, err := result.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
// if update sql does not affect any rows, the database may be broken,
// so re-insert the row of version data here.
if _, err := insertTasksVersion(ctx, ownerID, repoID); err != nil {
return err
}
}
return nil
}
func IncreaseTaskVersion(ctx context.Context, ownerID, repoID int64) error {
ctx, commiter, err := db.TxContext(ctx)
if err != nil {
return err
}
defer commiter.Close()
// 1. increase global
if err := increaseTasksVersionByScope(ctx, 0, 0); err != nil {
log.Error("IncreaseTasksVersionByScope(Global): %v", err)
return err
}
// 2. increase owner
if ownerID > 0 {
if err := increaseTasksVersionByScope(ctx, ownerID, 0); err != nil {
log.Error("IncreaseTasksVersionByScope(Owner): %v", err)
return err
}
}
// 3. increase repo
if repoID > 0 {
if err := increaseTasksVersionByScope(ctx, 0, repoID); err != nil {
log.Error("IncreaseTasksVersionByScope(Repo): %v", err)
return err
}
}
return commiter.Commit()
}

View file

@ -515,6 +515,8 @@ var migrations = []Migration{
NewMigration("Alter Actions Artifact table", v1_21.AlterActionArtifactTable), NewMigration("Alter Actions Artifact table", v1_21.AlterActionArtifactTable),
// v266 -> v267 // v266 -> v267
NewMigration("Reduce commit status", v1_21.ReduceCommitStatus), NewMigration("Reduce commit status", v1_21.ReduceCommitStatus),
// v267 -> v268
NewMigration("Add action_tasks_version table", v1_21.CreateActionTasksVersionTable),
} }
// GetCurrentDBVersion returns the current db version // GetCurrentDBVersion returns the current db version

View file

@ -0,0 +1,23 @@
// Copyright 2023 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package v1_21 //nolint
import (
"code.gitea.io/gitea/modules/timeutil"
"xorm.io/xorm"
)
func CreateActionTasksVersionTable(x *xorm.Engine) error {
type ActionTasksVersion struct {
ID int64 `xorm:"pk autoincr"`
OwnerID int64 `xorm:"UNIQUE(owner_repo)"`
RepoID int64 `xorm:"INDEX UNIQUE(owner_repo)"`
Version int64
CreatedUnix timeutil.TimeStamp `xorm:"created"`
UpdatedUnix timeutil.TimeStamp `xorm:"updated"`
}
return x.Sync(new(ActionTasksVersion))
}

View file

@ -127,20 +127,39 @@ func (s *Service) Declare(
// FetchTask assigns a task to the runner // FetchTask assigns a task to the runner
func (s *Service) FetchTask( func (s *Service) FetchTask(
ctx context.Context, ctx context.Context,
_ *connect.Request[runnerv1.FetchTaskRequest], req *connect.Request[runnerv1.FetchTaskRequest],
) (*connect.Response[runnerv1.FetchTaskResponse], error) { ) (*connect.Response[runnerv1.FetchTaskResponse], error) {
runner := GetRunner(ctx) runner := GetRunner(ctx)
var task *runnerv1.Task var task *runnerv1.Task
tasksVersion := req.Msg.TasksVersion // task version from runner
latestVersion, err := actions_model.GetTasksVersionByScope(ctx, runner.OwnerID, runner.RepoID)
if err != nil {
return nil, status.Errorf(codes.Internal, "query tasks version failed: %v", err)
} else if latestVersion == 0 {
if err := actions_model.IncreaseTaskVersion(ctx, runner.OwnerID, runner.RepoID); err != nil {
return nil, status.Errorf(codes.Internal, "fail to increase task version: %v", err)
}
// if we don't increase the value of `latestVersion` here,
// the response of FetchTask will return tasksVersion as zero.
// and the runner will treat it as an old version of Gitea.
latestVersion++
}
if tasksVersion != latestVersion {
// if the task version in request is not equal to the version in db,
// it means there may still be some tasks not be assgined.
// try to pick a task for the runner that send the request.
if t, ok, err := pickTask(ctx, runner); err != nil { if t, ok, err := pickTask(ctx, runner); err != nil {
log.Error("pick task failed: %v", err) log.Error("pick task failed: %v", err)
return nil, status.Errorf(codes.Internal, "pick task: %v", err) return nil, status.Errorf(codes.Internal, "pick task: %v", err)
} else if ok { } else if ok {
task = t task = t
} }
}
res := connect.NewResponse(&runnerv1.FetchTaskResponse{ res := connect.NewResponse(&runnerv1.FetchTaskResponse{
Task: task, Task: task,
TasksVersion: latestVersion,
}) })
return res, nil return res, nil
} }