encoderd: add fast hardware clip transcoding (#38508)

openpilot/system/athena/athenad.py

@@ -34,6 +34,7 @@ from openpilot.cereal import log
 from opendbc.car.structs import car
 from openpilot.cereal.services import SERVICE_LIST
 from openpilot.common.api import Api, get_key_pair
+from openpilot.common.basedir import BASEDIR
 from openpilot.common.utils import CallbackReader, get_upload_stream
 from openpilot.common.params import Params
 from openpilot.common.realtime import set_core_affinity
@@ -423,32 +424,40 @@ class VideoClips:
     threading.Thread(target=self._worker, name="video_clip", daemon=True).start()
 
   def _encode(self, clip: Clip, inputs: Iterable[str], output_path: str, start_time: float, duration: float) -> None:
-    # TODO: use hardware accelerated decoding and encoding
-    command = [
-      "ffmpeg", "-hide_banner", "-loglevel", "error", "-nostdin", "-y",
-      "-r", str(CAMERA_FPS * clip.speedup), "-f", "concat", "-safe", "0", "-protocol_whitelist", "file,pipe", "-c:v", "hevc",
-      "-i", "pipe:0", "-ss", str(start_time / clip.speedup), "-t", str(duration / clip.speedup),
-      "-map", "0:v:0", "-an", "-r", str(CAMERA_FPS), "-c:v", "libx264", "-preset", "veryfast",
-      "-b:v", f"{clip.bitrate}M", "-pix_fmt", "yuv420p", "-movflags", "+faststart+use_metadata_tags",
-      "-metadata", f"ai.comma.clip.settings={json.dumps(asdict(clip), separators=(',', ':'))}", output_path,
-    ]
+    inputs = list(inputs)
+    metadata = json.dumps(asdict(clip), separators=(',', ':'))
+    if PC:
+      command = [
+        "ffmpeg", "-hide_banner", "-loglevel", "error", "-nostdin", "-y",
+        "-r", str(CAMERA_FPS * clip.speedup), "-f", "concat", "-safe", "0", "-protocol_whitelist", "file,pipe", "-c:v", "hevc",
+        "-i", "pipe:0", "-ss", str(start_time / clip.speedup), "-t", str(duration / clip.speedup),
+        "-map", "0:v:0", "-an", "-r", str(CAMERA_FPS), "-c:v", "libx264", "-preset", "veryfast",
+        "-b:v", f"{clip.bitrate}M", "-pix_fmt", "yuv420p", "-movflags", "+faststart+use_metadata_tags",
+        "-metadata", f"ai.comma.clip.settings={metadata}", output_path,
+      ]
+    else:
+      command = [os.path.join(BASEDIR, "openpilot/system/loggerd/encoderd"), "--clip", output_path,
+                 str(start_time), str(duration), "--bitrate", str(clip.bitrate * 1_000_000),
+                 "--speedup", str(clip.speedup), "--metadata", metadata, "--", *inputs]
 
     with self.lock:
       if self.clips.get(clip.filename) is not clip:
         return
-      process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, text=True)
+      process = subprocess.Popen(command, stdin=subprocess.PIPE if PC else subprocess.DEVNULL,
+                                 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, text=True)
       self.transcode_proc = (clip.filename, process)
     try:
-      if process.stdin is None:
-        raise RuntimeError("ffmpeg stdin is unavailable")
-      process.stdin.write("ffconcat version 1.0\n")
-      for path in inputs:
-        escaped_path = path.replace("'", "'\\''")
-        process.stdin.write(f"file 'file:{escaped_path}'\noption framerate {CAMERA_FPS}\nduration {SEGMENT_LENGTH}\n")
-      process.stdin.close()
+      if PC:
+        if process.stdin is None:
+          raise RuntimeError("ffmpeg stdin is unavailable")
+        process.stdin.write("ffconcat version 1.0\n")
+        for path in inputs:
+          escaped_path = path.replace("'", "'\\''")
+          process.stdin.write(f"file 'file:{escaped_path}'\noption framerate {CAMERA_FPS}\nduration {SEGMENT_LENGTH}\n")
+        process.stdin.close()
       process.wait()
       if process.returncode != 0:
-        raise RuntimeError(f"ffmpeg exited with code {process.returncode}")
+        raise RuntimeError(f"clip encoder exited with code {process.returncode}")
     finally:
       with suppress(OSError):
         if process.stdin is not None:

openpilot/system/athena/tests/test_athenad.py

@@ -103,6 +103,14 @@ class TestAthenadMethods(OpenpilotTestCase):
       f.write(data)
     return fn
 
+  @staticmethod
+  def _video_clips(clip):
+    clips = object.__new__(athenad.VideoClips)
+    clips.lock = threading.Condition()
+    clips.clips = {clip.filename: clip}
+    clips.transcode_proc = None
+    return clips
+
 
   # *** test cases ***
 
@@ -173,6 +181,56 @@ class TestAthenadMethods(OpenpilotTestCase):
     assert resp, 'list empty!'
     assert len(resp) == len(expected)
 
+  def test_video_clip_hardware_encoder(self, mocker):
+    clip = athenad.VideoClips.Clip("route", "fcamera.hevc", 10, 130, 2, 4, "clip.mp4", 123)
+    clips = self._video_clips(clip)
+    process = mocker.Mock(stdin=None, returncode=0)
+    process.poll.return_value = 0
+    popen = mocker.patch("openpilot.system.athena.athenad.subprocess.Popen", return_value=process)
+    mocker.patch.object(athenad, "PC", False)
+
+    clips._encode(clip, ["segment0", "segment1"], "output.mp4", 10, 120)
+
+    metadata = json.dumps(asdict(clip), separators=(',', ':'))
+    assert popen.call_args.args[0] == [
+      os.path.join(athenad.BASEDIR, "openpilot/system/loggerd/encoderd"), "--clip", "output.mp4", "10", "120",
+      "--bitrate", "2000000", "--speedup", "4", "--metadata", metadata, "--", "segment0", "segment1",
+    ]
+    assert popen.call_args.kwargs["stdin"] == athenad.subprocess.DEVNULL
+    assert clips.transcode_proc is None
+
+  def test_video_clip_hardware_encoder_failure(self, mocker):
+    clip = athenad.VideoClips.Clip("route", "fcamera.hevc", 0, 60, 1, 1, "clip.mp4", 123)
+    clips = self._video_clips(clip)
+    process = mocker.Mock(stdin=None, returncode=1)
+    process.poll.return_value = 1
+    mocker.patch("openpilot.system.athena.athenad.subprocess.Popen", return_value=process)
+    mocker.patch.object(athenad, "PC", False)
+
+    with self.assertRaisesRegex(RuntimeError, "clip encoder exited with code 1"):
+      clips._encode(clip, ["segment"], "output.mp4", 0, 60)
+    assert clips.transcode_proc is None
+
+  def test_video_clip_software_fallback(self, mocker):
+    clip = athenad.VideoClips.Clip("route", "fcamera.hevc", 10, 30, 3, 2, "clip.mp4", 123)
+    clips = self._video_clips(clip)
+    stdin = mocker.Mock()
+    process = mocker.Mock(stdin=stdin, returncode=0)
+    process.poll.return_value = 0
+    popen = mocker.patch("openpilot.system.athena.athenad.subprocess.Popen", return_value=process)
+    mocker.patch.object(athenad, "PC", True)
+
+    clips._encode(clip, ["segment'0", "segment1"], "output.mp4", 10, 20)
+
+    command = popen.call_args.args[0]
+    assert ["-r", "40"] == command[command.index("-r"):command.index("-r") + 2]
+    assert ["-ss", "5.0"] == command[command.index("-ss"):command.index("-ss") + 2]
+    assert ["-t", "10.0"] == command[command.index("-t"):command.index("-t") + 2]
+    assert ["-b:v", "3M"] == command[command.index("-b:v"):command.index("-b:v") + 2]
+    writes = [call.args[0] for call in stdin.write.call_args_list]
+    assert "file 'file:segment'\\''0'\n" in writes[1]
+    assert writes[-1].startswith("file 'file:segment1'")
+
   def test_strip_extension(self):
     # any requested log file with an invalid extension won't return as existing
     fn = self._create_file('qlog.bz2')

openpilot/system/loggerd/SConscript

@@ -5,7 +5,7 @@ frameworks = []
 
 src = ['logger.cc', 'zstd_writer.cc', 'video_writer.cc', 'encoder/encoder.cc', 'encoder/jpeg_encoder.cc']
 if arch == "larch64":
-  src += ['encoder/v4l_encoder.cc']
+  src += ['clip_encoder.cc', 'encoder/v4l_encoder.cc', 'encoder/v4l_decoder.cc']
 else:
   src += ['encoder/ffmpeg_encoder.cc']
   if arch == "Darwin":

openpilot/system/loggerd/clip_encoder.cc

@@ -0,0 +1,274 @@
+#include "system/loggerd/clip_encoder.h"
+
+#include <algorithm>
+#include <array>
+#include <cmath>
+#include <exception>
+#include <filesystem>
+#include <thread>
+#include <unistd.h>
+#include <utility>
+#include <vector>
+
+extern "C" {
+#include <libavcodec/avcodec.h>
+#include <libavformat/avformat.h>
+}
+
+#include "common/swaglog.h"
+#include "system/loggerd/encoder/v4l_decoder.h"
+#include "system/loggerd/encoder/v4l_encoder.h"
+#include "system/loggerd/loggerd.h"
+#include "system/loggerd/video_writer.h"
+
+namespace {
+
+constexpr double SEGMENT_DURATION = 60.0;
+constexpr int CLIP_FPS = 20;
+constexpr double PARALLEL_CLIP_MIN_DURATION = 2 * SEGMENT_DURATION;
+
+const EncoderInfo clip_encoder_info = {
+  .publish_name = "livestreamRoadEncodeData",
+  .record = false,
+  .fps = CLIP_FPS,
+  .get_settings = [](int) { return EncoderSettings::StreamEncoderSettings(); },
+  INIT_ENCODE_FUNCTIONS(LivestreamRoadEncode),
+};
+
+bool open_input(const std::string &path, AVFormatContext **ctx, int *stream_index) {
+  if (avformat_open_input(ctx, path.c_str(), nullptr, nullptr) < 0 ||
+      avformat_find_stream_info(*ctx, nullptr) < 0 ||
+      (*stream_index = av_find_best_stream(*ctx, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0)) < 0) {
+    LOGE("failed to open clip input %s", path.c_str());
+    avformat_close_input(ctx);
+    return false;
+  }
+  return true;
+}
+
+void remove_file(const std::string &path) {
+  std::error_code error;
+  std::filesystem::remove(path, error);
+}
+
+int encode_clip_worker(const std::vector<std::string> &inputs, int width, int height,
+                       double start_time, double duration, int bitrate, int speedup,
+                       int64_t frame_offset, int64_t *encoded_frames,
+                       V4LEncoder::PacketCallback packet_callback) try {
+  EncoderInfo encoder_info = clip_encoder_info;
+  encoder_info.get_settings = [bitrate](int) {
+    return EncoderSettings{.encode_type = cereal::EncodeIndex::Type::QCAMERA_H264,
+                           .bitrate = bitrate, .gop_size = 5};
+  };
+  V4LDecoder decoder;
+  V4LEncoder::Options options = {
+    .packet_callback = std::move(packet_callback),
+    .input_format = V4L2_PIX_FMT_NV12_UBWC,
+    .input_done_callback = [&decoder](VisionBuf *buf) { decoder.releaseFrame(buf); },
+    .max_performance = true,
+  };
+  V4LEncoder encoder(encoder_info, width, height, std::move(options));
+  encoder.encoder_open();
+  if (!decoder.init(V4LDecoder::DEVICE, width, height, V4L2_PIX_FMT_HEVC, true, V4L2_PIX_FMT_NV12_UBWC)) return 1;
+
+  const int64_t first_frame = std::floor(start_time * CLIP_FPS);
+  const int64_t end_frame = std::ceil((start_time + duration) * CLIP_FPS);
+  const int64_t source_frames = end_frame - first_frame;
+  const int64_t first_output_frame = (speedup - frame_offset % speedup) % speedup;
+  const int64_t expected_output_frames = first_output_frame < source_frames ?
+    1 + (source_frames - first_output_frame - 1) / speedup : 0;
+  int64_t input_frame = 0;
+  int64_t output_frame = 0;
+  int64_t received_frames = 0;
+  bool failed = false;
+  auto pump_decoder = [&](int timeout_ms) {
+    V4LDecodedFrame frame;
+    if (!decoder.pump(frame, timeout_ms)) return false;
+    if (!frame.buf) return true;
+    ++received_frames;
+    const int64_t source_frame = (int64_t)frame.token - 1;
+    if (source_frame < first_frame) {
+      decoder.releaseFrame(frame.buf);
+      return true;
+    }
+    if ((frame_offset + source_frame - first_frame) % speedup != 0) {
+      decoder.releaseFrame(frame.buf);
+      return true;
+    }
+
+    VisionIpcBufExtra extra = {};
+    extra.frame_id = output_frame;
+    extra.timestamp_sof = output_frame * 1000000000ULL / CLIP_FPS;
+    extra.timestamp_eof = extra.timestamp_sof;
+    if (encoder.encode_frame(frame.buf, &extra) < 0) {
+      decoder.releaseFrame(frame.buf);
+      return false;
+    }
+
+    ++output_frame;
+    return true;
+  };
+
+  for (const std::string &input : inputs) {
+    AVFormatContext *ctx = nullptr;
+    int stream_index = -1;
+    if (!open_input(input, &ctx, &stream_index)) { failed = true; break; }
+    AVPacket packet = {};
+    while (input_frame < end_frame && av_read_frame(ctx, &packet) >= 0) {
+      if (packet.stream_index != stream_index) {
+        av_packet_unref(&packet);
+        continue;
+      }
+      if (packet.size <= 0 || (size_t)packet.size > decoder.maxPacketSize()) {
+        LOGE("decoder packet too large: %d > %zu", packet.size, decoder.maxPacketSize());
+        av_packet_unref(&packet);
+        failed = true;
+        break;
+      }
+
+      // Keep several compressed packets in flight so the firmware can sustain
+      // decode/encode overlap and does not downclock due to a shallow queue.
+      while (!decoder.queuePacket(&packet, input_frame + 1)) {
+        if (!pump_decoder(-1)) {
+          failed = true;
+          break;
+        }
+      }
+      av_packet_unref(&packet);
+      if (failed) break;
+
+      ++input_frame;
+    }
+    av_packet_unref(&packet);
+    avformat_close_input(&ctx);
+    if (failed || input_frame >= end_frame) break;
+  }
+
+  if (!failed) decoder.sendEOS();
+  for (int empty_polls = 0; !failed && received_frames < input_frame;) {
+    const int64_t before = received_frames;
+    failed = !pump_decoder(1000);
+    empty_polls = received_frames == before ? empty_polls + 1 : 0;
+    if (empty_polls == 5) failed = true;
+  }
+
+  encoder.encoder_close();
+  if (failed || input_frame < end_frame || output_frame != expected_output_frames) {
+    LOGE("clip failed: input=%lld/%lld decoded=%lld encoded=%lld/%lld",
+         (long long)input_frame, (long long)end_frame, (long long)received_frames,
+         (long long)output_frame, (long long)expected_output_frames);
+    return 1;
+  }
+  *encoded_frames = output_frame;
+  return 0;
+} catch (const std::exception &e) {
+  LOGE("clip worker failed: %s", e.what());
+  return 1;
+}
+
+struct SpoolPacket {
+  uint32_t size;
+  int64_t timestamp;
+  bool keyframe;
+};
+
+}  // namespace
+
+int encode_clip(const std::vector<std::string> &inputs, const std::string &output,
+                double start_time, double duration, int bitrate, int speedup,
+                const std::string &metadata) {
+  if (inputs.empty() || !std::isfinite(start_time) || !std::isfinite(duration) ||
+      start_time < 0 || duration <= 0 || bitrate <= 0 || speedup <= 0) {
+    return 1;
+  }
+
+  // Inputs are consecutive loggerd segments. Skip whole files before the clip
+  // so a late start does not spend hardware time decoding discarded minutes.
+  const double available_duration = inputs.size() * SEGMENT_DURATION;
+  if (start_time >= available_duration || duration > available_duration - start_time) return 1;
+  const size_t skipped_segments = start_time / SEGMENT_DURATION;
+  const std::vector<std::string> clip_inputs(inputs.begin() + skipped_segments, inputs.end());
+  const double local_start = start_time - skipped_segments * SEGMENT_DURATION;
+
+  AVFormatContext *ctx = nullptr;
+  int stream = -1;
+  if (!open_input(clip_inputs.front(), &ctx, &stream)) return 1;
+  AVCodecParameters *codec = ctx->streams[stream]->codecpar;
+  const int width = codec->width, height = codec->height;
+  const bool valid_codec = codec->codec_id == AV_CODEC_ID_HEVC && width > 0 && height > 0;
+  avformat_close_input(&ctx);
+  if (!valid_codec) return 1;
+
+  std::filesystem::path output_path(output);
+  const std::string output_dir = output_path.has_parent_path() ? output_path.parent_path() : ".";
+  VideoWriter writer(output_dir.c_str(), output_path.filename().c_str(), true,
+                     width, height, CLIP_FPS, cereal::EncodeIndex::Type::QCAMERA_H264);
+  if (!metadata.empty()) writer.set_metadata("ai.comma.clip.settings", metadata.c_str());
+  V4LEncoder::PacketCallback write_packet = [&writer](uint8_t *data, size_t size, int64_t timestamp,
+                                                      bool config, bool keyframe) {
+    writer.write(data, size, timestamp, config, keyframe);
+  };
+
+  if (clip_inputs.size() < 2 || duration < PARALLEL_CLIP_MIN_DURATION) {
+    int64_t encoded_frames = 0;
+    const bool success = encode_clip_worker(clip_inputs, width, height, local_start, duration,
+                                            bitrate, speedup, 0, &encoded_frames, write_packet) == 0;
+    if (!success) remove_file(output);
+    return success ? 0 : 1;
+  }
+
+  const size_t split = std::clamp<size_t>(std::llround((local_start + duration / 2) / SEGMENT_DURATION),
+                                           1, clip_inputs.size() - 1);
+  const double split_time = split * SEGMENT_DURATION;
+  const std::array<std::vector<std::string>, 2> shard_inputs = {
+    std::vector<std::string>(clip_inputs.begin(), clip_inputs.begin() + split),
+    std::vector<std::string>(clip_inputs.begin() + split, clip_inputs.end()),
+  };
+  const std::array<double, 2> shard_starts = {local_start, 0};
+  const std::array<double, 2> shard_durations = {
+    split_time - local_start, local_start + duration - split_time,
+  };
+  const std::string spool_path = output + ".encoderd-" + std::to_string(getpid()) + ".tmp";
+  FILE *spool = fopen(spool_path.c_str(), "w+b");
+  if (!spool) {
+    remove_file(output);
+    return 1;
+  }
+  remove_file(spool_path);
+  bool spool_ok = true;
+  V4LEncoder::PacketCallback spool_packet = [&](uint8_t *data, size_t size, int64_t timestamp,
+                                                bool config, bool keyframe) {
+    if (config) return;
+    const SpoolPacket packet = {(uint32_t)size, timestamp, keyframe};
+    spool_ok &= fwrite(&packet, sizeof(packet), 1, spool) == 1 && fwrite(data, 1, size, spool) == size;
+  };
+  std::array<int, 2> results = {1, 1};
+  std::array<int64_t, 2> encoded_frames = {};
+  const std::array<int64_t, 2> frame_offsets = {
+    0, (int64_t)std::llround(split_time * CLIP_FPS) - (int64_t)std::floor(local_start * CLIP_FPS),
+  };
+  std::array<std::thread, 2> workers;
+
+  for (size_t i = 0; i < workers.size(); ++i) {
+    workers[i] = std::thread([&, i]() {
+      results[i] = encode_clip_worker(shard_inputs[i], width, height, shard_starts[i], shard_durations[i],
+                                      bitrate, speedup, frame_offsets[i], &encoded_frames[i],
+                                      i == 0 ? write_packet : spool_packet);
+    });
+  }
+  for (std::thread &worker : workers) worker.join();
+
+  rewind(spool);
+  SpoolPacket packet;
+  std::vector<uint8_t> data;
+  const int64_t timestamp_offset = encoded_frames[0] * 1000000 / CLIP_FPS;
+  while (spool_ok && fread(&packet, sizeof(packet), 1, spool) == 1) {
+    data.resize(packet.size);
+    spool_ok = fread(data.data(), 1, data.size(), spool) == data.size();
+    if (spool_ok) writer.write(data.data(), data.size(), packet.timestamp + timestamp_offset, false, packet.keyframe);
+  }
+  fclose(spool);
+  bool success = results[0] == 0 && results[1] == 0 && spool_ok;
+  if (!success) remove_file(output);
+  return success ? 0 : 1;
+}

openpilot/system/loggerd/clip_encoder.h

@@ -0,0 +1,10 @@
+#pragma once
+
+#include <string>
+#include <vector>
+
+// inputs are consecutive 60-second loggerd HEVC segments; start_time is
+// relative to the beginning of the first input.
+int encode_clip(const std::vector<std::string> &inputs, const std::string &output,
+                double start_time, double duration, int bitrate = 5'000'000,
+                int speedup = 1, const std::string &metadata = {});

openpilot/system/loggerd/encoder/v4l_decoder.cc

@@ -1,13 +1,19 @@
 #include "system/loggerd/encoder/v4l_decoder.h"
 
 #include <assert.h>
+#include <cerrno>
+#include <climits>
 #include <linux/v4l2-controls.h>
 #include <linux/videodev2.h>
+#include <sys/ioctl.h>
+#include <unistd.h>
 
 
 #include "common/swaglog.h"
 #include "common/util.h"
 
+constexpr int OFFLINE_CORE_PLACEMENT_RATE = 80 << 16;
+
 // echo "0xFFFF" > /sys/kernel/debug/msm_vidc/debug_level
 
 static void copyBuffer(VisionBuf *src_buf, VisionBuf *dst_buf) {
@@ -32,99 +38,122 @@ V4LDecoder::~V4LDecoder() {
   }
 }
 
-bool V4LDecoder::init(const char* dev, size_t width, size_t height, uint64_t codec) {
+bool V4LDecoder::init(const char* dev, size_t width, size_t height, uint64_t codec,
+                   bool direct_mode, uint32_t capture_fourcc) {
   LOG("Initializing msm_vidc device %s", dev);
   this->w = width;
   this->h = height;
-  this->fd = open(dev, O_RDWR, 0);
+  this->direct = direct_mode;
+  this->capture_format = capture_fourcc;
+  this->fd = open(dev, O_RDWR | O_NONBLOCK, 0);
   if (fd < 0) {
     LOGE("failed to open video device %s", dev);
     return false;
   }
   subscribeEvents();
   v4l2_buf_type out_type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;
-  setPlaneFormat(out_type, V4L2_PIX_FMT_HEVC); // Also allocates the output buffer
+  setPlaneFormat(out_type, codec); // Also allocates the output buffers
   setFPS(FPS);
+  if (direct) {
+    struct v4l2_control ctrls[] = {
+      // A finite real-time load lets the driver place decode and encode on separate cores.
+      { .id = V4L2_CID_MPEG_VIDC_VIDEO_OPERATING_RATE, .value = OFFLINE_CORE_PLACEMENT_RATE },
+      { .id = V4L2_CID_MPEG_VIDC_VIDEO_PRIORITY, .value = V4L2_MPEG_VIDC_VIDEO_PRIORITY_REALTIME_ENABLE },
+    };
+    for (auto ctrl : ctrls) {
+      util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL offline decode failed");
+    }
+  }
   request_buffers(fd, out_type, OUTPUT_BUFFER_COUNT);
   util::safe_ioctl(fd, VIDIOC_STREAMON, &out_type, "VIDIOC_STREAMON OUTPUT failed");
   restartCapture();
-  setupPolling();
+  pfd = {fd, POLLIN | POLLOUT | POLLWRNORM | POLLRDNORM | POLLPRI, 0};
 
   this->initialized = true;
   return true;
 }
 
 VisionBuf* V4LDecoder::decodeFrame(AVPacket *pkt, VisionBuf *buf) {
-  assert(initialized && (pkt != nullptr) && (buf != nullptr));
+  assert(initialized && !direct && pkt != nullptr && buf != nullptr);
+  bool queued = false;
+  while (true) {
+    if (!queued) queued = queuePacket(pkt, 0);
+    V4LDecodedFrame frame;
+    if (!pump(frame, -1)) return nullptr;
+    if (!frame.buf) continue;
 
-  this->frame_ready = false;
-  this->current_output_buf = buf;
-  bool sent_packet = false;
+    VisionBuf *decoded = frame.buf;
+    copyBuffer(decoded, buf);
+    releaseFrame(decoded);
+    return buf;
+  }
+}
 
-  while (!this->frame_ready) {
-    if (!sent_packet) {
-      int buf_index = getBufferUnlocked();
-      if (buf_index >= 0) {
-        assert(buf_index < out_buf_cnt);
-        sendPacket(buf_index, pkt);
-        sent_packet = true;
-      }
-    }
+void V4LDecoder::releaseFrame(VisionBuf *buf) {
+  assert(buf >= cap_bufs && buf < cap_bufs + CAPTURE_BUFFER_COUNT);
+  queueCaptureBuffer(buf - cap_bufs);
+}
 
-    if (poll(pfd, nfds, -1) < 0) {
+bool V4LDecoder::queuePacket(const AVPacket *pkt, uint64_t token) {
+  int buf_index = getBufferUnlocked();
+  return buf_index >= 0 && sendPacket(buf_index, pkt, token);
+}
+
+bool V4LDecoder::pump(V4LDecodedFrame &frame, int timeout_ms) {
+  frame = {};
+  int rc;
+  while (true) {
+    rc = poll(&pfd, 1, timeout_ms);
+    if (rc < 0) {
+      if (errno == EINTR) continue;
       LOGE("poll() error: %d", errno);
-      return nullptr;
-    }
-
-    if (VisionBuf* result = processEvents()) {
-      return result;
+      return false;
     }
+    break;
   }
 
-  return buf;
+  if (rc == 0) return true;
+
+  int result;
+
+  // Port changes must be handled before capture DQ so no old-format surface is
+  // handed to a client after the driver has requested a capture flush.
+  while ((result = handleEvent()) > 0) {}
+  if (result < 0) return false;
+
+  while ((result = handleOutput()) > 0) {}
+  if (result < 0) return false;
+
+  result = handleCapture(&frame);
+  return result >= 0;
 }
 
-VisionBuf* V4LDecoder::processEvents() {
-  for (int idx = 0; idx < nfds; idx++) {
-    short revents = pfd[idx].revents;
-    if (!revents) continue;
-
-    if (idx == ev[EV_VIDEO]) {
-      if (revents & (POLLIN | POLLRDNORM)) {
-        VisionBuf *result = handleCapture();
-        if (result == this->current_output_buf) {
-          this->frame_ready = true;
-        }
-      }
-      if (revents & (POLLOUT | POLLWRNORM)) {
-        handleOutput();
-      }
-      if (revents & POLLPRI) {
-        handleEvent();
-      }
-    } else {
-      LOGE("Unexpected event on fd %d", pfd[idx].fd);
-    }
-  }
-  return nullptr;
-}
-
-VisionBuf* V4LDecoder::handleCapture() {
+int V4LDecoder::handleCapture(V4LDecodedFrame *frame) {
   struct v4l2_buffer buf = {0};
   struct v4l2_plane planes[1] = {0};
   buf.type          = V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE;
   buf.memory        = V4L2_MEMORY_USERPTR;
   buf.m.planes      = planes;
   buf.length        = 1;
-  util::safe_ioctl(this->fd, VIDIOC_DQBUF, &buf, "VIDIOC_DQBUF CAPTURE failed");
-
-  if (this->reconfigure_pending || buf.m.planes[0].bytesused == 0) {
-    return nullptr;
+  int err = HANDLE_EINTR(ioctl(this->fd, VIDIOC_DQBUF, &buf));
+  if (err < 0 && errno == EAGAIN) return 0;
+  if (err < 0) {
+    LOGE("VIDIOC_DQBUF CAPTURE failed: %d", errno);
+    return -1;
   }
 
-  copyBuffer(&cap_bufs[buf.index], this->current_output_buf);
-  queueCaptureBuffer(buf.index);
-  return this->current_output_buf;
+  const bool has_payload = buf.m.planes[0].bytesused != 0;
+  const bool eos = (buf.flags & V4L2_QCOM_BUF_FLAG_EOS) != 0;
+
+  frame->buf = nullptr;
+  if (!reconfigure_pending && has_payload) {
+    frame->buf = &cap_bufs[buf.index];
+    frame->token = (uint64_t)buf.timestamp.tv_sec * 1000000ULL + buf.timestamp.tv_usec;
+  } else if (!reconfigure_pending && !eos) {
+    queueCaptureBuffer(buf.index);
+  }
+
+  return 1;
 }
 
 bool V4LDecoder::subscribeEvents() {
@@ -146,20 +175,17 @@ bool V4LDecoder::setPlaneFormat(enum v4l2_buf_type type, uint32_t fourcc) {
   util::safe_ioctl(fd, VIDIOC_S_FMT, &fmt, "VIDIOC_S_FMT failed");
   if (type == V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE) {
     this->out_buf_size = pix->plane_fmt[0].sizeimage;
-    int ion_size = this->out_buf_size * OUTPUT_BUFFER_COUNT; // Output (input) buffers are ION buffer.
-    this->out_buf.allocate(ion_size); // mmap rw
     for (int i = 0; i < OUTPUT_BUFFER_COUNT; i++) {
-      this->out_buf_off[i] = i * this->out_buf_size;
-      this->out_buf_addr[i] = (char *)this->out_buf.addr + this->out_buf_off[i];
+      this->out_bufs[i].allocate(this->out_buf_size);
       this->out_buf_flag[i] = false;
     }
-    LOGD("Set output buffer size to %d, count %d, addr %p", this->out_buf_size, OUTPUT_BUFFER_COUNT, this->out_buf.addr);
+    LOGD("Set output buffer size to %d, count %d, addr %p", this->out_buf_size, OUTPUT_BUFFER_COUNT, this->out_bufs[0].addr);
   } else if (type == V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE) {
     request_buffers(this->fd, type, CAPTURE_BUFFER_COUNT);
     util::safe_ioctl(fd, VIDIOC_G_FMT, &fmt, "VIDIOC_G_FMT failed");
     const __u32 y_size    = pix->plane_fmt[0].sizeimage;
     const __u32 y_stride  = pix->plane_fmt[0].bytesperline;
-    for (int i = 0; i < CAPTURE_BUFFER_COUNT; i++) {
+    for (size_t i = 0; i < CAPTURE_BUFFER_COUNT; i++) {
       size_t uv_offset = (size_t)y_stride * pix->height;
       size_t required = uv_offset + (y_stride * pix->height / 2); // enough for Y + UV. For linear NV12, UV plane starts at y_stride * height.
       size_t alloc_size = std::max<size_t>(y_size, required);
@@ -191,18 +217,31 @@ bool V4LDecoder::restartCapture() {
     util::safe_ioctl(this->fd, VIDIOC_REQBUFS, &reqbuf, "VIDIOC_REQBUFS failed");
     for (size_t i = 0; i < CAPTURE_BUFFER_COUNT; ++i) {
       this->cap_bufs[i].free();
-      this->cap_buf_flag[i] = false; // mark as not queued
       cap_bufs[i].~VisionBuf();
       new (&cap_bufs[i]) VisionBuf();
     }
   }
   // setup, start and queue capture buffers
   setDBP();
-  setPlaneFormat(type, V4L2_PIX_FMT_NV12);
+  setPlaneFormat(type, capture_format);
+  if (direct) {
+    struct v4l2_control ctrl = {
+      .id = V4L2_CID_MPEG_VIDC_VIDEO_OPERATING_RATE,
+      .value = OFFLINE_CORE_PLACEMENT_RATE,
+    };
+    util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL placement decode failed");
+  }
   util::safe_ioctl(this->fd, VIDIOC_STREAMON, &type, "VIDIOC_STREAMON CAPTURE failed");
   for (size_t i = 0; i < CAPTURE_BUFFER_COUNT; ++i) {
     queueCaptureBuffer(i);
   }
+  if (direct) {
+    struct v4l2_control ctrl = {
+      .id = V4L2_CID_MPEG_VIDC_VIDEO_OPERATING_RATE,
+      .value = INT_MAX,
+    };
+    util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL turbo decode failed");
+  }
 
   return true;
 }
@@ -224,27 +263,28 @@ bool V4LDecoder::queueCaptureBuffer(int i) {
   planes[0].bytesused     = this->cap_bufs[i].len;
   planes[0].data_offset   = 0;
   util::safe_ioctl(this->fd, VIDIOC_QBUF, &buf, "VIDIOC_QBUF failed");
-  this->cap_buf_flag[i] = true; // mark as queued
   return true;
 }
 
-bool V4LDecoder::queueOutputBuffer(int i, size_t size) {
+bool V4LDecoder::queueOutputBuffer(int i, size_t size, uint64_t token) {
   struct v4l2_buffer buf = {0};
   struct v4l2_plane planes[1] = {0};
 
   buf.type                = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;
   buf.memory              = V4L2_MEMORY_USERPTR;
   buf.index               = i;
+  buf.flags               = V4L2_BUF_FLAG_TIMESTAMP_COPY;
+  buf.timestamp.tv_sec    = token / 1000000ULL;
+  buf.timestamp.tv_usec   = token % 1000000ULL;
   buf.m.planes            = planes;
   buf.length              = 1;
   // decoded frame plane
-  planes[0].m.userptr     = (unsigned long)this->out_buf_off[i]; // check this
+  planes[0].m.userptr     = (unsigned long)this->out_bufs[i].addr;
   planes[0].length        = this->out_buf_size;
-  planes[0].reserved[0]   = this->out_buf.fd; // ION fd
+  planes[0].reserved[0]   = this->out_bufs[i].fd; // ION fd
   planes[0].reserved[1]   = 0;
   planes[0].bytesused     = size;
   planes[0].data_offset   = 0;
-  assert((this->out_buf_off[i] & 0xfff) == 0);          // must be 4 KiB aligned
   assert(this->out_buf_size % 4096 == 0);               // ditto for size
 
   util::safe_ioctl(this->fd, VIDIOC_QBUF, &buf, "VIDIOC_QBUF failed");
@@ -266,27 +306,19 @@ bool V4LDecoder::setDBP() {
   return true;
 }
 
-bool V4LDecoder::setupPolling() {
-  // Initialize poll array
-  pfd[EV_VIDEO] = {fd, POLLIN | POLLOUT | POLLWRNORM | POLLRDNORM | POLLPRI, 0};
-  ev[EV_VIDEO] = EV_VIDEO;
-  nfds = 1;
-  return true;
-}
-
-bool V4LDecoder::sendPacket(int buf_index, AVPacket *pkt) {
-  assert(buf_index >= 0 && buf_index < out_buf_cnt);
+bool V4LDecoder::sendPacket(int buf_index, const AVPacket *pkt, uint64_t token) {
+  assert(buf_index >= 0 && buf_index < OUTPUT_BUFFER_COUNT);
   assert(pkt != nullptr && pkt->data != nullptr && pkt->size > 0);
+  assert((size_t)pkt->size <= (size_t)this->out_buf_size);
   // Prepare output buffer
-  memset(this->out_buf_addr[buf_index], 0, this->out_buf_size);
-  uint8_t * data = (uint8_t *)this->out_buf_addr[buf_index];
+  uint8_t * data = (uint8_t *)this->out_bufs[buf_index].addr;
   memcpy(data, pkt->data, pkt->size);
-  queueOutputBuffer(buf_index, pkt->size);
+  queueOutputBuffer(buf_index, pkt->size, token);
   return true;
 }
 
 int V4LDecoder::getBufferUnlocked() {
-  for (int i = 0; i < this->out_buf_cnt; i++) {
+  for (int i = 0; i < OUTPUT_BUFFER_COUNT; i++) {
     if (!out_buf_flag[i]) {
       return i;
     }
@@ -295,22 +327,32 @@ int V4LDecoder::getBufferUnlocked() {
 }
 
 
-bool V4LDecoder::handleOutput() {
+int V4LDecoder::handleOutput() {
   struct v4l2_buffer buf = {0};
   struct v4l2_plane planes[1];
   buf.type      = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;
   buf.memory    = V4L2_MEMORY_USERPTR;
   buf.m.planes  = planes;
   buf.length    = 1;
-  util::safe_ioctl(this->fd, VIDIOC_DQBUF, &buf, "VIDIOC_DQBUF OUTPUT failed");
+  int err = HANDLE_EINTR(ioctl(this->fd, VIDIOC_DQBUF, &buf));
+  if (err < 0 && errno == EAGAIN) return 0;
+  if (err < 0) {
+    LOGE("VIDIOC_DQBUF OUTPUT failed: %d", errno);
+    return -1;
+  }
   this->out_buf_flag[buf.index] = false; // mark as not queued
-  return true;
+  return 1;
 }
 
-bool V4LDecoder::handleEvent() {
+int V4LDecoder::handleEvent() {
   // dequeue event
   struct v4l2_event event = {0};
-  util::safe_ioctl(this->fd, VIDIOC_DQEVENT, &event, "VIDIOC_DQEVENT failed");
+  int err = HANDLE_EINTR(ioctl(this->fd, VIDIOC_DQEVENT, &event));
+  if (err < 0 && (errno == EAGAIN || errno == ENOENT)) return 0;
+  if (err < 0) {
+    LOGE("VIDIOC_DQEVENT failed: %d", errno);
+    return -1;
+  }
   switch (event.type) {
     case V4L2_EVENT_MSM_VIDC_PORT_SETTINGS_CHANGED_INSUFFICIENT: {
       unsigned int *ptr     = (unsigned int *)event.u.data;
@@ -342,5 +384,10 @@ bool V4LDecoder::handleEvent() {
     default:
       break;
   }
-  return true;
+  return 1;
+}
+
+void V4LDecoder::sendEOS() {
+  struct v4l2_decoder_cmd command = { .cmd = V4L2_DEC_CMD_STOP };
+  util::safe_ioctl(fd, VIDIOC_DECODER_CMD, &command, "VIDIOC_DECODER_CMD STOP failed");
 }

openpilot/system/loggerd/encoder/v4l_decoder.h

@@ -13,16 +13,40 @@ extern "C" {
 #define V4L2_EVENT_MSM_VIDC_START (V4L2_EVENT_PRIVATE_START + 0x00001000)
 #define V4L2_EVENT_MSM_VIDC_FLUSH_DONE (V4L2_EVENT_MSM_VIDC_START + 1)
 #define V4L2_EVENT_MSM_VIDC_PORT_SETTINGS_CHANGED_INSUFFICIENT (V4L2_EVENT_MSM_VIDC_START + 3)
+#ifndef V4L2_CID_MPEG_MSM_VIDC_BASE
 #define V4L2_CID_MPEG_MSM_VIDC_BASE 0x00992000
+#endif
+#ifndef V4L2_CID_MPEG_VIDC_VIDEO_DPB_COLOR_FORMAT
 #define V4L2_CID_MPEG_VIDC_VIDEO_DPB_COLOR_FORMAT (V4L2_CID_MPEG_MSM_VIDC_BASE + 44)
+#endif
+#ifndef V4L2_CID_MPEG_VIDC_VIDEO_STREAM_OUTPUT_MODE
 #define V4L2_CID_MPEG_VIDC_VIDEO_STREAM_OUTPUT_MODE (V4L2_CID_MPEG_MSM_VIDC_BASE + 22)
+#endif
+#ifndef V4L2_PIX_FMT_NV12_UBWC
+#define V4L2_PIX_FMT_NV12_UBWC v4l2_fourcc('Q', '1', '2', '8')
+#endif
+#ifndef V4L2_CID_MPEG_VIDC_VIDEO_PRIORITY
+#define V4L2_CID_MPEG_VIDC_VIDEO_PRIORITY (V4L2_CID_MPEG_MSM_VIDC_BASE + 52)
+#define V4L2_MPEG_VIDC_VIDEO_PRIORITY_REALTIME_ENABLE 0
+#define V4L2_MPEG_VIDC_VIDEO_PRIORITY_REALTIME_DISABLE 1
+#endif
+#ifndef V4L2_CID_MPEG_VIDC_VIDEO_OPERATING_RATE
+#define V4L2_CID_MPEG_VIDC_VIDEO_OPERATING_RATE (V4L2_CID_MPEG_MSM_VIDC_BASE + 53)
+#endif
 #define V4L2_QCOM_CMD_FLUSH_CAPTURE (1 << 1)
 #define V4L2_QCOM_CMD_FLUSH (4)
+#ifndef V4L2_QCOM_BUF_FLAG_EOS
+#define V4L2_QCOM_BUF_FLAG_EOS 0x02000000
+#endif
 
 #define OUTPUT_BUFFER_COUNT 	8
-#define CAPTURE_BUFFER_COUNT 	8
+#define CAPTURE_BUFFER_COUNT 	16
 #define FPS 									20
 
+struct V4LDecodedFrame {
+  VisionBuf *buf = nullptr;
+  uint64_t token = 0;
+};
 
 class V4LDecoder {
 public:
@@ -31,8 +55,16 @@ public:
   V4LDecoder() = default;
   ~V4LDecoder();
 
-  bool init(const char* dev, size_t width, size_t height, uint64_t codec);
+  bool init(const char* dev, size_t width, size_t height, uint64_t codec,
+            bool direct_mode = false, uint32_t capture_fourcc = V4L2_PIX_FMT_NV12);
   VisionBuf* decodeFrame(AVPacket* pkt, VisionBuf* buf);
+  // queuePacket() and pump() are single-threaded. releaseFrame() may be called
+  // from a consumer thread after a direct capture surface is no longer needed.
+  bool queuePacket(const AVPacket *pkt, uint64_t token);
+  bool pump(V4LDecodedFrame &frame, int timeout_ms);
+  void releaseFrame(VisionBuf *buf);
+  void sendEOS();
+  size_t maxPacketSize() const { return out_buf_size; }
 
   AVFormatContext* avctx = nullptr;
   int fd = 0;
@@ -40,50 +72,34 @@ public:
 private:
   bool initialized = false;
   bool reconfigure_pending = false;
-  bool frame_ready = false;
+  bool direct = false;
+  uint32_t capture_format = V4L2_PIX_FMT_NV12;
 
-  VisionBuf* current_output_buf = nullptr;
-  VisionBuf out_buf;                          // Single input buffer
+  VisionBuf out_bufs[OUTPUT_BUFFER_COUNT];    // Distinct dma-buf per in-flight packet
   VisionBuf cap_bufs[CAPTURE_BUFFER_COUNT];   // Capture (output) buffers
 
-  size_t w = 1928, h = 1208;
-  size_t cap_height = 0, cap_width = 0;
-
-  int cap_buf_size = 0;
+  size_t w = 0, h = 0;
   int out_buf_size = 0;
 
-  size_t cap_plane_off[CAPTURE_BUFFER_COUNT] = {0};
-  size_t cap_plane_stride[CAPTURE_BUFFER_COUNT] = {0};
-  bool cap_buf_flag[CAPTURE_BUFFER_COUNT] = {false};
-
-  size_t out_buf_off[OUTPUT_BUFFER_COUNT] = {0};
-  void* out_buf_addr[OUTPUT_BUFFER_COUNT] = {0};
   bool out_buf_flag[OUTPUT_BUFFER_COUNT] = {false};
-  const int out_buf_cnt = OUTPUT_BUFFER_COUNT;
 
   const int subscriptions[2] = {
     V4L2_EVENT_MSM_VIDC_FLUSH_DONE,
     V4L2_EVENT_MSM_VIDC_PORT_SETTINGS_CHANGED_INSUFFICIENT
   };
 
-  enum { EV_VIDEO, EV_COUNT };
-  struct pollfd pfd[EV_COUNT] = {0};
-  int ev[EV_COUNT] = {-1};
-  int nfds = 0;
+  struct pollfd pfd = {};
 
-  VisionBuf* processEvents();
-  bool setupOutput();
   bool subscribeEvents();
   bool setPlaneFormat(v4l2_buf_type type, uint32_t fourcc);
   bool setFPS(uint32_t fps);
   bool restartCapture();
   bool queueCaptureBuffer(int i);
-  bool queueOutputBuffer(int i, size_t size);
+  bool queueOutputBuffer(int i, size_t size, uint64_t token);
   bool setDBP();
-  bool setupPolling();
-  bool sendPacket(int buf_index, AVPacket* pkt);
+  bool sendPacket(int buf_index, const AVPacket* pkt, uint64_t token);
   int getBufferUnlocked();
-  VisionBuf* handleCapture();
-  bool handleOutput();
-  bool handleEvent();
+  int handleCapture(V4LDecodedFrame *frame);
+  int handleOutput();
+  int handleEvent();
 };

openpilot/system/loggerd/encoder/v4l_encoder.cc

@@ -2,6 +2,7 @@
 #include <string>
 #include <sys/ioctl.h>
 #include <poll.h>
+#include <utility>
 
 #include "system/loggerd/encoder/v4l_encoder.h"
 #include "common/util.h"
@@ -119,12 +120,17 @@ void V4LEncoder::dequeue_handler(V4LEncoder *e) {
       } else if (flags & V4L2_QCOM_BUF_FLAG_CODECCONFIG) {
         // save header
         header = kj::heapArray<capnp::byte>(buf, bytesused);
+        if (e->packet_callback) e->packet_callback(header.begin(), header.size(), ts, true, false);
       } else {
         VisionIpcBufExtra extra = e->extras.pop();
         assert(extra.timestamp_eof/1000 == ts); // stay in sync
         frame_id = extra.frame_id;
         ++idx;
-        e->publisher_publish(e->segment_num, idx, extra, flags, header, kj::arrayPtr<capnp::byte>(buf, bytesused));
+        if (e->packet_callback) {
+          e->packet_callback(buf, bytesused, ts, false, flags & V4L2_BUF_FLAG_KEYFRAME);
+        } else {
+          e->publisher_publish(e->segment_num, idx, extra, flags, header, kj::arrayPtr<capnp::byte>(buf, bytesused));
+        }
       }
 
       if (env_debug_encoder) {
@@ -139,13 +145,19 @@ void V4LEncoder::dequeue_handler(V4LEncoder *e) {
     if (pfd.revents & (POLLOUT | POLLWRNORM)) {
       unsigned int index;
       dequeue_buffer(e->fd, V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE, &index);
+      VisionBuf *input_buf = e->input_bufs[index].exchange(nullptr);
+      if (input_buf && e->input_done_callback) e->input_done_callback(input_buf);
       e->free_buf_in.push(index);
     }
   }
 }
 
 V4LEncoder::V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_height)
-    : VideoEncoder(encoder_info, in_width, in_height) {
+    : V4LEncoder(encoder_info, in_width, in_height, Options{}) {}
+
+V4LEncoder::V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_height, Options options)
+    : VideoEncoder(encoder_info, in_width, in_height), packet_callback(std::move(options.packet_callback)),
+      input_done_callback(std::move(options.input_done_callback)) {
   fd = HANDLE_EINTR(open("/dev/v4l/by-path/platform-aa00000.qcom_vidc-video-index1", O_RDWR|O_NONBLOCK));
   assert(fd >= 0);
 
@@ -194,7 +206,7 @@ V4LEncoder::V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_hei
       .pix_mp = {
         .width = (unsigned int)in_width,
         .height = (unsigned int)in_height,
-        .pixelformat = V4L2_PIX_FMT_NV12,
+        .pixelformat = options.input_format,
         .field = V4L2_FIELD_ANY,
         .colorspace = V4L2_COLORSPACE_470_SYSTEM_BG,
       }
@@ -221,6 +233,13 @@ V4LEncoder::V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_hei
       util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL failed");
     }
   }
+  if (options.max_performance) {
+    struct v4l2_control ctrl = {
+      .id = V4L2_CID_MPEG_VIDC_VIDEO_PRIORITY,
+      .value = V4L2_MPEG_VIDC_VIDEO_PRIORITY_REALTIME_ENABLE,
+    };
+    util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL offline encode failed");
+  }
 
   if (is_h265) {
     struct v4l2_control ctrls[] = {
@@ -282,6 +301,7 @@ int V4LEncoder::encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra) {
 
   // reserve buffer
   int buffer_in = free_buf_in.pop();
+  input_bufs[buffer_in].store(buf);
 
   // push buffer
   extras.push(*extra);

openpilot/system/loggerd/encoder/v4l_encoder.h

@@ -1,14 +1,27 @@
 #pragma once
 
+#include <atomic>
+#include <functional>
+
 #include "common/queue.h"
 #include "system/loggerd/encoder/encoder.h"
 
-#define BUF_IN_COUNT 7
+#define BUF_IN_COUNT 9
 #define BUF_OUT_COUNT 6
 
 class V4LEncoder : public VideoEncoder {
 public:
+  using PacketCallback = std::function<void(uint8_t *, size_t, int64_t, bool, bool)>;
+  using InputDoneCallback = std::function<void(VisionBuf *)>;
+  struct Options {
+    PacketCallback packet_callback;
+    uint32_t input_format = V4L2_PIX_FMT_NV12;
+    InputDoneCallback input_done_callback;
+    bool max_performance = false;
+  };
+
   V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_height);
+  V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_height, Options options);
   ~V4LEncoder();
   int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra);
   void encoder_open();
@@ -23,12 +36,14 @@ private:
   int segment_num = -1;
   int counter = 0;
   int current_bitrate = -1;
-
   SafeQueue<VisionIpcBufExtra> extras;
+  PacketCallback packet_callback;
+  InputDoneCallback input_done_callback;
 
   static void dequeue_handler(V4LEncoder *e);
   std::thread dequeue_handler_thread;
 
   VisionBuf buf_out[BUF_OUT_COUNT];
+  std::atomic<VisionBuf *> input_bufs[BUF_IN_COUNT] = {};
   SafeQueue<unsigned int> free_buf_in;
 };

openpilot/system/loggerd/encoderd.cc

@@ -1,5 +1,12 @@
 #include <cassert>
+#ifdef __TICI__
+#include <exception>
+#include <stdexcept>
+#endif
 
+#ifdef __TICI__
+#include "system/loggerd/clip_encoder.h"
+#endif
 #include "system/loggerd/loggerd.h"
 #include "system/loggerd/encoder/jpeg_encoder.h"
 
@@ -171,6 +178,37 @@ void encoderd_thread(const LogCameraInfo (&cameras)[N]) {
 }
 
 int main(int argc, char* argv[]) {
+#ifdef __TICI__
+  if (argc > 1 && std::string(argv[1]) == "--clip") {
+    if (argc < 6) {
+      fprintf(stderr, "usage: encoderd --clip OUTPUT START DURATION [--bitrate BPS] [--speedup N] "
+                      "[--metadata JSON] SEGMENT [SEGMENT ...]\n");
+      return 2;
+    }
+    try {
+      int bitrate = 5'000'000;
+      int speedup = 1;
+      std::string metadata;
+      int input_arg = 5;
+      while (input_arg < argc && std::string(argv[input_arg]).rfind("--", 0) == 0) {
+        const std::string option = argv[input_arg++];
+        if (option == "--") break;
+        if (input_arg == argc) throw std::invalid_argument("missing clip option value");
+        if (option == "--bitrate") bitrate = std::stoi(argv[input_arg++]);
+        else if (option == "--speedup") speedup = std::stoi(argv[input_arg++]);
+        else if (option == "--metadata") metadata = argv[input_arg++];
+        else throw std::invalid_argument("unknown clip option: " + option);
+      }
+      if (input_arg == argc) throw std::invalid_argument("missing clip input");
+      std::vector<std::string> inputs(argv + input_arg, argv + argc);
+      return encode_clip(inputs, argv[2], std::stod(argv[3]), std::stod(argv[4]),
+                         bitrate, speedup, metadata);
+    } catch (const std::exception &e) {
+      fprintf(stderr, "clip encoding failed: %s\n", e.what());
+      return 1;
+    }
+  }
+#endif
   if (!Hardware::PC()) {
     int ret;
     ret = util::set_realtime_priority(52);

openpilot/system/loggerd/video_writer.cc

@@ -49,6 +49,11 @@ VideoWriter::VideoWriter(const char *path, const char *filename, bool remuxing,
   }
 }
 
+void VideoWriter::set_metadata(const char *key, const char *value) {
+  assert(remuxing && !header_written);
+  av_dict_set(&ofmt_ctx->metadata, key, value, 0);
+}
+
 void VideoWriter::initialize_audio(int sample_rate) {
   assert(this->ofmt_ctx->oformat->audio_codec != AV_CODEC_ID_NONE); // check output format supports audio streams
   const AVCodec *audio_avcodec = avcodec_find_encoder(AV_CODEC_ID_AAC);
@@ -106,7 +111,10 @@ void VideoWriter::write(uint8_t *data, int len, long long timestamp, bool codecc
       int err = avcodec_parameters_from_context(out_stream->codecpar, codec_ctx);
       assert(err >= 0);
       // if there is an audio stream, it must be initialized before this point
-      err = avformat_write_header(ofmt_ctx, NULL);
+      AVDictionary *options = nullptr;
+      if (ofmt_ctx->metadata) av_dict_set(&options, "movflags", "+faststart+use_metadata_tags", 0);
+      err = avformat_write_header(ofmt_ctx, &options);
+      av_dict_free(&options);
       assert(err >= 0);
       header_written = true;
     } else {

openpilot/system/loggerd/video_writer.h

@@ -13,6 +13,7 @@ extern "C" {
 class VideoWriter {
 public:
   VideoWriter(const char *path, const char *filename, bool remuxing, int width, int height, int fps, cereal::EncodeIndex::Type codec);
+  void set_metadata(const char *key, const char *value);
   void write(uint8_t *data, int len, long long timestamp, bool codecconfig, bool keyframe);
   void write_audio(uint8_t *data, int len, long long timestamp, int sample_rate);