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 errFn func(string) // when set, stderr lines go here instead of fn
129 mu sync.Mutex
130 outQ []string
131 errQ []string
132 signal chan struct{}
133 closed chan struct{}
134 finished chan struct{}
135 out, err *lineWriter
136}
137
138func newDispatcher(fn func(string)) *dispatcher {
139 d := &dispatcher{fn: fn, signal: make(chan struct{}, 1), closed: make(chan struct{}), finished: make(chan struct{})}
140 push := func(q *[]string) func(string) {
141 return func(line string) {
142 d.mu.Lock()
143 *q = append(*q, line)
144 d.mu.Unlock()
145 select {
146 case d.signal <- struct{}{}:
147 default:
148 }
149 }
150 }
151 d.out = &lineWriter{fn: push(&d.outQ)}
152 d.err = &lineWriter{fn: push(&d.errQ)}
153 go d.loop()
154 return d
155}
156
157func (d *dispatcher) drain() {
158 d.mu.Lock()
159 errs, outs := d.errQ, d.outQ
160 d.errQ, d.outQ = nil, nil
161 d.mu.Unlock()
162 for _, l := range errs {
163 if d.errFn != nil {
164 d.errFn(l)
165 continue
166 }
167 d.fn(l)
168 }
169 for _, l := range outs {
170 d.fn(l)
171 }
172}
173
174func (d *dispatcher) loop() {
175 defer close(d.finished)
176 for {
177 select {
178 case <-d.signal:
179 time.Sleep(dispatchBatchDelay)
180 d.drain()
181 case <-d.closed:
182 d.drain()
183 return
184 }
185 }
186}
187
188// Close flushes the readers and delivers what is left. Call it once the
189// process and its pipes are done.
190func (d *dispatcher) Close() {
191 d.err.Flush()
192 d.out.Flush()
193 close(d.closed)
194 <-d.finished
195}
196
197// procSpec describes one process to run.
198type procSpec struct {
199 Argv []string
200 Dir string
201 Env []string
202 // Credential, when set, is the user the process runs as.
203 Credential *syscall.Credential
204 // OnLine receives the lines of stdout and stderr, one call at a time, in
205 // the order of dispatcher.
206 OnLine func(line string)
207 // OnErrLine, when set, receives the lines of stderr instead of OnLine.
208 OnErrLine func(line string)
209 // Stdin, when set, is written to the standard input of the process (the
210 // password of docker login); otherwise stdin is empty.
211 Stdin string
212 // Cancel starts the SIGINT, SIGTERM, SIGKILL escalation when closed.
213 Cancel <-chan struct{}
214
215 TermAfter, KillAfter, WaitDelay time.Duration
216 // Signal delivers a signal to the process the step started (the shell);
217 // tests replace it.
218 Signal func(pid int, sig syscall.Signal)
219}
220
221// procResult is how a process ended.
222type procResult struct {
223 ExitCode int
224 Cancelled bool // Cancel was closed while the process ran
225}
226
227// runProcess starts the process in its own process group, streams its output
228// and waits. It is ProcessInvoker.ExecuteAsync for one process: stdin is
229// closed, and a closed Cancel starts the escalation. After the shell exits the
230// group is not touched (background children survive, like on a hosted runner
231// until the orphan cleanup of "Complete job").
232func runProcess(spec procSpec) (procResult, error) {
233 if spec.TermAfter == 0 {
234 spec.TermAfter = DefaultTermAfter
235 }
236 if spec.KillAfter == 0 {
237 spec.KillAfter = DefaultKillAfter
238 }
239 if spec.WaitDelay == 0 {
240 spec.WaitDelay = DefaultWaitDelay
241 }
242 if spec.Signal == nil {
243 // Like the official runner, which signals the process it started and
244 // not its group: ProcessInvoker.SendSignal calls kill(_proc.Id, signal)
245 // for SIGINT and SIGTERM (Runner.Sdk/ProcessInvoker.cs:829-833), and the
246 // last resort is Process.Kill() of that process alone (:855-869, which
247 // is named "KillProcessTree" but does not kill the tree). What the shell
248 // started is left to the shell: bash forwards nothing to a foreground
249 // child it waits for, and the children that outlive the step are
250 // killed by "Complete job" (the orphan cleanup, JobExtension.cs:891-939).
251 // The step is still started in its own process group, so that a signal
252 // sent to the executor's group (init forwarding a SIGTERM) does not reach
253 // it before the executor decides.
254 spec.Signal = func(pid int, sig syscall.Signal) { _ = syscall.Kill(pid, sig) }
255 }
256 cmd := exec.Command(spec.Argv[0], spec.Argv[1:]...)
257 cmd.Dir = spec.Dir
258 cmd.Env = spec.Env
259 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true, Credential: spec.Credential}
260 disp := newDispatcher(spec.OnLine)
261 disp.errFn = spec.OnErrLine
262 if spec.Stdin != "" {
263 cmd.Stdin = strings.NewReader(spec.Stdin)
264 }
265 cmd.Stdout, cmd.Stderr = disp.out, disp.err
266 cmd.WaitDelay = spec.WaitDelay
267 if err := cmd.Start(); err != nil {
268 return procResult{}, err
269 }
270 pid := cmd.Process.Pid
271 setOOMScoreAdj(pid)
272
273 var mu sync.Mutex
274 done := false
275 cancelled := false
276 var timers []*time.Timer
277 stopWatch := make(chan struct{})
278 watchDone := make(chan struct{})
279 go func() {
280 defer close(watchDone)
281 select {
282 case <-spec.Cancel:
283 case <-stopWatch:
284 return
285 }
286 mu.Lock()
287 defer mu.Unlock()
288 if done {
289 return
290 }
291 cancelled = true
292 spec.Signal(pid, syscall.SIGINT)
293 send := func(sig syscall.Signal) func() {
294 return func() {
295 mu.Lock()
296 defer mu.Unlock()
297 if !done {
298 spec.Signal(pid, sig)
299 }
300 }
301 }
302 timers = append(timers,
303 time.AfterFunc(spec.TermAfter, send(syscall.SIGTERM)),
304 time.AfterFunc(spec.TermAfter+spec.KillAfter, send(syscall.SIGKILL)))
305 }()
306
307 err := cmd.Wait()
308 mu.Lock()
309 done = true
310 for _, t := range timers {
311 t.Stop()
312 }
313 was := cancelled
314 mu.Unlock()
315 close(stopWatch)
316 <-watchDone
317 disp.Close()
318
319 res := procResult{Cancelled: was}
320 var ee *exec.ExitError
321 switch {
322 case err == nil, errors.Is(err, exec.ErrWaitDelay):
323 case errors.As(err, &ee):
324 res.ExitCode = ee.ExitCode()
325 if res.ExitCode < 0 { // killed by a signal: 128+signal, as .NET reports it
326 if ws, ok := ee.Sys().(syscall.WaitStatus); ok && ws.Signaled() {
327 res.ExitCode = 128 + int(ws.Signal())
328 } else {
329 res.ExitCode = 1
330 }
331 }
332 default:
333 return res, err
334 }
335 return res, nil
336}
337
338// setOOMScoreAdj makes the step process a likelier OOM victim than the
339// executor (ProcessInvoker.DecreaseProcessPriority, ProcessInvoker.cs:575).
340func setOOMScoreAdj(pid int) {
341 _ = os.WriteFile(fmt.Sprintf("/proc/%d/oom_score_adj", pid), []byte("500"), 0o644)
342}
343
344// startError turns the error of a failed start into the text of .NET's
345// Win32Exception, which the official runner logs (the corpus recording
346// "working directory that does not exist" has it).
347func startError(shellPath, dir string, err error) string {
348 if errors.Is(err, os.ErrNotExist) || errors.Is(err, syscall.ENOENT) {
349 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. No such file or directory", shellPath, dir)
350 }
351 if errors.Is(err, syscall.ENOTDIR) {
352 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. Not a directory", shellPath, dir)
353 }
354 if errors.Is(err, os.ErrPermission) {
355 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. Permission denied", shellPath, dir)
356 }
357 return fmt.Sprintf("An error occurred trying to start process '%s' with working directory '%s'. %v", shellPath, dir, err)
358}
359
360// ProcessTrackingID is the variable that marks every process of a job so that
361// "Complete job" can find the orphans (Constants.ProcessTrackingId).
362const ProcessTrackingID = "RUNNER_TRACKING_ID"
363
364// procEntry is a process seen in /proc.
365type procEntry struct {
366 PID int
367 Name string // comm, as .NET's ProcessName
368}
369
370func (p procEntry) key() string { return strconv.Itoa(p.PID) + "_" + p.Name }
371
372// snapshotProcesses lists the processes (JobExtension.SnapshotProcesses).
373func snapshotProcesses() []procEntry {
374 ents, err := os.ReadDir("/proc")
375 if err != nil {
376 return nil
377 }
378 var out []procEntry
379 for _, e := range ents {
380 pid, err := strconv.Atoi(e.Name())
381 if err != nil {
382 continue
383 }
384 comm, err := os.ReadFile("/proc/" + e.Name() + "/comm")
385 if err != nil {
386 continue
387 }
388 out = append(out, procEntry{pid, strings.TrimSuffix(string(comm), "\n")})
389 }
390 sort.Slice(out, func(i, j int) bool { return out[i].PID < out[j].PID })
391 return out
392}
393
394// processTrackingValue reads RUNNER_TRACKING_ID from the environment of a
395// process; "" when it cannot be read.
396func processTrackingValue(pid int) string {
397 b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/environ")
398 if err != nil {
399 return ""
400 }
401 prefix := []byte(ProcessTrackingID + "=")
402 for _, kv := range bytes.Split(b, []byte{0}) {
403 if bytes.HasPrefix(kv, prefix) {
404 return string(kv[len(prefix):])
405 }
406 }
407 return ""
408}
409
410// killOrphans kills the processes that carry the tracking id and were not
411// there before the job (JobExtension.FinalizeJob, JobExtension.cs:891-939).
412// logf receives the "Terminate orphan process" lines.
413func killOrphans(id string, before map[string]bool, logf func(string)) {
414 self := os.Getpid()
415 for _, p := range snapshotProcesses() {
416 if p.PID == self || before[p.key()] {
417 continue
418 }
419 if !strings.EqualFold(processTrackingValue(p.PID), id) {
420 continue
421 }
422 logf(fmt.Sprintf("Terminate orphan process: pid (%d) (%s)", p.PID, p.Name))
423 _ = syscall.Kill(p.PID, syscall.SIGKILL)
424 }
425}
426
427var _ io.Writer = (*lineWriter)(nil)