diff options
author | Wolfy-J <[email protected]> | 2019-05-04 19:24:25 +0300 |
---|---|---|
committer | Wolfy-J <[email protected]> | 2019-05-04 19:24:25 +0300 |
commit | 2efc533f2aac215d487a80020b0f9bf4ae5209c3 (patch) | |
tree | a80a7a74dc4ca8c290b8b1bf1f6d24535b5ae3d7 /service | |
parent | 726b31008e73ab83d0582305c28a8cf62322e47a (diff) |
watchers renamed to controllers
Diffstat (limited to 'service')
-rw-r--r-- | service/container.go | 2 | ||||
-rw-r--r-- | service/http/service.go | 4 | ||||
-rw-r--r-- | service/watcher/config.go | 14 | ||||
-rw-r--r-- | service/watcher/controller.go (renamed from service/watcher/watcher.go) | 90 | ||||
-rw-r--r-- | service/watcher/service.go | 18 | ||||
-rw-r--r-- | service/watcher/state_filter.go (renamed from service/watcher/state_watch.go) | 12 |
6 files changed, 70 insertions, 70 deletions
diff --git a/service/container.go b/service/container.go index abeaf369..ea1819d8 100644 --- a/service/container.go +++ b/service/container.go @@ -16,7 +16,7 @@ var errNoConfig = fmt.Errorf("no config has been provided") // implement service.HydrateConfig. const InitMethod = "Init" -// Services can serve. Services can provide Init method which must return (bool, error) signature and might accept +// Service can serve. Services can provide Init method which must return (bool, error) signature and might accept // other services and/or configs as dependency. type Service interface { // Serve serves. diff --git a/service/http/service.go b/service/http/service.go index 800d3ca9..b76d8893 100644 --- a/service/http/service.go +++ b/service/http/service.go @@ -33,14 +33,14 @@ type Service struct { mdwr []middleware mu sync.Mutex rr *roadrunner.Server - watcher roadrunner.Watcher + watcher roadrunner.Controller handler *Handler http *http.Server https *http.Server } // Watch attaches watcher. -func (s *Service) Watch(w roadrunner.Watcher) { +func (s *Service) Watch(w roadrunner.Controller) { s.watcher = w } diff --git a/service/watcher/config.go b/service/watcher/config.go index 74be517a..8151005d 100644 --- a/service/watcher/config.go +++ b/service/watcher/config.go @@ -8,11 +8,11 @@ import ( // Configures set of Services. type Config struct { - // Interval defines the update duration for underlying watchers, default 1s. + // Interval defines the update duration for underlying controllers, default 1s. Interval time.Duration // Services declares list of services to be watched. - Services map[string]*watcherConfig + Services map[string]*controllerConfig } // Hydrate must populate Config values using given Config source. Must return error if Config is not valid. @@ -36,13 +36,13 @@ func (c *Config) InitDefaults() error { return nil } -// Watchers returns list of defined Services -func (c *Config) Watchers(l listener) (watchers map[string]roadrunner.Watcher) { - watchers = make(map[string]roadrunner.Watcher) +// Controllers returns list of defined Services +func (c *Config) Controllers(l listener) (controllers map[string]roadrunner.Controller) { + controllers = make(map[string]roadrunner.Controller) for name, cfg := range c.Services { - watchers[name] = &watcher{lsn: l, tick: c.Interval, cfg: cfg} + controllers[name] = &controller{lsn: l, tick: c.Interval, cfg: cfg} } - return watchers + return controllers } diff --git a/service/watcher/watcher.go b/service/watcher/controller.go index 65a2eeeb..38eddf84 100644 --- a/service/watcher/watcher.go +++ b/service/watcher/controller.go @@ -21,11 +21,11 @@ const ( EventMaxExecTTL ) -// handles watcher events +// handles controller events type listener func(event int, ctx interface{}) -// defines the watcher behaviour -type watcherConfig struct { +// defines the controller behaviour +type controllerConfig struct { // MaxMemory defines maximum amount of memory allowed for worker. In megabytes. MaxMemory uint64 @@ -39,54 +39,54 @@ type watcherConfig struct { MaxExecTTL int64 } -type watcher struct { +type controller struct { lsn listener tick time.Duration - cfg *watcherConfig + cfg *controllerConfig // list of workers which are currently working - sw *stateWatcher + sw *stateFilter stop chan interface{} } -// watch the pool state -func (wch *watcher) watch(p roadrunner.Pool) { - wch.loadWorkers(p) +// control the pool state +func (c *controller) control(p roadrunner.Pool) { + c.loadWorkers(p) now := time.Now() - if wch.cfg.MaxExecTTL != 0 { - for _, w := range wch.sw.find( + if c.cfg.MaxExecTTL != 0 { + for _, w := range c.sw.find( roadrunner.StateWorking, - now.Add(-time.Second*time.Duration(wch.cfg.MaxExecTTL)), + now.Add(-time.Second*time.Duration(c.cfg.MaxExecTTL)), ) { eID := w.State().NumExecs() - err := fmt.Errorf("max exec time reached (%vs)", wch.cfg.MaxExecTTL) + err := fmt.Errorf("max exec time reached (%vs)", c.cfg.MaxExecTTL) // make sure worker still on initial request if p.Remove(w, err) && w.State().NumExecs() == eID { go w.Kill() - wch.report(EventMaxExecTTL, w, err) + c.report(EventMaxExecTTL, w, err) } } } // locale workers which are in idle mode for too long - if wch.cfg.MaxIdleTTL != 0 { - for _, w := range wch.sw.find( + if c.cfg.MaxIdleTTL != 0 { + for _, w := range c.sw.find( roadrunner.StateReady, - now.Add(-time.Second*time.Duration(wch.cfg.MaxIdleTTL)), + now.Add(-time.Second*time.Duration(c.cfg.MaxIdleTTL)), ) { - err := fmt.Errorf("max idle time reached (%vs)", wch.cfg.MaxIdleTTL) + err := fmt.Errorf("max idle time reached (%vs)", c.cfg.MaxIdleTTL) if p.Remove(w, err) { - wch.report(EventMaxIdleTTL, w, err) + c.report(EventMaxIdleTTL, w, err) } } } } -func (wch *watcher) loadWorkers(p roadrunner.Pool) { +func (c *controller) loadWorkers(p roadrunner.Pool) { now := time.Now() for _, w := range p.Workers() { @@ -100,52 +100,52 @@ func (wch *watcher) loadWorkers(p roadrunner.Pool) { continue } - if wch.cfg.TTL != 0 && now.Sub(w.Created).Seconds() >= float64(wch.cfg.TTL) { - err := fmt.Errorf("max TTL reached (%vs)", wch.cfg.TTL) + if c.cfg.TTL != 0 && now.Sub(w.Created).Seconds() >= float64(c.cfg.TTL) { + err := fmt.Errorf("max TTL reached (%vs)", c.cfg.TTL) if p.Remove(w, err) { - wch.report(EventMaxTTL, w, err) + c.report(EventMaxTTL, w, err) } continue } - if wch.cfg.MaxMemory != 0 && s.MemoryUsage >= wch.cfg.MaxMemory*1024*1024 { - err := fmt.Errorf("max allowed memory reached (%vMB)", wch.cfg.MaxMemory) + if c.cfg.MaxMemory != 0 && s.MemoryUsage >= c.cfg.MaxMemory*1024*1024 { + err := fmt.Errorf("max allowed memory reached (%vMB)", c.cfg.MaxMemory) if p.Remove(w, err) { - wch.report(EventMaxMemory, w, err) + c.report(EventMaxMemory, w, err) } continue } - // watch the worker state changes - wch.sw.push(w) + // control the worker state changes + c.sw.push(w) } - wch.sw.sync(now) + c.sw.sync(now) } -// throw watcher event -func (wch *watcher) report(event int, worker *roadrunner.Worker, caused error) { - if wch.lsn != nil { - wch.lsn(event, roadrunner.WorkerError{Worker: worker, Caused: caused}) +// throw controller event +func (c *controller) report(event int, worker *roadrunner.Worker, caused error) { + if c.lsn != nil { + c.lsn(event, roadrunner.WorkerError{Worker: worker, Caused: caused}) } } -// Attach watcher to the pool -func (wch *watcher) Attach(pool roadrunner.Pool) roadrunner.Watcher { - wp := &watcher{ - tick: wch.tick, - lsn: wch.lsn, - cfg: wch.cfg, - sw: newStateWatcher(), +// Attach controller to the pool +func (c *controller) Attach(pool roadrunner.Pool) roadrunner.Controller { + wp := &controller{ + tick: c.tick, + lsn: c.lsn, + cfg: c.cfg, + sw: newStateFilter(), stop: make(chan interface{}), } - go func(wp *watcher, pool roadrunner.Pool) { + go func(wp *controller, pool roadrunner.Pool) { ticker := time.NewTicker(wp.tick) for { select { case <-ticker.C: - wp.watch(pool) + wp.control(pool) case <-wp.stop: return } @@ -155,7 +155,7 @@ func (wch *watcher) Attach(pool roadrunner.Pool) roadrunner.Watcher { return wp } -// Detach watcher from the pool. -func (wch *watcher) Detach() { - close(wch.stop) +// Detach controller from the pool. +func (c *controller) Detach() { + close(c.stop) } diff --git a/service/watcher/service.go b/service/watcher/service.go index 0d419716..3db23b68 100644 --- a/service/watcher/service.go +++ b/service/watcher/service.go @@ -5,25 +5,25 @@ import ( "github.com/spiral/roadrunner/service" ) -// ID defines watcher service name. -const ID = "watch" +// ID defines controller service name. +const ID = "control" -// Watchable defines the ability to attach rr watcher. +// Watchable defines the ability to attach rr controller. type Watchable interface { - // Watch attaches watcher to the service. - Watch(w roadrunner.Watcher) + // Watch attaches controller to the service. + Watch(w roadrunner.Controller) } -// Services to watch the state of rr service inside other services. +// Services to control the state of rr service inside other services. type Service struct { cfg *Config lsns []func(event int, ctx interface{}) } -// Init watcher service +// Init controller service func (s *Service) Init(cfg *Config, c service.Container) (bool, error) { // mount Services to designated services - for id, watcher := range cfg.Watchers(s.throw) { + for id, watcher := range cfg.Controllers(s.throw) { svc, _ := c.Get(id) if watchable, ok := svc.(Watchable); ok { watchable.Watch(watcher) @@ -33,7 +33,7 @@ func (s *Service) Init(cfg *Config, c service.Container) (bool, error) { return true, nil } -// AddListener attaches server event watcher. +// AddListener attaches server event controller. func (s *Service) AddListener(l func(event int, ctx interface{})) { s.lsns = append(s.lsns, l) } diff --git a/service/watcher/state_watch.go b/service/watcher/state_filter.go index 3090d15d..d85f1308 100644 --- a/service/watcher/state_watch.go +++ b/service/watcher/state_filter.go @@ -5,7 +5,7 @@ import ( "time" ) -type stateWatcher struct { +type stateFilter struct { prev map[*roadrunner.Worker]state next map[*roadrunner.Worker]state } @@ -16,20 +16,20 @@ type state struct { since time.Time } -func newStateWatcher() *stateWatcher { - return &stateWatcher{ +func newStateFilter() *stateFilter { + return &stateFilter{ prev: make(map[*roadrunner.Worker]state), next: make(map[*roadrunner.Worker]state), } } // add new worker to be watched -func (sw *stateWatcher) push(w *roadrunner.Worker) { +func (sw *stateFilter) push(w *roadrunner.Worker) { sw.next[w] = state{state: w.State().Value(), numExecs: w.State().NumExecs()} } // update worker states. -func (sw *stateWatcher) sync(t time.Time) { +func (sw *stateFilter) sync(t time.Time) { for w := range sw.prev { if _, ok := sw.next[w]; !ok { delete(sw.prev, w) @@ -47,7 +47,7 @@ func (sw *stateWatcher) sync(t time.Time) { } // find all workers which spend given amount of time in a specific state. -func (sw *stateWatcher) find(state int64, since time.Time) (workers []*roadrunner.Worker) { +func (sw *stateFilter) find(state int64, since time.Time) (workers []*roadrunner.Worker) { for w, s := range sw.prev { if s.state == state && s.since.Before(since) { workers = append(workers, w) |