hamlab/internal/media/ffrun.go

package media

// Running one long ffmpeg: as the daemon's user directly, or as root through this binary's
// run-ffmpeg helper when the file needs root to read. The helper is told to finish by its
// standard input closing, which also happens by itself if the daemon goes away — so a
// transcode never outlives the console that asked for it.

import (
	"bufio"
	"bytes"
	"context"
	"errors"
	"io"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"

	"hamlab/internal/priv"
)

type ffproc struct {
	cmd  *exec.Cmd
	in   io.WriteCloser // the helper's stdin, root mode only
	done chan struct{}
	err  error
	tail *tailBuf
}

// startFF launches ffmpeg with args, under nice where there is one.
func startFF(c Caps, self string, root bool, args []string) (*ffproc, error) {
	var argv []string
	if root {
		argv = priv.Wrap(append([]string{self, "run-ffmpeg"}, args...)...)
	} else {
		argv = append([]string{c.FFmpeg}, args...)
		if c.Nice != "" {
			argv = append([]string{c.Nice, "-n", "10"}, argv...)
		}
	}
	cmd := exec.Command(argv[0], argv[1:]...)
	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	p := &ffproc{cmd: cmd, done: make(chan struct{}), tail: &tailBuf{}}
	cmd.Stderr = p.tail
	cmd.Stdout = io.Discard
	if root {
		in, err := cmd.StdinPipe()
		if err != nil {
			return nil, err
		}
		p.in = in
	}
	if err := cmd.Start(); err != nil {
		return nil, err
	}
	go func() {
		p.err = cmd.Wait()
		close(p.done)
	}()
	return p, nil
}

// exited reports whether ffmpeg has finished, and how.
func (p *ffproc) exited() (bool, error) {
	select {
	case <-p.done:
		return true, p.err
	default:
		return false, nil
	}
}

// end asks ffmpeg to finish and waits a little for it.
func (p *ffproc) end() {
	if done, _ := p.exited(); done {
		return
	}
	if p.in != nil {
		p.in.Close()
	} else {
		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
	}
	select {
	case <-p.done:
	case <-time.After(5 * time.Second):
		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL) // a root helper ignores this; it ends on its own within its own grace
		select {
		case <-p.done:
		case <-time.After(10 * time.Second):
		}
	}
}

// pause stops ffmpeg where it is (SIGSTOP), and resume lets it go on: a stream far enough
// ahead of its player waits rather than making the rest of the film. A root run is paused by
// its helper, which is told on its standard input.
func (p *ffproc) pause() error  { return p.signal(syscall.SIGSTOP, "stop\n") }
func (p *ffproc) resume() error { return p.signal(syscall.SIGCONT, "cont\n") }

func (p *ffproc) signal(sig syscall.Signal, word string) error {
	if done, _ := p.exited(); done {
		return errors.New("ended")
	}
	if p.in != nil {
		_, err := io.WriteString(p.in, word)
		return err
	}
	return syscall.Kill(-p.cmd.Process.Pid, sig)
}

// kill ends ffmpeg now, without waiting: a paused one too.
func (p *ffproc) kill() {
	if done, _ := p.exited(); done {
		return
	}
	if p.in != nil {
		p.in.Close() // the helper ends it, waking it first if it is paused
		return
	}
	_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL)
}

// lastWords is what ffmpeg said last, for the record.
func (p *ffproc) lastWords() string { return p.tail.last() }

// tailBuf keeps the last few KB written to it.
type tailBuf struct {
	mu sync.Mutex
	b  []byte
}

const tailKeep = 4096

func (t *tailBuf) Write(b []byte) (int, error) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.b = append(t.b, b...)
	if len(t.b) > tailKeep {
		t.b = t.b[len(t.b)-tailKeep:]
	}
	return len(b), nil
}

func (t *tailBuf) last() string {
	t.mu.Lock()
	defer t.mu.Unlock()
	return lastLine(string(bytes.TrimSpace(t.b)))
}

// RunFFmpegHelper implements `hamlab run-ffmpeg <args…>`: ffmpeg as this user (root, when
// the daemon asked through sudo), finishing when stdin closes. Returns ffmpeg's exit code.
func RunFFmpegHelper(args []string, stdin io.Reader, stderr io.Writer) int {
	ffmpeg, err := exec.LookPath("ffmpeg")
	if err != nil {
		io.WriteString(stderr, "ffmpeg is not installed\n")
		return 1
	}
	argv := []string{ffmpeg}
	if nice, err := exec.LookPath("nice"); err == nil {
		argv = []string{nice, "-n", "10", ffmpeg}
	}
	cmd := exec.Command(argv[0], append(argv[1:], args...)...)
	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	cmd.Stderr = stderr
	cmd.Stdout = io.Discard
	if err := cmd.Start(); err != nil {
		io.WriteString(stderr, err.Error()+"\n")
		return 1
	}
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		// "stop" and "cont" pause and resume it; the end of input (the daemon letting go)
		// ends it.
		sc := bufio.NewScanner(stdin)
		for sc.Scan() {
			switch sc.Text() {
			case "stop":
				_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGSTOP)
			case "cont":
				_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGCONT)
			}
		}
		cancel()
	}()
	done := make(chan error, 1)
	go func() { done <- cmd.Wait() }()
	var werr error
	select {
	case werr = <-done:
	case <-ctx.Done():
		_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM)
		_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGCONT) // a paused one can't act on TERM
		select {
		case werr = <-done:
		case <-time.After(5 * time.Second):
			_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
			werr = <-done
		}
	}
	var ee *exec.ExitError
	if errors.As(werr, &ee) {
		return ee.ExitCode()
	}
	if werr != nil {
		return 1
	}
	return 0
}

// clearDir empties a folder without removing it.
func clearDir(dir string) error {
	ents, err := os.ReadDir(dir)
	if err != nil {
		return err
	}
	for _, e := range ents {
		if err := os.RemoveAll(dir + "/" + e.Name()); err != nil {
			return err
		}
	}
	return nil
}