hiifong/act_runner
1
0 Fork 0
Code Issues 183 Pull Requests 10 Actions Packages Projects Releases 22 Wiki Activity

refactor: poll

Browse Source
This commit is contained in:
Jason Song committed 2023-04-04 17:35:40 +08:00
1 parent 50fb69a55c
commit 56b754269f
8 files changed
+89 -252
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cmd/daemon.go
+5 -27
View File
@@ -11,14 +11,13 @@ import (
"github.com/mattn/go-isatty"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"golang.org/x/sync/errgroup"
"gitea.com/gitea/act_runner/internal/app/poll"
"gitea.com/gitea/act_runner/internal/app/run"
"gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config"
"gitea.com/gitea/act_runner/internal/pkg/envcheck"
"gitea.com/gitea/act_runner/internal/pkg/labels"
"gitea.com/gitea/act_runner/poller"
)
func runDaemon(ctx context.Context, configFile *string) func(cmd *cobra.Command, args []string) error {
@@ -59,8 +58,6 @@ func runDaemon(ctx context.Context, configFile *string) func(cmd *cobra.Command,
}
}
var g errgroup.Group
cli := client.New(
reg.Address,
cfg.Runner.Insecure,
@@ -69,31 +66,12 @@ func runDaemon(ctx context.Context, configFile *string) func(cmd *cobra.Command,
version,
)
runner := run.NewRunner(cfg, reg, version)
poller := poller.New(
cli,
runner.Run,
cfg,
)
runner := run.NewRunner(cfg, reg, cli, version)
poller := poll.New(cfg, cli, runner)
g.Go(func() error {
l := log.WithField("capacity", cfg.Runner.Capacity).
WithField("endpoint", reg.Address)
l.Infoln("polling the remote server")
poller.Poll(ctx)
if err := poller.Poll(ctx); err != nil {
l.Errorf("poller error: %v", err)
}
poller.Wait()
return nil
})
err = g.Wait()
if err != nil {
log.WithError(err).
Errorln("shutting down the server")
}
return err
return nil
}
}
go.mod
+1
View File
@@ -18,6 +18,7 @@ require (
github.com/stretchr/testify v1.8.1
golang.org/x/sync v0.1.0
golang.org/x/term v0.6.0
golang.org/x/time v0.1.0
google.golang.org/protobuf v1.28.1
gopkg.in/yaml.v3 v3.0.1
gotest.tools/v3 v3.4.0
go.sum
+1
View File
@@ -686,6 +686,7 @@ golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/time v0.1.0 h1:xYY+Bajn2a7VBmTM5GikTmnK8ZuX8YgnQCqZpbBNtmA=
golang.org/x/time v0.1.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
internal/app/poll/poller.go
+80
View File
@@ -0,0 +1,80 @@
package poll
import (
"context"
"errors"
"sync"
"time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
"github.com/bufbuild/connect-go"
log "github.com/sirupsen/logrus"
"golang.org/x/time/rate"
"gitea.com/gitea/act_runner/internal/app/run"
"gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config"
)
type Poller struct {
client client.Client
runner *run.Runner
capacity int
}
func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
return &Poller{
client: client,
runner: runner,
capacity: cfg.Runner.Capacity,
}
}
func (p *Poller) Poll(ctx context.Context) {
limiter := rate.NewLimiter(rate.Every(2*time.Second), 1)
wg := &sync.WaitGroup{}
for i := 0; i < p.capacity; i++ {
wg.Add(1)
go p.poll(ctx, wg, limiter)
}
wg.Wait()
}
func (p *Poller) poll(ctx context.Context, wg *sync.WaitGroup, limiter *rate.Limiter) {
defer wg.Done()
for {
if err := limiter.Wait(ctx); err != nil {
if ctx.Err() != nil {
log.WithError(err).Debug("limiter wait failed")
}
return
}
task, ok := p.fetchTask(ctx)
if !ok {
continue
}
if err := p.runner.Run(ctx, task); err != nil {
log.WithError(err).Error("failed to run task")
}
}
}
func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
reqCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{}))
if errors.Is(err, context.DeadlineExceeded) {
err = nil
}
if err != nil {
log.WithError(err).Error("failed to fetch task")
return nil, false
}
if resp.Msg.Task == nil {
return nil, false
}
return resp.Msg.Task, true
}
internal/app/run/runner.go
+2 -3
View File
@@ -39,8 +39,7 @@ type Runner struct {
runningTasks sync.Map
}
func NewRunner(cfg *config.Config, reg *config.Registration, version string) *Runner {
cli := client.New(reg.Address, cfg.Runner.Insecure, reg.UUID, reg.Token, version)
func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client, version string) *Runner {
ls := labels.Labels{}
for _, v := range reg.Labels {
if l, err := labels.Parse(v); err == nil {
@@ -79,7 +78,7 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error {
defer r.runningTasks.Delete(task.Id)
}
ctx, cancel := context.WithCancel(ctx)
ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
defer cancel()
reporter := report.NewReporter(ctx, cancel, r.client, task)
var runErr error
poller/metric.go
-36
View File
@@ -1,36 +0,0 @@
// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package poller
import "sync/atomic"
// Metric interface
type Metric interface {
IncBusyWorker() int64
DecBusyWorker() int64
BusyWorkers() int64
}
var _ Metric = (*metric)(nil)
type metric struct {
busyWorkers int64
}
// NewMetric for default metric structure
func NewMetric() Metric {
return &metric{}
}
func (m *metric) IncBusyWorker() int64 {
return atomic.AddInt64(&m.busyWorkers, 1)
}
func (m *metric) DecBusyWorker() int64 {
return atomic.AddInt64(&m.busyWorkers, -1)
}
func (m *metric) BusyWorkers() int64 {
return atomic.LoadInt64(&m.busyWorkers)
}
poller/poller.go
-159
View File
@@ -1,159 +0,0 @@
// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package poller
import (
"context"
"errors"
"sync"
"time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
"github.com/bufbuild/connect-go"
log "github.com/sirupsen/logrus"
"gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config"
)
var ErrDataLock = errors.New("Data Lock Error")
func New(cli client.Client, dispatch func(context.Context, *runnerv1.Task) error, cfg *config.Config) *Poller {
return &Poller{
Client: cli,
Dispatch: dispatch,
routineGroup: newRoutineGroup(),
metric: &metric{},
ready: make(chan struct{}, 1),
cfg: cfg,
}
}
type Poller struct {
Client client.Client
Dispatch func(context.Context, *runnerv1.Task) error
sync.Mutex
routineGroup *routineGroup
metric *metric
ready chan struct{}
cfg *config.Config
}
func (p *Poller) schedule() {
p.Lock()
defer p.Unlock()
if int(p.metric.BusyWorkers()) >= p.cfg.Runner.Capacity {
return
}
select {
case p.ready <- struct{}{}:
default:
}
}
func (p *Poller) Wait() {
p.routineGroup.Wait()
}
func (p *Poller) handle(ctx context.Context, l *log.Entry) {
defer func() {
if r := recover(); r != nil {
l.Errorf("handle task panic: %+v", r)
}
}()
for {
select {
case <-ctx.Done():
return
default:
task, err := p.pollTask(ctx)
if task == nil || err != nil {
if err != nil {
l.Errorf("can't find the task: %v", err.Error())
}
time.Sleep(5 * time.Second)
break
}
p.metric.IncBusyWorker()
p.routineGroup.Run(func() {
defer p.schedule()
defer p.metric.DecBusyWorker()
if err := p.dispatchTask(ctx, task); err != nil {
l.Errorf("execute task: %v", err.Error())
}
})
return
}
}
}
func (p *Poller) Poll(ctx context.Context) error {
l := log.WithField("func", "Poll")
for {
// check worker number
p.schedule()
select {
// wait worker ready
case <-p.ready:
case <-ctx.Done():
return nil
}
p.handle(ctx, l)
}
}
func (p *Poller) pollTask(ctx context.Context) (*runnerv1.Task, error) {
l := log.WithField("func", "pollTask")
l.Info("poller: request stage from remote server")
reqCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
// request a new build stage for execution from the central
// build server.
resp, err := p.Client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{}))
if err == context.Canceled || err == context.DeadlineExceeded {
l.WithError(err).Trace("poller: no stage returned")
return nil, nil
}
if err != nil && err == ErrDataLock {
l.WithError(err).Info("task accepted by another runner")
return nil, nil
}
if err != nil {
l.WithError(err).Error("cannot accept task")
return nil, err
}
// exit if a nil or empty stage is returned from the system
// and allow the runner to retry.
if resp.Msg.Task == nil || resp.Msg.Task.Id == 0 {
return nil, nil
}
return resp.Msg.Task, nil
}
func (p *Poller) dispatchTask(ctx context.Context, task *runnerv1.Task) error {
l := log.WithField("func", "dispatchTask")
defer func() {
e := recover()
if e != nil {
l.Errorf("panic error: %v", e)
}
}()
runCtx, cancel := context.WithTimeout(ctx, p.cfg.Runner.Timeout)
defer cancel()
return p.Dispatch(runCtx, task)
}
poller/thread.go
-27
View File
@@ -1,27 +0,0 @@
// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package poller
import "sync"
type routineGroup struct {
waitGroup sync.WaitGroup
}
func newRoutineGroup() *routineGroup {
return new(routineGroup)
}
func (g *routineGroup) Run(fn func()) {
g.waitGroup.Add(1)
go func() {
defer g.waitGroup.Done()
fn()
}()
}
func (g *routineGroup) Wait() {
g.waitGroup.Wait()
}