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
|
package service
import (
"os/exec"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
"github.com/spiral/errors"
"github.com/spiral/roadrunner/v2/plugins/logger"
"github.com/spiral/roadrunner/v2/utils"
)
// Process structure contains an information about process, restart information, log, errors, etc
type Process struct {
sync.Mutex
// command to execute
command *exec.Cmd
// rawCmd from the plugin
rawCmd string
Pid int
// root plugin error chan
errCh chan error
// logger
log logger.Logger
ExecTimeout time.Duration
RemainAfterExit bool
RestartSec uint64
// process start time
startTime time.Time
stopped uint64
}
// NewServiceProcess constructs service process structure
func NewServiceProcess(restartAfterExit bool, execTimeout time.Duration, restartDelay uint64, command string, l logger.Logger, errCh chan error) *Process {
return &Process{
rawCmd: command,
RestartSec: restartDelay,
ExecTimeout: execTimeout,
RemainAfterExit: restartAfterExit,
errCh: errCh,
log: l,
}
}
// write message to the log (stderr)
func (p *Process) Write(b []byte) (int, error) {
p.log.Info(utils.AsString(b))
return len(b), nil
}
func (p *Process) start() {
p.Lock()
defer p.Unlock()
const op = errors.Op("processor_start")
// crate fat-process here
p.createProcess()
// non blocking process start
err := p.command.Start()
if err != nil {
p.errCh <- errors.E(op, err)
return
}
// start process waiting routine
go p.wait()
// execHandler checks for the execTimeout
go p.execHandler()
// save start time
p.startTime = time.Now()
p.Pid = p.command.Process.Pid
}
// create command for the process
func (p *Process) createProcess() {
// cmdArgs contain command arguments if the command in form of: php <command> or ls <command> -i -b
var cmdArgs []string
cmdArgs = append(cmdArgs, strings.Split(p.rawCmd, " ")...)
if len(cmdArgs) < 2 {
p.command = exec.Command(p.rawCmd) //nolint:gosec
} else {
p.command = exec.Command(cmdArgs[0], cmdArgs[1:]...) //nolint:gosec
}
// redirect stderr into the Write function of the process.go
p.command.Stderr = p
}
// wait process for exit
func (p *Process) wait() {
// Wait error doesn't matter here
err := p.command.Wait()
if err != nil {
p.log.Error("process wait error", "error", err)
}
// wait for restart delay
if p.RemainAfterExit {
// wait for the delay
time.Sleep(time.Second * time.Duration(p.RestartSec))
// and start command again
p.start()
}
}
// stop can be only sent by the Endure when plugin stopped
func (p *Process) stop() {
atomic.StoreUint64(&p.stopped, 1)
}
func (p *Process) execHandler() {
tt := time.NewTicker(time.Second)
for range tt.C {
// lock here, because p.startTime could be changed during the check
p.Lock()
// if the exec timeout is set
if p.ExecTimeout != 0 {
// if stopped -> kill the process (SIGINT-> SIGKILL) and exit
if atomic.CompareAndSwapUint64(&p.stopped, 1, 1) {
err := p.command.Process.Signal(syscall.SIGINT)
if err != nil {
_ = p.command.Process.Signal(syscall.SIGKILL)
}
tt.Stop()
p.Unlock()
return
}
// check the running time for the script
if time.Now().After(p.startTime.Add(p.ExecTimeout)) {
err := p.command.Process.Signal(syscall.SIGINT)
if err != nil {
_ = p.command.Process.Signal(syscall.SIGKILL)
}
p.Unlock()
tt.Stop()
return
}
}
p.Unlock()
}
}
|
