2018-12-05 00:48:33 +03:00
|
|
|
package rutina
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
2020-01-12 16:40:01 +03:00
|
|
|
"errors"
|
2019-01-11 17:22:06 +03:00
|
|
|
"os"
|
|
|
|
"os/signal"
|
2018-12-05 00:48:33 +03:00
|
|
|
"sync"
|
2019-01-17 07:54:39 +03:00
|
|
|
"sync/atomic"
|
2019-04-03 23:55:11 +03:00
|
|
|
"syscall"
|
2020-01-12 16:40:01 +03:00
|
|
|
"time"
|
2018-12-05 00:48:33 +03:00
|
|
|
)
|
|
|
|
|
2020-01-12 16:40:01 +03:00
|
|
|
var (
|
|
|
|
ErrRunLimit = errors.New("rutina run limit")
|
|
|
|
ErrTimeoutOrKilled = errors.New("rutina timeouted or killed")
|
|
|
|
ErrProcessNotFound = errors.New("process not found")
|
|
|
|
ErrShutdown = errors.New("shutdown")
|
|
|
|
)
|
|
|
|
|
|
|
|
type logger func(format string, v ...interface{})
|
|
|
|
|
|
|
|
var nopLogger = func(format string, v ...interface{}) {}
|
|
|
|
|
2021-03-18 17:58:00 +03:00
|
|
|
// Rutina is routine manager
|
2018-12-05 00:48:33 +03:00
|
|
|
type Rutina struct {
|
2020-01-12 16:40:01 +03:00
|
|
|
ctx context.Context // State of application (started/stopped)
|
|
|
|
Cancel func() // Cancel func that stops all routines
|
|
|
|
wg sync.WaitGroup // WaitGroup that wait all routines to complete
|
|
|
|
onceErr sync.Once // Flag that prevents overwrite first error that shutdowns all routines
|
|
|
|
onceWait sync.Once // Flag that prevents wait already waited rutina
|
|
|
|
err error // First error that shutdowns all routines
|
|
|
|
logger logger // Optional logger
|
|
|
|
counter *uint64 // Optional counter that names routines with increment ids for debug purposes at logger
|
|
|
|
errCh chan error // Optional channel for errors when RestartIfError and DoNothingIfError
|
|
|
|
autoListenSignals []os.Signal // Optional listening os signals, default disabled
|
|
|
|
processes map[uint64]*process
|
|
|
|
mu sync.Mutex
|
2018-12-05 00:48:33 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// New instance with builtin context
|
2021-03-18 17:58:00 +03:00
|
|
|
func New(opts ...Options) *Rutina {
|
2020-01-12 16:40:01 +03:00
|
|
|
if opts == nil {
|
2021-03-18 17:58:00 +03:00
|
|
|
opts = []Options{}
|
2020-01-12 16:40:01 +03:00
|
|
|
}
|
2021-03-18 17:58:00 +03:00
|
|
|
options := composeOptions(opts)
|
|
|
|
ctx, cancel := context.WithCancel(options.ParentContext)
|
2019-04-04 12:02:49 +03:00
|
|
|
var counter uint64
|
2020-01-12 16:40:01 +03:00
|
|
|
return &Rutina{
|
|
|
|
ctx: ctx,
|
|
|
|
Cancel: cancel,
|
|
|
|
wg: sync.WaitGroup{},
|
|
|
|
onceErr: sync.Once{},
|
|
|
|
onceWait: sync.Once{},
|
|
|
|
err: nil,
|
2021-03-18 17:58:00 +03:00
|
|
|
logger: options.Logger,
|
2020-01-12 16:40:01 +03:00
|
|
|
counter: &counter,
|
2021-03-18 17:58:00 +03:00
|
|
|
errCh: options.Errors,
|
|
|
|
autoListenSignals: options.ListenOsSignals,
|
2020-01-12 16:40:01 +03:00
|
|
|
processes: map[uint64]*process{},
|
|
|
|
mu: sync.Mutex{},
|
2019-06-29 01:19:21 +03:00
|
|
|
}
|
2018-12-05 00:48:33 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// Go routine
|
2021-03-18 17:58:00 +03:00
|
|
|
func (r *Rutina) Go(doer func(ctx context.Context) error, opts ...RunOptions) uint64 {
|
|
|
|
options := composeRunOptions(opts)
|
2019-04-04 22:56:53 +03:00
|
|
|
// Check that context is not canceled yet
|
|
|
|
if r.ctx.Err() != nil {
|
2020-01-12 16:40:01 +03:00
|
|
|
return 0
|
2019-03-27 02:44:38 +03:00
|
|
|
}
|
2020-01-12 16:40:01 +03:00
|
|
|
|
|
|
|
r.mu.Lock()
|
|
|
|
id := atomic.AddUint64(r.counter, 1)
|
|
|
|
process := process{
|
|
|
|
id: id,
|
|
|
|
doer: doer,
|
2021-03-18 17:58:00 +03:00
|
|
|
onDone: options.OnDone,
|
|
|
|
onError: options.OnError,
|
|
|
|
restartLimit: options.MaxCount,
|
2020-01-12 16:40:01 +03:00
|
|
|
restartCount: 0,
|
2021-03-18 17:58:00 +03:00
|
|
|
timeout: options.Timeout,
|
2019-03-27 02:44:38 +03:00
|
|
|
}
|
2020-01-12 16:40:01 +03:00
|
|
|
r.processes[id] = &process
|
|
|
|
r.mu.Unlock()
|
2019-03-27 02:44:38 +03:00
|
|
|
|
2018-12-05 00:48:33 +03:00
|
|
|
r.wg.Add(1)
|
|
|
|
go func() {
|
2019-03-27 02:44:38 +03:00
|
|
|
defer r.wg.Done()
|
2020-01-12 16:40:01 +03:00
|
|
|
if err := process.run(r.ctx, r.errCh, r.logger); err != nil {
|
|
|
|
if err != ErrShutdown {
|
2019-04-04 22:56:53 +03:00
|
|
|
r.onceErr.Do(func() {
|
2019-03-27 02:44:38 +03:00
|
|
|
r.err = err
|
|
|
|
})
|
|
|
|
}
|
2020-01-12 16:40:01 +03:00
|
|
|
r.Cancel()
|
2018-12-05 00:48:33 +03:00
|
|
|
}
|
2020-01-12 16:40:01 +03:00
|
|
|
r.mu.Lock()
|
|
|
|
defer r.mu.Unlock()
|
|
|
|
delete(r.processes, process.id)
|
|
|
|
r.logger("completed #%d", process.id)
|
2018-12-05 00:48:33 +03:00
|
|
|
}()
|
2020-01-12 16:40:01 +03:00
|
|
|
return id
|
|
|
|
}
|
|
|
|
|
|
|
|
func (r *Rutina) Processes() []uint64 {
|
|
|
|
var procesess []uint64
|
|
|
|
for id, _ := range r.processes {
|
|
|
|
procesess = append(procesess, id)
|
|
|
|
}
|
|
|
|
return procesess
|
2018-12-05 00:48:33 +03:00
|
|
|
}
|
|
|
|
|
2019-06-29 01:19:21 +03:00
|
|
|
// Errors returns chan for all errors, event if DoNothingIfError or RestartIfError set.
|
2019-04-04 11:54:24 +03:00
|
|
|
// By default it nil. Use MixinErrChan to turn it on
|
|
|
|
func (r *Rutina) Errors() <-chan error {
|
|
|
|
return r.errCh
|
|
|
|
}
|
|
|
|
|
|
|
|
// ListenOsSignals is simple OS signals handler. By default listen syscall.SIGINT and syscall.SIGTERM
|
2019-04-01 11:38:03 +03:00
|
|
|
func (r *Rutina) ListenOsSignals(signals ...os.Signal) {
|
|
|
|
if len(signals) == 0 {
|
2019-04-03 23:55:11 +03:00
|
|
|
signals = []os.Signal{syscall.SIGINT, syscall.SIGTERM}
|
2019-04-01 11:38:03 +03:00
|
|
|
}
|
2019-04-04 10:15:18 +03:00
|
|
|
go func() {
|
2019-01-11 17:22:06 +03:00
|
|
|
sig := make(chan os.Signal, 1)
|
2019-04-01 11:38:03 +03:00
|
|
|
signal.Notify(sig, signals...)
|
2020-01-12 16:40:01 +03:00
|
|
|
r.logger("starting OS signals listener")
|
2019-01-11 17:22:06 +03:00
|
|
|
select {
|
2019-01-17 07:54:39 +03:00
|
|
|
case s := <-sig:
|
2020-01-12 16:40:01 +03:00
|
|
|
r.logger("stopping by OS signal (%v)", s)
|
2019-04-04 10:15:18 +03:00
|
|
|
r.Cancel()
|
|
|
|
case <-r.ctx.Done():
|
2019-01-11 17:22:06 +03:00
|
|
|
}
|
2019-04-04 10:15:18 +03:00
|
|
|
}()
|
2019-01-11 17:22:06 +03:00
|
|
|
}
|
|
|
|
|
2018-12-05 00:48:33 +03:00
|
|
|
// Wait all routines and returns first error or nil if all routines completes without errors
|
|
|
|
func (r *Rutina) Wait() error {
|
2020-01-12 16:40:01 +03:00
|
|
|
if len(r.autoListenSignals) > 0 {
|
|
|
|
r.ListenOsSignals(r.autoListenSignals...)
|
|
|
|
}
|
2019-04-04 22:56:53 +03:00
|
|
|
r.onceWait.Do(func() {
|
|
|
|
r.wg.Wait()
|
|
|
|
if r.errCh != nil {
|
|
|
|
close(r.errCh)
|
|
|
|
}
|
|
|
|
})
|
2018-12-05 00:48:33 +03:00
|
|
|
return r.err
|
|
|
|
}
|
2019-04-03 23:55:11 +03:00
|
|
|
|
2020-01-12 16:40:01 +03:00
|
|
|
// Kill process by id
|
|
|
|
func (r *Rutina) Kill(id uint64) error {
|
|
|
|
p, ok := r.processes[id]
|
|
|
|
if !ok {
|
|
|
|
return ErrProcessNotFound
|
2019-04-04 22:56:53 +03:00
|
|
|
}
|
2020-01-12 16:40:01 +03:00
|
|
|
if p.cancel != nil {
|
|
|
|
p.cancel()
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
type process struct {
|
|
|
|
id uint64
|
|
|
|
doer func(ctx context.Context) error
|
|
|
|
cancel func()
|
|
|
|
onDone Policy
|
|
|
|
onError Policy
|
|
|
|
restartLimit *int
|
|
|
|
restartCount int
|
|
|
|
timeout *time.Duration
|
2019-04-04 22:56:53 +03:00
|
|
|
}
|
|
|
|
|
2020-01-12 16:40:01 +03:00
|
|
|
func (p *process) run(pctx context.Context, errCh chan error, logger logger) error {
|
|
|
|
var ctx context.Context
|
|
|
|
if p.timeout != nil {
|
|
|
|
ctx, p.cancel = context.WithTimeout(pctx, *p.timeout)
|
|
|
|
defer p.cancel()
|
|
|
|
} else {
|
|
|
|
ctx, p.cancel = context.WithCancel(pctx)
|
|
|
|
}
|
|
|
|
for {
|
|
|
|
logger("starting process #%d", p.id)
|
|
|
|
p.restartCount++
|
|
|
|
currentAction := p.onDone
|
|
|
|
err := p.doer(ctx)
|
|
|
|
if err != nil {
|
|
|
|
if p.onError == Shutdown {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
currentAction = p.onError
|
|
|
|
logger("error on process #%d: %s", p.id, err)
|
|
|
|
if errCh != nil {
|
|
|
|
errCh <- err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
switch currentAction {
|
|
|
|
case DoNothing:
|
|
|
|
return nil
|
|
|
|
case Shutdown:
|
|
|
|
return ErrShutdown
|
|
|
|
case Restart:
|
|
|
|
if ctx.Err() != nil {
|
|
|
|
if p.onError == Shutdown {
|
|
|
|
return ErrTimeoutOrKilled
|
|
|
|
} else {
|
|
|
|
if errCh != nil {
|
|
|
|
errCh <- ErrTimeoutOrKilled
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if p.restartLimit == nil || p.restartCount > *p.restartLimit {
|
|
|
|
logger("run count limit process #%d", p.id)
|
|
|
|
if p.onError == Shutdown {
|
|
|
|
return ErrRunLimit
|
|
|
|
} else {
|
|
|
|
if errCh != nil {
|
|
|
|
errCh <- ErrRunLimit
|
|
|
|
}
|
2022-08-06 13:30:40 +03:00
|
|
|
return ErrRunLimit
|
2020-01-12 16:40:01 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
logger("restarting process #%d", p.id)
|
|
|
|
}
|
2019-04-03 23:55:11 +03:00
|
|
|
}
|
|
|
|
}
|