1// Package ffmpeg encodes silent SDR video by streaming raw frames to an FFmpeg
2// child process that muxes fragmented MP4 into a reserved output file.
3//
4// Encoder settings favour a small footprint over compression: libx264 veryfast
5// with -tune zerolatency (no B-frames, no lookahead, sliced threads), two
6// encoder threads and one filter thread. Measured on 60 paced 30 fps 1080p
7// frames, against the earlier "fast" preset with 10-frame lookahead, this cut
8// peak RSS from about 335 MB to about 141 MB and CPU time by more than half,
9// and made the file about 43% larger (3.8 MB to 5.4 MB) at the same CRF 23.
10// ultrafast would save only about 8 MB more but grow files a further 37%.
11package ffmpeg
12
13import (
14 "bytes"
15 "context"
16 "errors"
17 "fmt"
18 "io"
19 "os"
20 "os/exec"
21 "strconv"
22 "sync"
23 "sync/atomic"
24 "syscall"
25 "time"
26
27 "github.com/bnema/nefercap/internal/adapters/capturefile"
28 "github.com/bnema/nefercap/internal/ports"
29)
30
31const (
32 defaultBinary = "ffmpeg"
33 // closeTimeout bounds how long Close waits for the encoder to drain after
34 // EOF before killing it.
35 closeTimeout = 15 * time.Second
36 // killTimeout bounds the wait for the reaper after a kill.
37 killTimeout = 5 * time.Second
38 // exitProbe bounds how long a failed pipe write waits for the exit status.
39 exitProbe = time.Second
40 // waitDelay bounds Wait's I/O draining once the process has exited.
41 waitDelay = 2 * time.Second
42 // stderrLimit caps retained encoder diagnostics.
43 stderrLimit = 4 << 10
44)
45
46// ErrNoFrames is returned by Close when no frame was written, so no video exists.
47var ErrNoFrames = errors.New("ffmpeg: no frames written")
48
49// past is a write deadline that is always expired.
50var past = time.Unix(1, 0)
51
52const (
53 stateIdle int32 = iota
54 stateRunning
55 stateClosing
56 stateClosed
57)
58
59// failure is the sticky first error. fatal marks a real stream failure (as
60// opposed to a caller's cancellation or abort), which Close must report even
61// if the encoder itself exits cleanly.
62type failure struct {
63 err error
64 fatal bool
65}
66
67// Writer implements ports.VideoWriter. Start, Write and Close have one owner;
68// Abort may be called concurrently with Write or a draining Close.
69// A Writer is single-use.
70type Writer struct {
71 binary string
72 closeTimeout time.Duration
73
74 state atomic.Int32
75 mu sync.Mutex // serializes state transitions
76
77 // Set by Start before state becomes running; read-only afterwards.
78 cmd *exec.Cmd
79 stdin *os.File
80 res *capturefile.Reservation
81 width int
82 height int
83 format ports.PixelFormat
84 stderr *boundedBuffer
85 waitDone chan struct{} // closed once cmd.Wait has returned
86 waitErr error // valid after waitDone is closed
87
88 frames atomic.Int64
89 cause atomic.Pointer[failure] // sticky first failure or interruption
90
91 watchMu sync.Mutex
92 watchDead bool
93 // Saved reports whether Close retained the reserved file. Read it only
94 // after the owning Close or Abort returns, independently of capture errors.
95 Saved bool
96 startStop func() bool
97 writeStop func() bool
98 writeDone <-chan struct{}
99 writeGen uint64
100 active atomic.Uint64 // generation of the Write using a per-call watcher
101
102 abortCh chan struct{}
103 abortOnce sync.Once
104 termDone chan struct{}
105 termErr error
106}
107
108var _ ports.VideoWriter = (*Writer)(nil)
109
110// New returns a Writer that runs the ffmpeg found on PATH.
111func New() *Writer { return NewWithBinary(defaultBinary) }
112
113// NewWithBinary returns a Writer that runs the given encoder executable, which
114// must accept FFmpeg's command line. It is used to test with helper processes.
115func NewWithBinary(binary string) *Writer {
116 return &Writer{
117 binary: binary,
118 closeTimeout: closeTimeout,
119 abortCh: make(chan struct{}),
120 termDone: make(chan struct{}),
121 }
122}
123
124// outputSize validates settings and returns the encoded dimensions. Zero
125// dimensions retain the source resolution exactly, which yuv420p only allows
126// for even sizes; odd sources are rejected rather than cropped. Explicit
127// sizes must be even and no larger than a captured frame may be.
128func outputSize(f ports.Frame, s ports.VideoSettings) (int, int, error) {
129 if s.FPS < 1 || s.FPS > ports.MaxFPS {
130 return 0, 0, fmt.Errorf("ffmpeg: fps %d outside 1..%d", s.FPS, ports.MaxFPS)
131 }
132 if s.Width == 0 && s.Height == 0 {
133 if f.Width%2 != 0 || f.Height%2 != 0 {
134 return 0, 0, fmt.Errorf("ffmpeg: source %dx%d has an odd dimension; set an even output size to scale", f.Width, f.Height)
135 }
136 return f.Width, f.Height, nil
137 }
138 if s.Width < 2 || s.Height < 2 || s.Width > ports.MaxDimension || s.Height > ports.MaxDimension ||
139 s.Width%2 != 0 || s.Height%2 != 0 {
140 return 0, 0, fmt.Errorf("ffmpeg: output size %dx%d must be even, positive and at most %d", s.Width, s.Height, ports.MaxDimension)
141 }
142 if s.Width > ports.MaxFrameBytes/ports.BytesPerPixel/s.Height {
143 return 0, 0, fmt.Errorf("ffmpeg: output size %dx%d exceeds %d bytes per frame", s.Width, s.Height, ports.MaxFrameBytes)
144 }
145 return s.Width, s.Height, nil
146}
147
148// buildArgs returns the FFmpeg argument vector (no shell). Input is raw bgr0
149// on stdin; MP4 is fragmented to fd 3, an inherited pre-opened output file.
150// The scale filter always runs: with unchanged geometry it only performs the
151// required RGB to BT.709 limited-range YUV conversion and does not resample.
152func buildArgs(f ports.Frame, outW, outH, fps int) []string {
153 filter := "scale=" + strconv.Itoa(outW) + ":" + strconv.Itoa(outH) +
154 ":flags=bicubic:out_color_matrix=bt709:out_range=tv,format=yuv420p"
155 rate := strconv.Itoa(fps)
156 return []string{
157 "-hide_banner", "-nostats", "-loglevel", "error",
158 "-f", "rawvideo", "-pix_fmt", "bgr0",
159 "-video_size", strconv.Itoa(f.Width) + "x" + strconv.Itoa(f.Height),
160 "-framerate", rate, "-i", "pipe:0",
161 "-filter_threads", "1", "-an", "-vf", filter,
162 "-c:v", "libx264", "-threads", "2", "-preset", "veryfast", "-tune", "zerolatency", "-crf", "23",
163 "-g", strconv.Itoa(fps * 2), "-r", rate,
164 "-colorspace", "bt709", "-color_primaries", "bt709", "-color_trc", "bt709", "-color_range", "tv",
165 "-movflags", "frag_keyframe+empty_moov+default_base_moof",
166 "-f", "mp4", "pipe:3",
167 }
168}
169
170// Start launches the encoder and reserves path with O_EXCL; an existing path
171// yields ports.ErrPathExists and is never touched. Frame geometry and format
172// are fixed from firstFrame; the frame itself is not written. Cancelling ctx
173// interrupts a blocked Write but does not kill the encoder: call Close to
174// finalize the video.
175func (w *Writer) Start(ctx context.Context, first ports.Frame, path string, s ports.VideoSettings) error {
176 w.mu.Lock()
177 defer w.mu.Unlock()
178 if w.state.Load() != stateIdle || w.cmd != nil {
179 return errors.New("ffmpeg: writer already started")
180 }
181 if err := ctx.Err(); err != nil {
182 return err
183 }
184 if err := first.Validate(); err != nil {
185 return err
186 }
187 if path == "" {
188 return errors.New("ffmpeg: empty path")
189 }
190 outW, outH, err := outputSize(first, s)
191 if err != nil {
192 return err
193 }
194 bin, err := exec.LookPath(w.binary)
195 if err != nil {
196 return fmt.Errorf("ffmpeg: %w", err)
197 }
198 out, res, err := capturefile.Create(path)
199 if err != nil {
200 return fmt.Errorf("ffmpeg: %w", err)
201 }
202 // From here we own the created file and remove it (not a replacement) on failure.
203 fail := func(err error, closers ...*os.File) error {
204 for _, c := range closers {
205 _ = c.Close()
206 }
207 if rerr := res.Remove(); rerr != nil {
208 err = errors.Join(err, fmt.Errorf("ffmpeg: %w", rerr))
209 }
210 return err
211 }
212 pr, pw, err := os.Pipe()
213 if err != nil {
214 return fail(fmt.Errorf("ffmpeg: stdin pipe: %w", err), out)
215 }
216 stderr := &boundedBuffer{limit: stderrLimit}
217 cmd := exec.Command(bin, buildArgs(first, outW, outH, s.FPS)...)
218 cmd.Stdin = pr
219 cmd.Stderr = stderr
220 cmd.ExtraFiles = []*os.File{out}
221 cmd.WaitDelay = waitDelay
222 // Own process group: a terminal SIGINT aimed at the recorder must not kill
223 // the encoder before it receives EOF and finalizes the file.
224 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
225 if err := cmd.Start(); err != nil {
226 _ = pr.Close()
227 _ = pw.Close()
228 return fail(fmt.Errorf("ffmpeg: start: %w", err), out)
229 }
230 // The child holds its own copies of these descriptors.
231 _ = pr.Close()
232 _ = out.Close()
233
234 w.cmd, w.stdin, w.res, w.stderr = cmd, pw, res, stderr
235 w.width, w.height, w.format = first.Width, first.Height, first.Format
236 w.waitDone = make(chan struct{})
237 go func() { // sole reaper: runs Wait exactly once, also after any Kill
238 w.waitErr = cmd.Wait()
239 close(w.waitDone)
240 }()
241 if ctx.Done() != nil {
242 w.startStop = context.AfterFunc(ctx, func() { w.interrupt(ctx.Err()) })
243 }
244 w.state.Store(stateRunning)
245 return nil
246}
247
248// interrupt records the first failure and unblocks a pending pipe write.
249func (w *Writer) interrupt(err error) {
250 w.cause.CompareAndSwap(nil, &failure{err: err})
251 _ = w.stdin.SetWriteDeadline(past)
252}
253
254// watch makes cancellation of a Write context interrupt the pipe. The watcher
255// is registered once per distinct context, not per frame, and only fires while
256// a Write using it is active.
257func (w *Writer) watch(ctx context.Context) (gen uint64, ok bool) {
258 done := ctx.Done()
259 if done == nil {
260 return 0, false
261 }
262 w.watchMu.Lock()
263 defer w.watchMu.Unlock()
264 if w.watchDead {
265 return 0, false
266 }
267 if done != w.writeDone {
268 if w.writeStop != nil {
269 w.writeStop()
270 }
271 w.writeGen++
272 g := w.writeGen
273 w.writeDone = done
274 w.writeStop = context.AfterFunc(ctx, func() {
275 if w.active.Load() == g {
276 w.interrupt(ctx.Err())
277 }
278 })
279 }
280 return w.writeGen, true
281}
282
283// Write sends one frame. Rows are written straight from borrowed storage, so
284// pixels are not retained. Width, height and format must match Start.
285func (w *Writer) Write(ctx context.Context, f ports.Frame) error {
286 if w.state.Load() != stateRunning {
287 return ports.ErrClosed
288 }
289 if c := w.cause.Load(); c != nil {
290 return c.err
291 }
292 if err := ctx.Err(); err != nil {
293 return err
294 }
295 if err := f.Validate(); err != nil {
296 return err
297 }
298 if f.Width != w.width || f.Height != w.height || f.Format != w.format {
299 return ports.ErrGeometryChanged
300 }
301 gen, watched := w.watch(ctx)
302 if watched {
303 w.active.Store(gen)
304 defer w.active.Store(0)
305 if err := ctx.Err(); err != nil {
306 return err
307 }
308 }
309 if err := writeRows(w.stdin, f); err != nil {
310 return w.writeFailed(err)
311 }
312 w.frames.Add(1)
313 return nil
314}
315
316func (w *Writer) writeFailed(err error) error {
317 if w.state.Load() != stateRunning {
318 return ports.ErrClosed
319 }
320 if c := w.cause.Load(); c != nil {
321 return c.err
322 }
323 // The encoder most likely exited; give it a moment to report why.
324 t := time.NewTimer(exitProbe)
325 defer t.Stop()
326 select {
327 case <-w.waitDone:
328 case <-t.C:
329 }
330 msg := fmt.Errorf("ffmpeg: write frame: %w%s", err, w.diagnostics())
331 w.cause.CompareAndSwap(nil, &failure{err: msg, fatal: true})
332 return w.cause.Load().err
333}
334
335// diagnostics describes the exit state and captured stderr once the process
336// has been reaped; it is empty while the process is still running.
337func (w *Writer) diagnostics() string {
338 select {
339 case <-w.waitDone:
340 default:
341 return ""
342 }
343 s := ""
344 if w.waitErr != nil {
345 s = ": " + w.waitErr.Error()
346 }
347 if d := w.stderr.String(); d != "" {
348 s += ": " + d
349 }
350 return s
351}
352
353// writeRows writes the frame in top-to-bottom row order without copying it.
354// A tightly packed frame is a single write.
355func writeRows(dst io.Writer, f ports.Frame) error {
356 rowLen := f.Width * ports.BytesPerPixel
357 if f.Stride == rowLen {
358 return writeFull(dst, f.Pixels[:rowLen*f.Height])
359 }
360 for y := 0; y < f.Height; y++ {
361 if err := writeFull(dst, f.Row(y)); err != nil {
362 return err
363 }
364 }
365 return nil
366}
367
368// writeFull enforces the io.Writer contract: n < len(p) needs a non-nil error.
369func writeFull(dst io.Writer, p []byte) error {
370 n, err := dst.Write(p)
371 if err == nil && n < len(p) {
372 return io.ErrShortWrite
373 }
374 return err
375}
376
377// Close signals EOF, waits for the encoder to finalize the file (bounded, then
378// kills it) and reports the outcome. The video is kept when the encoder exits
379// cleanly after at least one frame, including after a cancelled context; it is
380// removed otherwise. Close and Abort are terminal and the first one wins:
381// later Close calls return the first Close result. Close after an Abort returns
382// its cleanup result (nil unless reaping or removal failed); nothing was kept.
383// Close may run concurrently with Abort as described on Abort, but not Write.
384func (w *Writer) Close() error { return w.terminate(false) }
385
386// Abort discards the recording. When it is the first terminal operation it
387// interrupts a pending Write, kills the encoder and removes the output file,
388// even if frames were already written and the encoder was healthy; use Close to
389// keep a video. It returns nil unless the encoder could not be reaped or the
390// file could not be removed. It is safe concurrently with Write and idempotent.
391//
392// If Abort arrives while a Close is still draining, it escalates that Close to
393// a kill and the output is removed, unless the encoder had already exited: an
394// exit observed first lets Close finish and keep the video. Close's result is
395// authoritative in that race, and the Abort call itself then returns nil. Abort
396// after Close has finished is a no-op.
397func (w *Writer) Abort() error { return w.terminate(true) }
398
399func (w *Writer) terminate(abort bool) error {
400 w.mu.Lock()
401 first := false
402 switch w.state.Load() {
403 case stateIdle:
404 w.state.Store(stateClosed)
405 close(w.termDone)
406 w.mu.Unlock()
407 return nil
408 case stateRunning:
409 w.state.Store(stateClosing)
410 first = true
411 }
412 w.mu.Unlock()
413
414 if abort {
415 w.abortOnce.Do(func() { close(w.abortCh) })
416 w.interrupt(ports.ErrClosed)
417 }
418 if first {
419 w.termErr = w.shutdown(abort)
420 w.state.Store(stateClosed)
421 close(w.termDone)
422 } else {
423 <-w.termDone
424 }
425 if abort && !first {
426 return nil // the first terminal call owns the result
427 }
428 return w.termErr
429}
430
431func (w *Writer) shutdown(abort bool) error {
432 w.watchMu.Lock()
433 w.watchDead = true
434 for _, stop := range []func() bool{w.startStop, w.writeStop} {
435 if stop != nil {
436 stop()
437 }
438 }
439 w.watchMu.Unlock()
440
441 _ = w.stdin.Close() // EOF: the encoder finalizes the file
442
443 if abort {
444 return w.discard()
445 }
446 err := w.awaitExit()
447 if err != nil {
448 _ = w.cmd.Process.Kill() // ErrProcessDone is harmless
449 kt := time.NewTimer(killTimeout)
450 defer kt.Stop()
451 select {
452 case <-w.waitDone:
453 case <-kt.C:
454 err = errors.Join(err, errors.New("ffmpeg: encoder not reaped after kill"))
455 }
456 }
457 if err == nil {
458 select {
459 case <-w.waitDone:
460 if w.waitErr != nil {
461 err = fmt.Errorf("ffmpeg: encoder failed%s", w.diagnostics())
462 }
463 default:
464 }
465 } else {
466 err = fmt.Errorf("%w%s", err, w.diagnostics())
467 }
468 if c := w.cause.Load(); err == nil && c != nil && c.fatal {
469 // The stream broke mid-recording even though the encoder exited cleanly.
470 err = c.err
471 }
472 if err == nil && w.frames.Load() == 0 {
473 err = ErrNoFrames
474 }
475 if err != nil {
476 err = errors.Join(err, w.removeOutput())
477 } else {
478 w.Saved = w.res.Owns()
479 }
480 return err
481}
482
483// awaitExit waits for the encoder to exit after EOF. It returns nil if the
484// process has exited, else the reason the wait was cut short (timeout or
485// Abort). An exit that is already visible wins over a timeout or Abort that
486// became ready at the same time; the remaining race is inherent and documented
487// on Abort.
488func (w *Writer) awaitExit() error {
489 timer := time.NewTimer(w.closeTimeout)
490 defer timer.Stop()
491 var err error
492 select {
493 case <-w.waitDone:
494 return nil
495 case <-timer.C:
496 err = fmt.Errorf("ffmpeg: encoder did not finish within %s; killed", w.closeTimeout)
497 case <-w.abortCh:
498 err = errors.New("ffmpeg: shutdown aborted; encoder killed")
499 }
500 select {
501 case <-w.waitDone:
502 return nil
503 default:
504 return err
505 }
506}
507
508// discard kills the encoder, waits for the reaper and removes the output.
509func (w *Writer) discard() error {
510 _ = w.cmd.Process.Kill() // ErrProcessDone is harmless
511 var err error
512 kt := time.NewTimer(killTimeout)
513 defer kt.Stop()
514 select {
515 case <-w.waitDone:
516 case <-kt.C:
517 err = errors.New("ffmpeg: encoder not reaped after kill")
518 }
519 return errors.Join(err, w.removeOutput())
520}
521
522// removeOutput deletes the file Start reserved, but not a replacement.
523func (w *Writer) removeOutput() error {
524 if err := w.res.Remove(); err != nil {
525 return fmt.Errorf("ffmpeg: %w", err)
526 }
527 return nil
528}
529
530// boundedBuffer keeps the first limit bytes written and discards the rest, so
531// a chatty encoder cannot grow memory.
532type boundedBuffer struct {
533 mu sync.Mutex
534 buf []byte
535 limit int
536 truncated bool
537}
538
539func (b *boundedBuffer) Write(p []byte) (int, error) {
540 b.mu.Lock()
541 defer b.mu.Unlock()
542 if room := b.limit - len(b.buf); room > 0 {
543 if room > len(p) {
544 room = len(p)
545 }
546 b.buf = append(b.buf, p[:room]...)
547 if room < len(p) {
548 b.truncated = true
549 }
550 } else if len(p) > 0 {
551 b.truncated = true
552 }
553 return len(p), nil
554}
555
556func (b *boundedBuffer) String() string {
557 b.mu.Lock()
558 defer b.mu.Unlock()
559 s := string(bytes.TrimSpace(b.buf))
560 if b.truncated {
561 s += " [truncated]"
562 }
563 return s
564}