-
-
Notifications
You must be signed in to change notification settings - Fork 128
/
application.go
257 lines (212 loc) · 6.16 KB
/
application.go
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
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
package gen
// http://erlang.org/doc/apps/kernel/application.html
import (
"fmt"
"sync"
"time"
"github.com/ergo-services/ergo/etf"
)
type ApplicationStartType = string
const (
// start types:
// ApplicationStartPermanent If a permanent application terminates,
// all other applications and the runtime system (node) are also terminated.
ApplicationStartPermanent = "permanent"
// ApplicationStartTemporary If a temporary application terminates,
// this is reported but no other applications are terminated.
ApplicationStartTemporary = "temporary"
// ApplicationStartTransient If a transient application terminates
// with reason normal, this is reported but no other applications are
// terminated. If a transient application terminates abnormally, that
// is with any other reason than normal, all other applications and
// the runtime system (node) are also terminated.
ApplicationStartTransient = "transient"
)
// ApplicationBehavior interface
type ApplicationBehavior interface {
ProcessBehavior
Load(args ...etf.Term) (ApplicationSpec, error)
Start(process Process, args ...etf.Term)
}
type ApplicationSpec struct {
sync.Mutex
Name string
Description string
Version string
Lifespan time.Duration
Applications []string
Environment map[string]interface{}
Children []ApplicationChildSpec
Process Process
StartType ApplicationStartType
}
type ApplicationChildSpec struct {
Child ProcessBehavior
Name string
Args []etf.Term
process Process
}
// Application is implementation of ProcessBehavior interface
type Application struct{}
type ApplicationInfo struct {
Name string
Description string
Version string
PID etf.Pid
}
func (a *Application) ProcessInit(p Process, args ...etf.Term) (ProcessState, error) {
spec, ok := p.Env("spec").(*ApplicationSpec)
if !ok {
return ProcessState{}, fmt.Errorf("ProcessInit: not an ApplicationBehavior")
}
// remove variable from the env
p.SetEnv("spec", nil)
p.SetTrapExit(true)
if spec.Environment != nil {
for k, v := range spec.Environment {
p.SetEnv(k, v)
}
}
if !a.startChildren(p, spec.Children[:]) {
a.stopChildren(p.Self(), spec.Children[:], "failed")
return ProcessState{}, fmt.Errorf("failed")
}
behavior, ok := p.Behavior().(ApplicationBehavior)
if !ok {
return ProcessState{}, fmt.Errorf("ProcessInit: not an ApplicationBehavior")
}
behavior.Start(p, args...)
spec.Process = p
return ProcessState{
Process: p,
State: spec,
}, nil
}
func (a *Application) ProcessLoop(ps ProcessState, started chan<- bool) string {
spec := ps.State.(*ApplicationSpec)
defer func() { spec.Process = nil }()
if spec.Lifespan == 0 {
spec.Lifespan = time.Hour * 24 * 365 * 100 // let's define default lifespan 100 years :)
}
chs := ps.ProcessChannels()
timer := time.NewTimer(spec.Lifespan)
// timer must be stopped explicitly to prevent of timer leaks
// due to its not GCed until the timer fires
defer timer.Stop()
started <- true
for {
select {
case ex := <-chs.GracefulExit:
terminated := ex.From
reason := ex.Reason
if ex.From == ps.Self() {
childrenStopped := a.stopChildren(terminated, spec.Children, reason)
if !childrenStopped {
fmt.Printf("Warining: application can't be stopped. Some of the children are still running")
continue
}
return ex.Reason
}
unknownChild := true
for i := range spec.Children {
child := spec.Children[i].process
if child == nil {
continue
}
if child.Self() == terminated {
unknownChild = false
break
}
}
if unknownChild {
continue
}
switch spec.StartType {
case ApplicationStartPermanent:
a.stopChildren(terminated, spec.Children, string(reason))
fmt.Printf("Application child %s (at %s) stopped with reason %s (permanent: node is shutting down)\n",
terminated, ps.NodeName(), reason)
ps.NodeStop()
return "shutdown"
case ApplicationStartTransient:
if reason == "normal" || reason == "shutdown" {
fmt.Printf("Application child %s (at %s) stopped with reason %s (transient)\n",
terminated, ps.NodeName(), reason)
continue
}
a.stopChildren(terminated, spec.Children, reason)
fmt.Printf("Application child %s (at %s) stopped with reason %s. (transient: node is shutting down)\n",
terminated, ps.NodeName(), reason)
ps.NodeStop()
return string(reason)
case ApplicationStartTemporary:
fmt.Printf("Application child %s (at %s) stopped with reason %s (temporary)\n",
terminated, ps.NodeName(), reason)
}
case direct := <-chs.Direct:
switch direct.Message.(type) {
case MessageDirectChildren:
pids := []etf.Pid{}
for i := range spec.Children {
if spec.Children[i].process == nil {
continue
}
pids = append(pids, spec.Children[i].process.Self())
}
direct.Message = pids
direct.Err = nil
direct.Reply <- direct
default:
direct.Message = nil
direct.Err = ErrUnsupportedRequest
direct.Reply <- direct
}
case <-ps.Context().Done():
// node is down or killed using p.Kill()
return "kill"
case <-timer.C:
// time to die
ps.SetTrapExit(false)
go ps.Exit("normal")
case <-chs.Mailbox:
// do nothing
}
}
}
func (a *Application) stopChildren(from etf.Pid, children []ApplicationChildSpec, reason string) bool {
childrenStopped := true
for i := range children {
child := children[i].process
if child == nil {
continue
}
if child.Self() == from {
continue
}
if !child.IsAlive() {
continue
}
if err := child.Exit(reason); err != nil {
childrenStopped = false
continue
}
if err := child.WaitWithTimeout(5 * time.Second); err != nil {
childrenStopped = false
continue
}
children[i].process = nil
}
return childrenStopped
}
func (a *Application) startChildren(parent Process, children []ApplicationChildSpec) bool {
for i := range children {
// i know, it looks weird to use the funcion from supervisor file.
// will move it to somewhere else, but let it be there for a while.
p := startChild(parent, children[i].Name, children[i].Child, children[i].Args...)
if p == nil {
return false
}
children[i].process = p
}
return true
}