vikarti.anatra/safing-portbase
1
0
Fork 0
mirror of https://github.com/safing/portbase synced 2026-05-01 21:21:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Split worker types

Browse source
This commit is contained in:
Daniel 2019-09-20 22:04:26 +02:00
parent aaa7aaf648
commit 77ed019f27
2 changed files with 49 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

62
modules/worker.go
View file

@ -8,12 +8,13 @@ import (
"github.com/safing/portbase/log"
)
var (
serviceBackoffDuration = 2 * time.Second
// Worker Default Configuration
const (
DefaultBackoffDuration = 2 * time.Second
)
// StartWorker starts a generic worker that does not fit to be a Task or MicroTask, such as long running (and possibly mostly idle) sessions. You may declare a worker as a service, which will then be automatically restarted in case of an error.
func (m *Module) StartWorker(name string, service bool, fn func(context.Context) error) error {
// RunWorker directly runs a generic worker that does not fit to be a Task or MicroTask, such as long running (and possibly mostly idle) sessions. A call to RunWorker blocks until the worker is finished.
func (m *Module) RunWorker(name string, fn func(context.Context) error) error {
atomic.AddInt32(m.workerCnt, 1)
m.waitGroup.Add(1)
defer func() {
@ -21,28 +22,43 @@ func (m *Module) StartWorker(name string, service bool, fn func(context.Context)
m.waitGroup.Done()
}()
return m.runWorker(name, fn)
}
// StartServiceWorker starts a generic worker, which is automatically restarted in case of an error. A call to StartServiceWorker runs the service-worker in a new goroutine and returns immediately. `backoffDuration` specifies how to long to wait before restarts, multiplied by the number of failed attempts. Pass `0` for the default backoff duration. For custom error remediation functionality, build your own error handling procedure using calls to RunWorker.
func (m *Module) StartServiceWorker(name string, backoffDuration time.Duration, fn func(context.Context) error) {
go m.runServiceWorker(name, backoffDuration, fn)
}
func (m *Module) runServiceWorker(name string, backoffDuration time.Duration, fn func(context.Context) error) {
atomic.AddInt32(m.workerCnt, 1)
m.waitGroup.Add(1)
defer func() {
atomic.AddInt32(m.workerCnt, -1)
m.waitGroup.Done()
}()
if backoffDuration == 0 {
backoffDuration = DefaultBackoffDuration
}
failCnt := 0
if service {
for {
if m.ShutdownInProgress() {
return nil
}
err := m.runWorker(name, fn)
if err != nil {
// log error and restart
failCnt++
sleepFor := time.Duration(failCnt) * serviceBackoffDuration
log.Errorf("module %s service-worker %s failed (%d): %s - restarting in %s", m.Name, name, failCnt, err, sleepFor)
time.Sleep(sleepFor)
} else {
// clean finish
return nil
}
for {
if m.ShutdownInProgress() {
return
}
err := m.runWorker(name, fn)
if err != nil {
// log error and restart
failCnt++
sleepFor := time.Duration(failCnt) * backoffDuration
log.Errorf("%s: service-worker %s failed (%d): %s - restarting in %s", m.Name, name, failCnt, err, sleepFor)
time.Sleep(sleepFor)
} else {
// finish
return
}
} else {
return m.runWorker(name, fn)
}
}