1// SPDX-License-Identifier: MIT
2
3package executor
4
5import (
6 "bytes"
7 "errors"
8 "fmt"
9 "io"
10 "os"
11 "os/exec"
12 "sort"
13 "strconv"
14 "strings"
15 "sync"
16 "syscall"
17 "time"
18 "unicode/utf8"
19)
20
21// Cancellation timings of the official runner (Runner.Sdk/ProcessInvoker.cs:32):
22// SIGINT at once, SIGTERM after 7.5 s, kill 2.5 s later.
23const (
24 DefaultTermAfter = 7500 * time.Millisecond
25 DefaultKillAfter = 2500 * time.Millisecond
26 // DefaultWaitDelay is how long the executor waits for the output pipes to
27 // close after the shell exited, before it gives up on background children
28 // that still hold them: ProcessInvoker.ProcessExitedHandler waits 5 s
29 // (ProcessInvoker.cs:491).
30 DefaultWaitDelay = 5 * time.Second
31)
32
33// lineWriter turns a byte stream into lines like StreamReader.ReadLine: "\n",
34// "\r" and "\r\n" end a line, and invalid UTF-8 becomes U+FFFD. It calls fn
35// for every complete line, in order, from the goroutine that writes.
36type lineWriter struct {
37 fn func(line string)
38 buf []byte
39 pending bool // the last byte seen was "\r": a following "\n" is part of it
40}
41
42// maxLineBytes bounds what a line may hold before it is emitted unfinished: a
43// step that prints gigabytes without a line feed must not fill the memory of
44// the executor. The official runner reads lines with StreamReader.ReadLine and
45// has no such bound; a line cut here is sent as consecutive lines.
46const maxLineBytes = 1 << 20
47
48func (w *lineWriter) Write(p []byte) (int, error) {
49 n := len(p)
50 for len(p) > 0 {
51 if w.pending {
52 w.pending = false
53 if p[0] == '\n' {
54 p = p[1:]
55 continue
56 }
57 }
58 i := bytes.IndexAny(p, "\r\n")
59 if i < 0 {
60 for len(w.buf)+len(p) > maxLineBytes {
61 room := maxLineBytes - len(w.buf)
62 w.buf = append(w.buf, p[:room]...)
63 p = p[room:]
64 w.emitCut()
65 }
66 w.buf = append(w.buf, p...)
67 break
68 }
69 w.buf = append(w.buf, p[:i]...)
70 w.emit()
71 w.pending = p[i] == '\r'
72 p = p[i+1:]
73 }
74 return n, nil
75}
76
77// emitCut emits the buffer without a line end, cutting before an incomplete
78// UTF-8 character so that it is not replaced by U+FFFD.
79func (w *lineWriter) emitCut() {
80 cut := len(w.buf)
81 for back := 1; back <= utf8.UTFMax && back <= len(w.buf); back++ {
82 if utf8.RuneStart(w.buf[len(w.buf)-back]) {
83 if !utf8.FullRune(w.buf[len(w.buf)-back:]) {
84 cut = len(w.buf) - back
85 }
86 break
87 }
88 }
89 rest := append([]byte(nil), w.buf[cut:]...)
90 w.buf = w.buf[:cut]
91 w.emit()
92 w.buf = append(w.buf, rest...)
93}
94
95func (w *lineWriter) emit() {
96 b := w.buf
97 if !utf8.Valid(b) {
98 b = bytes.ToValidUTF8(b, []byte("\ufffd"))
99 }
100 line := string(b)
101 w.buf = w.buf[:0]
102 w.fn(line)
103}
104
105// Flush emits the last line when the stream ended without a terminator.
106func (w *lineWriter) Flush() {
107 if len(w.buf) > 0 {
108 w.emit()
109 }
110}
111
112// dispatchBatchDelay is how long the dispatcher lets lines pile up after the
113// first one before it hands a batch over. The official runner hands over
114// whatever its reader threads queued when its loop wakes up (a thread-pool
115// wake-up, about a millisecond); Go wakes in microseconds, so without this
116// pause the interleaving of a fast script's stdout and stderr would differ
117// from the official runner's more often.
118const dispatchBatchDelay = 2 * time.Millisecond
119
120// dispatcher reproduces how ProcessInvoker delivers output: stdout and stderr
121// are read by two readers into two queues, and a loop that wakes up on a
122// signal hands over all queued stderr lines first and then all queued stdout
123// lines (ProcessInvoker.ProcessOutput, ProcessInvoker.cs:392-436). Within a
124// stream the order is kept; between the streams a batch puts stderr first,
125// which is visible with `bash -x` (the corpus recording of shell-custom).
126type dispatcher struct {
127 fn func(string)
128 mu sync.Mutex
129 outQ []string
130 errQ []string
131 signal chan struct{}
132 closed chan struct{}
133 finished chan struct{}
134 out, err *lineWriter
135}
136
137func newDispatcher(fn func(string)) *dispatcher {
138 d := &dispatcher{fn: fn, signal: make(chan struct{}, 1), closed: make(chan struct{}), finished: make(chan struct{})}
139 push := func(q *[]string) func(string) {
140 return func(line string) {
141 d.mu.Lock()
142 *q = append(*q, line)
143 d.mu.Unlock()
144 select {
145 case d.signal <- struct{}{}:
146 default:
147 }
148 }
149 }
150 d.out = &lineWriter{fn: push(&d.outQ)}
151 d.err = &lineWriter{fn: push(&d.errQ)}
152 go d.loop()
153 return d
154}
155
156func (d *dispatcher) drain() {
157 d.mu.Lock()
158 errs, outs := d.errQ, d.outQ
159 d.errQ, d.outQ = nil, nil
160 d.mu.Unlock()
161 for _, l := range errs {
162 d.fn(l)
163 }
164 for _, l := range outs {
165 d.fn(l)
166 }
167}
168
169func (d *dispatcher) loop() {
170 defer close(d.finished)
171 for {
172 select {
173 case <-d.signal:
174 time.Sleep(dispatchBatchDelay)
175 d.drain()
176 case <-d.closed:
177 d.drain()
178 return
179 }
180 }
181}
182
183// Close flushes the readers and delivers what is left. Call it once the
184// process and its pipes are done.
185func (d *dispatcher) Close() {
186 d.err.Flush()
187 d.out.Flush()
188 close(d.closed)
189 <-d.finished
190}
191
192// procSpec describes one process to run.
193type procSpec struct {
194 Argv []string
195 Dir string
196 Env []string
197 // Credential, when set, is the user the process runs as.
198 Credential *syscall.Credential
199 // OnLine receives the lines of stdout and stderr, one call at a time, in
200 // the order of dispatcher.
201 OnLine func(line string)
202 // Cancel starts the SIGINT, SIGTERM, SIGKILL escalation when closed.
203 Cancel <-chan struct{}
204
205 TermAfter, KillAfter, WaitDelay time.Duration
206 // Signal delivers a signal to the process the step started (the shell);
207 // tests replace it.
208 Signal func(pid int, sig syscall.Signal)
209}
210
211// procResult is how a process ended.
212type procResult struct {
213 ExitCode int
214 Cancelled bool // Cancel was closed while the process ran
215}
216
217// runProcess starts the process in its own process group, streams its output
218// and waits. It is ProcessInvoker.ExecuteAsync for one process: stdin is
219// closed, and a closed Cancel starts the escalation. After the shell exits the
220// group is not touched (background children survive, like on a hosted runner
221// until the orphan cleanup of "Complete job").
222func runProcess(spec procSpec) (procResult, error) {
223 if spec.TermAfter == 0 {
224 spec.TermAfter = DefaultTermAfter
225 }
226 if spec.KillAfter == 0 {
227 spec.KillAfter = DefaultKillAfter
228 }
229 if spec.WaitDelay == 0 {
230 spec.WaitDelay = DefaultWaitDelay
231 }
232 if spec.Signal == nil {
233 // Like the official runner, which signals the process it started and
234 // not its group: ProcessInvoker.SendSignal calls kill(_proc.Id, signal)
235 // for SIGINT and SIGTERM (Runner.Sdk/ProcessInvoker.cs:829-833), and the
236 // last resort is Process.Kill() of that process alone (:855-869, which
237 // is named "KillProcessTree" but does not kill the tree). What the shell
238 // started is left to the shell: bash forwards nothing to a foreground
239 // child it waits for, and the children that outlive the step are
240 // killed by "Complete job" (the orphan cleanup, JobExtension.cs:891-939).
241 // The step is still started in its own process group, so that a signal
242 // sent to the executor's group (init forwarding a SIGTERM) does not reach
243 // it before the executor decides.
244 spec.Signal = func(pid int, sig syscall.Signal) { _ = syscall.Kill(pid, sig) }
245 }
246 cmd := exec.Command(spec.Argv[0], spec.Argv[1:]...)
247 cmd.Dir = spec.Dir
248 cmd.Env = spec.Env
249 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true, Credential: spec.Credential}
250 disp := newDispatcher(spec.OnLine)
251 cmd.Stdout, cmd.Stderr = disp.out, disp.err
252 cmd.WaitDelay = spec.WaitDelay
253 if err := cmd.Start(); err != nil {
254 return procResult{}, err
255 }
256 pid := cmd.Process.Pid
257 setOOMScoreAdj(pid)
258
259 var mu sync.Mutex
260 done := false
261 cancelled := false
262 var timers []*time.Timer
263 stopWatch := make(chan struct{})
264 watchDone := make(chan struct{})
265 go func() {
266 defer close(watchDone)
267 select {
268 case <-spec.Cancel:
269 case <-stopWatch:
270 return
271 }
272 mu.Lock()
273 defer mu.Unlock()
274 if done {
275 return
276 }
277 cancelled = true
278 spec.Signal(pid, syscall.SIGINT)
279 send := func(sig syscall.Signal) func() {
280 return func() {
281 mu.Lock()
282 defer mu.Unlock()
283 if !done {
284 spec.Signal(pid, sig)
285 }
286 }
287 }
288 timers = append(timers,
289 time.AfterFunc(spec.TermAfter, send(syscall.SIGTERM)),
290 time.AfterFunc(spec.TermAfter+spec.KillAfter, send(syscall.SIGKILL)))
291 }()
292
293 err := cmd.Wait()
294 mu.Lock()
295 done = true
296 for _, t := range timers {
297 t.Stop()
298 }
299 was := cancelled
300 mu.Unlock()
301 close(stopWatch)
302 <-watchDone
303 disp.Close()
304
305 res := procResult{Cancelled: was}
306 var ee *exec.ExitError
307 switch {
308 case err == nil, errors.Is(err, exec.ErrWaitDelay):
309 case errors.As(err, &ee):
310 res.ExitCode = ee.ExitCode()
311 if res.ExitCode < 0 { // killed by a signal: 128+signal, as .NET reports it
312 if ws, ok := ee.Sys().(syscall.WaitStatus); ok && ws.Signaled() {
313 res.ExitCode = 128 + int(ws.Signal())
314 } else {
315 res.ExitCode = 1
316 }
317 }
318 default:
319 return res, err
320 }
321 return res, nil
322}
323
324// setOOMScoreAdj makes the step process a likelier OOM victim than the
325// executor (ProcessInvoker.DecreaseProcessPriority, ProcessInvoker.cs:575).
326func setOOMScoreAdj(pid int) {
327 _ = os.WriteFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid), []byte("500"), 0o644)
328}
329
330// startError turns the error of a failed start into the text of .NET's
331// Win32Exception, which the official runner logs (the corpus recording
332// "working directory that does not exist" has it).
333func startError(shellPath, dir string, err error) string {
334 if errors.Is(err, os.ErrNotExist) || errors.Is(err, syscall.ENOENT) {
335 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. No such file or directory", shellPath, dir)
336 }
337 if errors.Is(err, syscall.ENOTDIR) {
338 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. Not a directory", shellPath, dir)
339 }
340 if errors.Is(err, os.ErrPermission) {
341 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. Permission denied", shellPath, dir)
342 }
343 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. %v", shellPath, dir, err)
344}
345
346// ProcessTrackingID is the variable that marks every process of a job so that
347// "Complete job" can find the orphans (Constants.ProcessTrackingId).
348const ProcessTrackingID = "RUNNER_TRACKING_ID"
349
350// procEntry is a process seen in /proc.
351type procEntry struct {
352 PID int
353 Name string // comm, as .NET's ProcessName
354}
355
356func (p procEntry) key() string { return strconv.Itoa(p.PID) + "_" + p.Name }
357
358// snapshotProcesses lists the processes (JobExtension.SnapshotProcesses).
359func snapshotProcesses() []procEntry {
360 ents, err := os.ReadDir("/proc")
361 if err != nil {
362 return nil
363 }
364 var out []procEntry
365 for _, e := range ents {
366 pid, err := strconv.Atoi(e.Name())
367 if err != nil {
368 continue
369 }
370 comm, err := os.ReadFile("/proc/" + e.Name() + "/comm")
371 if err != nil {
372 continue
373 }
374 out = append(out, procEntry{pid, strings.TrimSuffix(string(comm), "\n")})
375 }
376 sort.Slice(out, func(i, j int) bool { return out[i].PID < out[j].PID })
377 return out
378}
379
380// processTrackingValue reads RUNNER_TRACKING_ID from the environment of a
381// process; "" when it cannot be read.
382func processTrackingValue(pid int) string {
383 b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/environ")
384 if err != nil {
385 return ""
386 }
387 prefix := []byte(ProcessTrackingID + "=")
388 for _, kv := range bytes.Split(b, []byte{0}) {
389 if bytes.HasPrefix(kv, prefix) {
390 return string(kv[len(prefix):])
391 }
392 }
393 return ""
394}
395
396// killOrphans kills the processes that carry the tracking id and were not
397// there before the job (JobExtension.FinalizeJob, JobExtension.cs:891-939).
398// logf receives the "Terminate orphan process" lines.
399func killOrphans(id string, before map[string]bool, logf func(string)) {
400 self := os.Getpid()
401 for _, p := range snapshotProcesses() {
402 if p.PID == self || before[p.key()] {
403 continue
404 }
405 if !strings.EqualFold(processTrackingValue(p.PID), id) {
406 continue
407 }
408 logf(fmt.Sprintf("Terminate orphan process: pid (%d) (%s)", p.PID, p.Name))
409 _ = syscall.Kill(p.PID, syscall.SIGKILL)
410 }
411}
412
413var _ io.Writer = (*lineWriter)(nil)