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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
package roadrunner
import (
"sync"
"os/exec"
"fmt"
)
const (
// EventPoolConstruct triggered when server creates new pool.
EventServerStart = iota + 200
// EventPoolConstruct triggered when server creates new pool.
EventServerStop
// EventServerFailure triggered when server is unable to replace dead pool.
EventServerFailure
// EventPoolConstruct triggered when server creates new pool.
EventPoolConstruct
// EventPoolDestruct triggered when server destroys existed pool.
EventPoolDestruct
)
// Service manages pool creation and swapping.
type Server struct {
// configures server, pool, cmd creation and factory.
cfg *ServerConfig
// worker command creator
cmd func() *exec.Cmd
// observes pool events (can be attached to multiple pools at the same time)
observer func(event int, ctx interface{})
// protects pool while the re-configuration
mu sync.Mutex
// indicates that server was started
started bool
// creates and connects to workers
factory Factory
// currently active pool instance
pool Pool
}
// NewServer creates new router. Make sure to call configure before the usage.
func NewServer(cmd func() *exec.Cmd, cfg *ServerConfig) *Server {
return &Server{cmd: cmd, cfg: cfg}
}
// Observe attaches server event watcher.
func (srv *Server) Observe(o func(event int, ctx interface{})) {
srv.observer = o
}
// Reconfigure re-configures underlying pool and destroys it's previous version if any. Reconfigure will ignore factory
// and relay settings.
func (srv *Server) Reconfigure(cfg *ServerConfig) error {
srv.mu.Lock()
if !srv.started {
srv.cfg = cfg
srv.mu.Unlock()
return nil
}
srv.mu.Unlock()
srv.mu.Lock()
previous := srv.pool
srv.mu.Unlock()
pool, err := NewPool(srv.cmd, srv.factory, cfg.Pool)
if err != nil {
return err
}
srv.mu.Lock()
srv.cfg.Pool, srv.pool = cfg.Pool, pool
srv.pool.Observe(srv.poolObserver)
srv.mu.Unlock()
srv.throw(EventPoolConstruct, pool)
if previous != nil {
go func(previous Pool) {
srv.throw(EventPoolDestruct, previous)
previous.Destroy()
}(previous)
}
return nil
}
// Reset resets the state of underlying pool and rebuilds all of it's workers.
func (srv *Server) Reset() error {
return srv.Reconfigure(srv.cfg)
}
// Start underlying worker pool, configure factory and command provider.
func (srv *Server) Start() (err error) {
srv.mu.Lock()
defer srv.mu.Unlock()
if srv.factory, err = srv.cfg.makeFactory(); err != nil {
return err
}
if srv.pool, err = NewPool(srv.cmd, srv.factory, srv.cfg.Pool); err != nil {
return err
}
srv.pool.Observe(srv.poolObserver)
srv.started = true
srv.throw(EventServerStart, srv)
return nil
}
// Stop underlying worker pool and close the factory.
func (srv *Server) Stop() error {
srv.mu.Lock()
defer srv.mu.Unlock()
if !srv.started {
return nil
}
srv.throw(EventPoolDestruct, srv.pool)
srv.pool.Destroy()
srv.factory.Close()
srv.cmd = nil
srv.factory = nil
srv.pool = nil
srv.started = false
srv.throw(EventServerStop, srv)
return nil
}
// Exec one task with given payload and context, returns result or error.
func (srv *Server) Exec(rqs *Payload) (rsp *Payload, err error) {
pool := srv.Pool()
if pool == nil {
return nil, fmt.Errorf("no associared pool")
}
return pool.Exec(rqs)
}
// Workers returns worker list associated with the server pool.
func (srv *Server) Workers() (workers []*Worker) {
p := srv.Pool()
if p == nil {
return nil
}
return p.Workers()
}
// Pool returns active pool or error.
func (srv *Server) Pool() Pool {
srv.mu.Lock()
defer srv.mu.Unlock()
return srv.pool
}
// Observe pool events.
func (srv *Server) poolObserver(event int, ctx interface{}) {
// bypassing to user specified observer
srv.throw(event, ctx)
if event == EventPoolError {
// pool failure, rebuilding
if err := srv.Reset(); err != nil {
srv.mu.Lock()
defer srv.mu.Unlock()
srv.started = false
srv.pool = nil
srv.factory = nil
// everything is dead, this is recoverable but heavy state
srv.throw(EventServerFailure, srv)
}
}
}
// throw invokes event handler if any.
func (srv *Server) throw(event int, ctx interface{}) {
if srv.observer != nil {
srv.observer(event, ctx)
}
}
|
