mirror/act_runner
1
0
Fork 0
mirror of https://gitea.com/gitea/act_runner.git synced 2025-02-06 09:54:48 +01:00
Code Issues Packages Projects Releases Wiki Activity

use atomic package

Browse Source
This commit is contained in:
sillyguodong 2023-07-24 14:02:09 +08:00
parent bf75f0c248
commit 7cd926e32d
No known key found for this signature in database
GPG Key ID: 4A0646FA86256DF4
1 changed files with 10 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
internal/app/poll/poller.go
View File

@ -8,6 +8,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"sync" "sync"
"sync/atomic"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1" runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
"github.com/bufbuild/connect-go" "github.com/bufbuild/connect-go"
@ -23,11 +24,9 @@ type Poller struct {
client client.Client client client.Client
runner *run.Runner runner *run.Runner
cfg *config.Config cfg *config.Config
tasksVersion atomic.Int64 // tasksVersion used to store the version of the last task fetched from the Gitea.
} }
// tasksVersion used to store the version of the last task fetched from the Gitea.
var tasksVersion int64
func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller { func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
return &Poller{ return &Poller{
client: client, client: client,
@ -81,7 +80,7 @@ func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
defer cancel() defer cancel()
resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{ resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{
TasksVersion: tasksVersion, TasksVersion: p.tasksVersion.Load(),
})) }))
if errors.Is(err, context.DeadlineExceeded) { if errors.Is(err, context.DeadlineExceeded) {
err = nil err = nil
@ -95,14 +94,16 @@ func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
return nil, false return nil, false
} }
tasksVersion = resp.Msg.TasksVersion if resp.Msg.TasksVersion > p.tasksVersion.Load() {
p.tasksVersion.Store(resp.Msg.TasksVersion)
}
if resp.Msg.Task == nil { if resp.Msg.Task == nil {
return nil, false return nil, false
} }
// got a task, set `tasksVersion` to zero to focre query db in next request. // got a task, set `tasksVersion` to zero to focre query db in next request.
tasksVersion = 0 p.tasksVersion.Store(resp.Msg.TasksVersion)
return resp.Msg.Task, true return resp.Msg.Task, true
} }