1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
package watcher
import (
"fmt"
"github.com/spiral/roadrunner"
"github.com/spiral/roadrunner/util"
"time"
)
const (
// EventMaxTTL thrown when worker is removed due MaxTTL being reached. Context is roadrunner.WorkerError
EventMaxTTL = iota + 8000
// EventMaxMemory caused when worker consumes more memory than allowed.
EventMaxMemory
)
// handles watcher events
type listener func(event int, ctx interface{})
// defines the watcher behaviour
type watcherConfig struct {
// MaxTTL defines maximum time worker is allowed to live.
MaxTTL time.Duration
// MaxMemory defines maximum amount of memory allowed for worker. In megabytes.
MaxMemory uint64
}
// Normalize watcher config and upscale the durations.
func (c *watcherConfig) Normalize() error {
// Always use second based definition for time durations
if c.MaxTTL < time.Microsecond {
c.MaxTTL = time.Second * time.Duration(c.MaxTTL.Nanoseconds())
}
return nil
}
type watcher struct {
lsn listener
interval time.Duration
cfg *watcherConfig
stop chan interface{}
// list of workers which are currently working
//working map[*roadrunner.Worker]time.Time
}
// watch the pool state
func (watch *watcher) watch(p roadrunner.Pool) {
now := time.Now()
for _, w := range p.Workers() {
if watch.cfg.MaxTTL != 0 && now.Sub(w.Created) >= watch.cfg.MaxTTL {
err := fmt.Errorf("max TTL reached (%s)", watch.cfg.MaxTTL)
if p.Remove(w, err) {
watch.report(EventMaxTTL, w, err)
}
}
state, err := util.WorkerState(w)
if err != nil {
continue
}
if watch.cfg.MaxMemory != 0 && state.MemoryUsage >= watch.cfg.MaxMemory*1024*1024 {
err := fmt.Errorf("max allowed memory reached (%vMB)", watch.cfg.MaxMemory)
if p.Remove(w, err) {
watch.report(EventMaxMemory, w, err)
}
}
}
}
// throw watcher event
func (watch *watcher) report(event int, worker *roadrunner.Worker, caused error) {
if watch.lsn != nil {
watch.lsn(event, roadrunner.WorkerError{Worker: worker, Caused: caused})
}
}
// Attach watcher to the pool
func (watch *watcher) Attach(pool roadrunner.Pool) roadrunner.Watcher {
wp := &watcher{
interval: watch.interval,
lsn: watch.lsn,
cfg: watch.cfg,
stop: make(chan interface{}),
}
go func(wp *watcher, pool roadrunner.Pool) {
ticker := time.NewTicker(wp.interval)
for {
select {
case <-ticker.C:
wp.watch(pool)
case <-wp.stop:
return
}
}
}(wp, pool)
return wp
}
// Detach watcher from the pool.
func (watch *watcher) Detach() {
close(watch.stop)
}
|
