IQ.Pilot Release Commit @ a0d13ad

This commit is contained in:
IQ.Lvbs CI [bot]
2026-08-04 14:48:16 -05:00
commit 2b6f6d7d3e
4529 changed files with 1131858 additions and 0 deletions

4
system/loggerd/.gitignore vendored Normal file
View File

@@ -0,0 +1,4 @@
loggerd
encoderd
bootlog
tests/test_logger

34
system/loggerd/SConscript Normal file
View File

@@ -0,0 +1,34 @@
Import('env', 'arch', 'messaging', 'common', 'visionipc')
libs = [common, messaging, visionipc,
'avformat', 'avcodec', 'avutil',
'yuv', 'OpenCL', 'pthread', 'zstd']
src = ['logger.cc', 'zstd_writer.cc', 'video_writer.cc', 'encoder/encoder.cc', 'encoder/v4l_encoder.cc', 'encoder/jpeg_encoder.cc']
if arch != "larch64":
src += ['encoder/ffmpeg_encoder.cc']
if arch == "Darwin":
# fix OpenCL
del libs[libs.index('OpenCL')]
env['FRAMEWORKS'] = ['OpenCL']
# exclude v4l
del src[src.index('encoder/v4l_encoder.cc')]
logger_lib = env.Library('logger', src)
libs.insert(0, logger_lib)
env.Program('loggerd', ['loggerd.cc'], LIBS=libs)
env.Program('encoderd', ['encoderd.cc'], LIBS=libs + ["jpeg"])
env.Program('bootlog.cc', LIBS=libs)
# Standalone hardware HEVC decoder (Venus/msm_vidc) for the offroad route viewer, reusing comma's
# tested decoder from tools/replay/qcom_decoder.cc.
if arch == "larch64":
# Compile comma's decoder under a distinct object name (tools/replay also builds qcom_decoder.o).
qcom_obj = env.Object('encoder/qcom_decoder_hwdec', '#tools/replay/qcom_decoder.cc')
env.Program('encoder/v4l_decode', ['encoder/v4l_decode.cc', qcom_obj],
LIBS=[common, visionipc, 'yuv', 'avformat', 'avcodec', 'avutil', 'pthread', 'OpenCL', 'zmq'])
if GetOption('extras'):
env.Program('tests/test_logger', ['tests/test_runner.cc', 'tests/test_logger.cc', 'tests/test_zstd_writer.cc'], LIBS=libs + ['curl', 'crypto'])

View File

68
system/loggerd/bootlog.cc Normal file
View File

@@ -0,0 +1,68 @@
#include <cassert>
#include <string>
#include "cereal/messaging/messaging.h"
#include "common/params.h"
#include "common/swaglog.h"
#include "system/loggerd/logger.h"
#include "system/loggerd/zstd_writer.h"
static kj::Array<capnp::word> build_boot_log() {
MessageBuilder msg;
auto boot = msg.initEvent().initBoot();
boot.setWallTimeNanos(nanos_since_epoch());
std::string pstore = "/sys/fs/pstore";
std::map<std::string, std::string> pstore_map = util::read_files_in_dir(pstore);
int i = 0;
auto lpstore = boot.initPstore().initEntries(pstore_map.size());
for (auto& kv : pstore_map) {
auto lentry = lpstore[i];
lentry.setKey(kv.first);
lentry.setValue(capnp::Data::Reader((const kj::byte*)kv.second.data(), kv.second.size()));
i++;
}
// Gather output of commands
std::vector<std::string> bootlog_commands = {
"[ -x \"$(command -v journalctl)\" ] && journalctl -o short-monotonic",
};
auto commands = boot.initCommands().initEntries(bootlog_commands.size());
for (int j = 0; j < bootlog_commands.size(); j++) {
auto lentry = commands[j];
lentry.setKey(bootlog_commands[j]);
const std::string result = util::check_output(bootlog_commands[j]);
lentry.setValue(capnp::Data::Reader((const kj::byte*)result.data(), result.size()));
}
boot.setLaunchLog(util::read_file("/tmp/launch_log"));
return capnp::messageToFlatArray(msg);
}
int main(int argc, char** argv) {
const std::string id = logger_get_identifier("BootCount");
const std::string path = Path::log_root() + "/boot/" + id + ".zst";
LOGW("bootlog to %s", path.c_str());
// Open bootlog
bool r = util::create_directories(Path::log_root() + "/boot/", 0775);
assert(r);
ZstdFileWriter file(path, LOG_COMPRESSION_LEVEL);
// Write initdata
file.write(logger_build_init_data().asBytes());
// Write bootlog
file.write(build_boot_log().asBytes());
// Write out bootlog param to match routes with bootlog
Params().put("CurrentBootlog", id.c_str());
return 0;
}

31
system/loggerd/config.py Normal file
View File

@@ -0,0 +1,31 @@
import os
from openpilot.system.hardware.hw import Paths
CAMERA_FPS = 20
SEGMENT_LENGTH = 60
PATH_DICT = {
"internal": Paths.log_root(),
"external": Paths.log_root_external()
}
def get_available_percent(default: float, path_type="internal") -> float:
try:
statvfs = os.statvfs(PATH_DICT[path_type])
available_percent = 100.0 * statvfs.f_bavail / statvfs.f_blocks
except (OSError, KeyError):
available_percent = default
return available_percent
def get_available_bytes(default: int, path_type="internal") -> int:
try:
statvfs = os.statvfs(PATH_DICT[path_type])
available_bytes = statvfs.f_bavail * statvfs.f_frsize
except (OSError, KeyError):
available_bytes = default
return available_bytes

View File

@@ -0,0 +1,45 @@
import os
from openpilot.common.swaglog import cloudlog
from openpilot.system.hardware.hw import Paths
PRESERVE_ATTR_NAME = b"user.preserve"
PRESERVE_ATTR_VALUE = b"1"
def recover_unclean_segments(log_root: str | None = None) -> list[str]:
# Segments with leftover .lock files are from a loggerd that never closed
# cleanly (power cut, crash). The video/log data in them is valid up to the
# last durable sync. Clear the stale locks so the deleter can manage them
# again, and preserve them: footage from an unclean shutdown is exactly the
# footage a dashcam must not throw away.
root = log_root if log_root is not None else Paths.log_root()
recovered = []
try:
dirs = os.listdir(root)
except OSError:
return recovered
for d in dirs:
seg_path = os.path.join(root, d)
if not os.path.isdir(seg_path):
continue
try:
locks = [f for f in os.listdir(seg_path) if f.endswith(".lock")]
if not locks:
continue
for lock in locks:
os.unlink(os.path.join(seg_path, lock))
setxattr = getattr(os, "setxattr", None) # not available on darwin
if setxattr is not None:
try:
setxattr(seg_path, PRESERVE_ATTR_NAME, PRESERVE_ATTR_VALUE)
except OSError:
pass
recovered.append(d)
except OSError:
cloudlog.exception(f"crash_recovery: failed to recover {seg_path}")
if recovered:
cloudlog.event("crash_recovery.recovered_unclean_segments", segments=sorted(recovered), error=True)
return recovered

117
system/loggerd/deleter.py Executable file
View File

@@ -0,0 +1,117 @@
#!/usr/bin/env python3
import os
import time
import shutil
import threading
from pathlib import Path
from openpilot.system.hardware.hw import Paths
from openpilot.common.swaglog import cloudlog
from openpilot.system.loggerd.config import get_available_bytes, get_available_percent
from openpilot.system.loggerd.uploader_common import listdir_by_creation
from openpilot.system.loggerd.xattr_cache import getxattr
MIN_BYTES = 5 * 1024 * 1024 * 1024
MIN_PERCENT = 10
DELETE_LAST = ['boot', 'crash']
PRESERVE_ATTR_NAME = 'user.preserve'
PRESERVE_ATTR_VALUE = b'1'
PRESERVE_COUNT = 5
def has_preserve_xattr(d: str) -> bool:
return getxattr(os.path.join(Paths.log_root(), d), PRESERVE_ATTR_NAME) == PRESERVE_ATTR_VALUE
def get_preserved_segments(dirs_by_creation: list[str]) -> set[str]:
# skip deleting most recent N preserved segments (and their prior segment)
preserved = set()
for n, d in enumerate(filter(has_preserve_xattr, reversed(dirs_by_creation))):
if n == PRESERVE_COUNT:
break
date_str, _, seg_str = d.rpartition("--")
# ignore non-segment directories
if not date_str:
continue
try:
seg_num = int(seg_str)
except ValueError:
continue
# preserve segment and two prior
for _seg_num in range(max(0, seg_num - 2), seg_num + 1):
preserved.add(f"{date_str}--{_seg_num}")
return preserved
def deleter_thread(exit_event: threading.Event):
while not exit_event.is_set():
out_of_bytes = get_available_bytes(default=MIN_BYTES + 1) < MIN_BYTES
out_of_percent = get_available_percent(default=MIN_PERCENT + 1) < MIN_PERCENT
if out_of_percent or out_of_bytes:
dirs = listdir_by_creation(Paths.log_root())
preserved_dirs = get_preserved_segments(dirs)
# remove the earliest directory we can
for delete_dir in sorted(dirs, key=lambda d: (d in DELETE_LAST, d in preserved_dirs)):
delete_path = os.path.join(Paths.log_root(), delete_dir)
if any(name.endswith(".lock") for name in os.listdir(delete_path)):
continue
if Path(Paths.log_root_external()).is_mount():
out_of_bytes_external = get_available_bytes(default=MIN_BYTES + 1, path_type="external") < MIN_BYTES
out_of_percent_external = get_available_percent(default=MIN_PERCENT + 1, path_type="external") < MIN_PERCENT
if out_of_percent_external or out_of_bytes_external:
dirs_external = listdir_by_creation(Paths.log_root_external())
# remove the earliest external directory we can
for delete_dir_external in sorted(dirs_external):
delete_path_external = os.path.join(Paths.log_root_external(), delete_dir_external)
try:
cloudlog.warning(f"deleting {delete_path_external}")
shutil.rmtree(delete_path_external)
break
except OSError:
cloudlog.exception(f"issue deleting {delete_path_external}")
# move directory from internal to external
path_external = os.path.join(Paths.log_root_external(), delete_dir)
try:
cloudlog.warning(f"moving {delete_path} to {path_external}")
start = time.monotonic()
shutil.move(delete_path, path_external)
cloudlog.warning(f"moved {delete_path} to {path_external} in {time.monotonic() - start:.2f}s")
break
except Exception:
cloudlog.error(f"issue moving {delete_path} to {path_external}")
try:
cloudlog.warning(f"deleting {delete_path}")
shutil.rmtree(delete_path)
break
except OSError:
cloudlog.exception(f"issue deleting {delete_path}")
continue
try:
cloudlog.info(f"deleting {delete_path}")
shutil.rmtree(delete_path)
break
except OSError:
cloudlog.exception(f"issue deleting {delete_path}")
exit_event.wait(.1)
else:
exit_event.wait(30)
def main():
deleter_thread(threading.Event())
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,42 @@
#include "system/loggerd/encoder/encoder.h"
VideoEncoder::VideoEncoder(const EncoderInfo &encoder_info, int in_width, int in_height)
: encoder_info(encoder_info), in_width(in_width), in_height(in_height) {
out_width = encoder_info.frame_width > 0 ? encoder_info.frame_width : in_width;
out_height = encoder_info.frame_height > 0 ? encoder_info.frame_height : in_height;
pm.reset(new PubMaster(std::vector{encoder_info.publish_name}));
}
void VideoEncoder::publisher_publish(int segment_num, uint32_t idx, VisionIpcBufExtra &extra,
unsigned int flags, kj::ArrayPtr<capnp::byte> header, kj::ArrayPtr<capnp::byte> dat) {
MessageBuilder msg;
auto event = msg.initEvent(true);
auto edat = (event.*(encoder_info.init_encode_data_func))();
auto edata = edat.initIdx();
struct timespec ts;
timespec_get(&ts, TIME_UTC);
edat.setUnixTimestampNanos((uint64_t)ts.tv_sec*1000000000 + ts.tv_nsec);
edata.setFrameId(extra.frame_id);
edata.setTimestampSof(extra.timestamp_sof);
edata.setTimestampEof(extra.timestamp_eof);
edata.setType(encoder_info.get_settings(in_width).encode_type);
edata.setEncodeId(cnt++);
edata.setSegmentNum(segment_num);
edata.setSegmentId(idx);
edata.setFlags(flags);
edata.setLen(dat.size());
edat.adoptData(msg.getOrphanage().referenceExternalData(dat));
edat.setWidth(out_width);
edat.setHeight(out_height);
if (flags & V4L2_BUF_FLAG_KEYFRAME) edat.setHeader(header);
uint32_t bytes_size = capnp::computeSerializedSizeInWords(msg) * sizeof(capnp::word);
if (msg_cache.size() < bytes_size) {
msg_cache.resize(bytes_size);
}
kj::ArrayOutputStream output_stream(kj::ArrayPtr<capnp::byte>(msg_cache.data(), bytes_size));
capnp::writeMessage(output_stream, msg);
pm->send(encoder_info.publish_name, msg_cache.data(), bytes_size);
}

View File

@@ -0,0 +1,46 @@
#pragma once
// has to be in this order
#ifdef __linux__
#include "third_party/linux/include/v4l2-controls.h"
#include <linux/videodev2.h>
#else
#define V4L2_BUF_FLAG_KEYFRAME 8
#endif
#include <cassert>
#include <cstdint>
#include <memory>
#include <thread>
#include <vector>
#include "cereal/messaging/messaging.h"
#include "msgq/visionipc/visionipc.h"
#include "common/queue.h"
#include "system/loggerd/loggerd.h"
class VideoEncoder {
public:
VideoEncoder(const EncoderInfo &encoder_info, int in_width, int in_height);
virtual ~VideoEncoder() {}
virtual int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra) = 0;
virtual void encoder_open() = 0;
virtual void encoder_close() = 0;
// Runtime bitrate update for adaptive livestream encoders. No-op by default.
virtual void set_bitrate(int bitrate) {}
// Force an IDR keyframe on the next encoded frame (fast client start / recovery). No-op by default.
virtual void request_keyframe() {}
void publisher_publish(int segment_num, uint32_t idx, VisionIpcBufExtra &extra, unsigned int flags, kj::ArrayPtr<capnp::byte> header, kj::ArrayPtr<capnp::byte> dat);
protected:
int in_width, in_height;
int out_width, out_height;
const EncoderInfo encoder_info;
private:
// total frames encoded
int cnt = 0;
std::unique_ptr<PubMaster> pm;
std::vector<capnp::byte> msg_cache;
};

View File

@@ -0,0 +1,150 @@
#include "system/loggerd/encoder/ffmpeg_encoder.h"
#include <fcntl.h>
#include <unistd.h>
#include <cassert>
#include <cstdio>
#include <cstdlib>
#define __STDC_CONSTANT_MACROS
#include "third_party/libyuv/include/libyuv.h"
extern "C" {
#include <libavcodec/avcodec.h>
#include <libavformat/avformat.h>
#include <libavutil/imgutils.h>
}
#include "common/swaglog.h"
#include "common/util.h"
const int env_debug_encoder = (getenv("DEBUG_ENCODER") != NULL) ? atoi(getenv("DEBUG_ENCODER")) : 0;
FfmpegEncoder::FfmpegEncoder(const EncoderInfo &encoder_info, int in_width, int in_height)
: VideoEncoder(encoder_info, in_width, in_height) {
frame = av_frame_alloc();
assert(frame);
frame->format = AV_PIX_FMT_YUV420P;
frame->width = out_width;
frame->height = out_height;
frame->linesize[0] = out_width;
frame->linesize[1] = out_width/2;
frame->linesize[2] = out_width/2;
convert_buf.resize(in_width * in_height * 3 / 2);
if (in_width != out_width || in_height != out_height) {
downscale_buf.resize(out_width * out_height * 3 / 2);
}
}
FfmpegEncoder::~FfmpegEncoder() {
encoder_close();
av_frame_free(&frame);
}
void FfmpegEncoder::encoder_open() {
auto codec_id = encoder_info.get_settings(in_width).encode_type == cereal::EncodeIndex::Type::QCAMERA_H264
? AV_CODEC_ID_H264
: AV_CODEC_ID_FFVHUFF;
const AVCodec *codec = avcodec_find_encoder(codec_id);
this->codec_ctx = avcodec_alloc_context3(codec);
assert(this->codec_ctx);
this->codec_ctx->width = frame->width;
this->codec_ctx->height = frame->height;
this->codec_ctx->pix_fmt = AV_PIX_FMT_YUV420P;
this->codec_ctx->time_base = (AVRational){ 1, encoder_info.fps };
int err = avcodec_open2(this->codec_ctx, codec, NULL);
assert(err >= 0);
is_open = true;
segment_num++;
counter = 0;
}
void FfmpegEncoder::encoder_close() {
if (!is_open) return;
avcodec_free_context(&codec_ctx);
is_open = false;
}
int FfmpegEncoder::encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra) {
assert(buf->width == this->in_width);
assert(buf->height == this->in_height);
uint8_t *cy = convert_buf.data();
uint8_t *cu = cy + in_width * in_height;
uint8_t *cv = cu + (in_width / 2) * (in_height / 2);
libyuv::NV12ToI420(buf->y, buf->stride,
buf->uv, buf->stride,
cy, in_width,
cu, in_width/2,
cv, in_width/2,
in_width, in_height);
if (downscale_buf.size() > 0) {
uint8_t *out_y = downscale_buf.data();
uint8_t *out_u = out_y + frame->width * frame->height;
uint8_t *out_v = out_u + (frame->width / 2) * (frame->height / 2);
libyuv::I420Scale(cy, in_width,
cu, in_width/2,
cv, in_width/2,
in_width, in_height,
out_y, frame->width,
out_u, frame->width/2,
out_v, frame->width/2,
frame->width, frame->height,
libyuv::kFilterNone);
frame->data[0] = out_y;
frame->data[1] = out_u;
frame->data[2] = out_v;
} else {
frame->data[0] = cy;
frame->data[1] = cu;
frame->data[2] = cv;
}
frame->pts = counter*50*1000; // 50ms per frame
int ret = counter;
int err = avcodec_send_frame(this->codec_ctx, frame);
if (err < 0) {
LOGE("avcodec_send_frame error %d", err);
ret = -1;
}
AVPacket pkt = {};
pkt.data = NULL;
pkt.size = 0;
while (ret >= 0) {
err = avcodec_receive_packet(this->codec_ctx, &pkt);
if (err == AVERROR_EOF) {
break;
} else if (err == AVERROR(EAGAIN)) {
// Encoder might need a few frames on startup to get started. Keep going
ret = 0;
break;
} else if (err < 0) {
LOGE("avcodec_receive_packet error %d", err);
ret = -1;
break;
}
if (env_debug_encoder) {
printf("%20s got %8d bytes flags %8x idx %4d id %8d\n", encoder_info.publish_name, pkt.size, pkt.flags, counter, extra->frame_id);
}
publisher_publish(segment_num, counter, *extra,
(pkt.flags & AV_PKT_FLAG_KEY) ? V4L2_BUF_FLAG_KEYFRAME : 0,
kj::arrayPtr<capnp::byte>(pkt.data, (size_t)0), // TODO: get the header
kj::arrayPtr<capnp::byte>(pkt.data, pkt.size));
counter++;
}
av_packet_unref(&pkt);
return ret;
}

View File

@@ -0,0 +1,34 @@
#pragma once
#include <cstdio>
#include <cstdlib>
#include <string>
#include <vector>
extern "C" {
#include <libavcodec/avcodec.h>
#include <libavformat/avformat.h>
#include <libavutil/imgutils.h>
}
#include "system/loggerd/encoder/encoder.h"
#include "system/loggerd/loggerd.h"
class FfmpegEncoder : public VideoEncoder {
public:
FfmpegEncoder(const EncoderInfo &encoder_info, int in_width, int in_height);
~FfmpegEncoder();
int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra);
void encoder_open();
void encoder_close();
private:
int segment_num = -1;
int counter = 0;
bool is_open = false;
AVCodecContext *codec_ctx;
AVFrame *frame = NULL;
std::vector<uint8_t> convert_buf;
std::vector<uint8_t> downscale_buf;
};

View File

@@ -0,0 +1,105 @@
#include "system/loggerd/encoder/jpeg_encoder.h"
#include <cassert>
#include <cstring>
JpegEncoder::JpegEncoder(const std::string &pusblish_name, int width, int height)
: publish_name(pusblish_name), thumbnail_width(width), thumbnail_height(height) {
yuv_buffer.resize((thumbnail_width * ((thumbnail_height + 15) & ~15) * 3) / 2);
pm = std::make_unique<PubMaster>(std::vector{pusblish_name.c_str()});
}
JpegEncoder::~JpegEncoder() {
if (out_buffer) {
free(out_buffer);
}
}
void JpegEncoder::pushThumbnail(VisionBuf *buf, const VisionIpcBufExtra &extra) {
generateThumbnail(buf->y, buf->uv, buf->width, buf->height, buf->stride);
MessageBuilder msg;
auto thumbnaild = msg.initEvent().initThumbnail();
thumbnaild.setFrameId(extra.frame_id);
thumbnaild.setTimestampEof(extra.timestamp_eof);
thumbnaild.setThumbnail({out_buffer, out_size});
pm->send(publish_name.c_str(), msg);
}
void JpegEncoder::generateThumbnail(const uint8_t *y_addr, const uint8_t *uv_addr, int width, int height, int stride) {
int downscale = width / thumbnail_width;
assert(downscale * thumbnail_height == height);
// make the buffer big enough. jpeg_write_raw_data requires 16-pixels aligned height to be used.
uint8_t *y_plane = yuv_buffer.data();
uint8_t *u_plane = y_plane + thumbnail_width * thumbnail_height;
uint8_t *v_plane = u_plane + (thumbnail_width * thumbnail_height) / 4;
{
// subsampled conversion from nv12 to yuv
for (int hy = 0; hy < thumbnail_height / 2; hy++) {
for (int hx = 0; hx < thumbnail_width / 2; hx++) {
int ix = hx * downscale + (downscale - 1) / 2;
int iy = hy * downscale + (downscale - 1) / 2;
y_plane[(hy * 2 + 0) * thumbnail_width + (hx * 2 + 0)] = y_addr[(iy * 2 + 0) * stride + ix * 2 + 0];
y_plane[(hy * 2 + 0) * thumbnail_width + (hx * 2 + 1)] = y_addr[(iy * 2 + 0) * stride + ix * 2 + 1];
y_plane[(hy * 2 + 1) * thumbnail_width + (hx * 2 + 0)] = y_addr[(iy * 2 + 1) * stride + ix * 2 + 0];
y_plane[(hy * 2 + 1) * thumbnail_width + (hx * 2 + 1)] = y_addr[(iy * 2 + 1) * stride + ix * 2 + 1];
u_plane[hy * thumbnail_width / 2 + hx] = uv_addr[iy * stride + ix * 2 + 0];
v_plane[hy * thumbnail_width / 2 + hx] = uv_addr[iy * stride + ix * 2 + 1];
}
}
}
compressToJpeg(y_plane, u_plane, v_plane);
}
void JpegEncoder::compressToJpeg(uint8_t *y_plane, uint8_t *u_plane, uint8_t *v_plane) {
struct jpeg_compress_struct cinfo;
struct jpeg_error_mgr jerr;
cinfo.err = jpeg_std_error(&jerr);
jpeg_create_compress(&cinfo);
if (out_buffer) {
free(out_buffer);
out_buffer = nullptr;
out_size = 0;
}
jpeg_mem_dest(&cinfo, &out_buffer, &out_size);
cinfo.image_width = thumbnail_width;
cinfo.image_height = thumbnail_height;
cinfo.input_components = 3;
jpeg_set_defaults(&cinfo);
jpeg_set_colorspace(&cinfo, JCS_YCbCr);
// configure sampling factors for yuv420.
cinfo.comp_info[0].h_samp_factor = 2; // Y
cinfo.comp_info[0].v_samp_factor = 2;
cinfo.comp_info[1].h_samp_factor = 1; // U
cinfo.comp_info[1].v_samp_factor = 1;
cinfo.comp_info[2].h_samp_factor = 1; // V
cinfo.comp_info[2].v_samp_factor = 1;
cinfo.raw_data_in = TRUE;
jpeg_set_quality(&cinfo, 50, TRUE);
jpeg_start_compress(&cinfo, TRUE);
JSAMPROW y[16], u[8], v[8];
JSAMPARRAY planes[3]{y, u, v};
for (int line = 0; line < cinfo.image_height; line += 16) {
for (int i = 0; i < 16; ++i) {
y[i] = y_plane + (line + i) * cinfo.image_width;
if (i % 2 == 0) {
int offset = (cinfo.image_width / 2) * ((i + line) / 2);
u[i / 2] = u_plane + offset;
v[i / 2] = v_plane + offset;
}
}
jpeg_write_raw_data(&cinfo, planes, 16);
}
jpeg_finish_compress(&cinfo);
jpeg_destroy_compress(&cinfo);
}

View File

@@ -0,0 +1,32 @@
#pragma once
#include <cstdio>
#include <cstdlib>
#include <cstddef>
#include <cstdint>
#include <jpeglib.h>
#include <vector>
#include <memory>
#include "cereal/messaging/messaging.h"
#include "msgq/visionipc/visionbuf.h"
class JpegEncoder {
public:
JpegEncoder(const std::string &pusblish_name, int width, int height);
~JpegEncoder();
void pushThumbnail(VisionBuf *buf, const VisionIpcBufExtra &extra);
private:
void generateThumbnail(const uint8_t *y, const uint8_t *uv, int width, int height, int stride);
void compressToJpeg(uint8_t *y_plane, uint8_t *u_plane, uint8_t *v_plane);
int thumbnail_width;
int thumbnail_height;
std::string publish_name;
std::vector<uint8_t> yuv_buffer;
std::unique_ptr<PubMaster> pm;
// JPEG output buffer
unsigned char* out_buffer = nullptr;
unsigned long out_size = 0;
};

View File

@@ -0,0 +1,91 @@
// Standalone hardware HEVC decoder for the offroad route viewer, built on comma's tested Venus
// (msm_vidc) decoder from tools/replay/qcom_decoder.cc. ffmpeg's generic hevc_v4l2m2m can't drive
// the downstream msm_vidc interface; MsmVidc has the exact firmware recipe (S_EXT_CTRLS
// STREAM_OUTPUT_MODE/DPB_COLOR_FORMAT, the msm_vidc PORT_SETTINGS_CHANGED event, ION USERPTR
// buffers). We demux the .hevc with libav, feed packets, and stream decoded NV12->rgb24 to stdout.
//
// Usage: v4l_decode <input.hevc>
// stdout: "DIM <w> <h>\n" once, then raw rgb24 frames (w*h*3 bytes each).
// Offroad the decoder + GPU are idle (camerad/modeld stopped), so the viewer can use it.
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <vector>
#include "tools/replay/qcom_decoder.h"
#include "third_party/libyuv/include/libyuv.h"
#include "third_party/linux/include/msm_media_info.h"
extern "C" {
#include <libavformat/avformat.h>
#include <libavcodec/avcodec.h>
}
int main(int argc, char **argv) {
if (argc < 2) {
fprintf(stderr, "usage: %s <input.hevc>\n", argv[0]);
return 2;
}
AVFormatContext *fmt = avformat_alloc_context();
fmt->probesize = 10 * 1024 * 1024; // raw .hevc needs a big probe to index all frames
if (avformat_open_input(&fmt, argv[1], nullptr, nullptr) != 0 ||
avformat_find_stream_info(fmt, nullptr) < 0) {
fprintf(stderr, "v4l_decode: cannot open %s\n", argv[1]);
return 1;
}
int vstream = av_find_best_stream(fmt, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0);
if (vstream < 0) { fprintf(stderr, "v4l_decode: no video stream\n"); return 1; }
AVCodecParameters *cp = fmt->streams[vstream]->codecpar;
if (cp->codec_id != AV_CODEC_ID_HEVC) {
fprintf(stderr, "v4l_decode: only HEVC supported (got codec %d)\n", cp->codec_id);
return 3;
}
const int w = cp->width, h = cp->height;
MsmVidc dec;
if (!dec.init(VIDEO_DEVICE, w, h, V4L2_PIX_FMT_HEVC)) {
fprintf(stderr, "v4l_decode: msm_vidc init failed\n");
return 1;
}
dec.avctx = fmt;
// Decoded-frame target buffer (Venus NV12 layout), matching how MsmVidc copies the capture plane.
const size_t y_stride = VENUS_Y_STRIDE(COLOR_FMT_NV12, w);
const size_t uv_offset = y_stride * h;
VisionBuf out;
out.allocate(uv_offset + y_stride * ((h + 1) / 2));
out.init_yuv(w, h, y_stride, uv_offset);
std::vector<uint8_t> argb((size_t)w * h * 4), rgb((size_t)w * h * 3);
// stdout carries only raw rgb24 frames (so this is a drop-in for the ffmpeg rgb pipe); dims go
// to stderr. The consumer already knows WxH from ffprobe.
fprintf(stderr, "v4l_decode: HW decode %dx%d y_stride=%zu\n", w, h, y_stride);
AVPacket pkt;
int npkt = 0, nframe = 0, rr;
while ((rr = av_read_frame(fmt, &pkt)) == 0) {
if (pkt.stream_index == vstream && pkt.size > 0) {
npkt++;
VisionBuf *res = dec.decodeFrame(&pkt, &out);
if (res != nullptr) {
nframe++;
libyuv::NV12ToARGB(out.y, out.stride, out.uv, out.stride, argb.data(), w * 4, w, h);
// ARGBToRAW gives R,G,B byte order (matches the R8G8B8 texture / ffmpeg rgb24). ARGBToRGB24
// would give B,G,R -> the blue tint / swapped red-blue.
libyuv::ARGBToRAW(argb.data(), w * 4, rgb.data(), w * 3, w, h);
if (fwrite(rgb.data(), 1, rgb.size(), stdout) != rgb.size()) {
av_packet_unref(&pkt);
break; // downstream (player) closed the pipe
}
}
}
av_packet_unref(&pkt);
}
fprintf(stderr, "v4l_decode: done rr=%d pkts=%d frames=%d\n", rr, npkt, nframe);
fflush(stdout);
avformat_close_input(&fmt);
return 0;
}

View File

@@ -0,0 +1,356 @@
#include <cassert>
#include <string>
#include <sys/ioctl.h>
#include <poll.h>
#include "system/loggerd/encoder/v4l_encoder.h"
#include "common/util.h"
#include "common/timing.h"
#include "third_party/libyuv/include/libyuv.h"
#include "third_party/linux/include/msm_media_info.h"
// has to be in this order
#include "third_party/linux/include/v4l2-controls.h"
#include <linux/videodev2.h>
#define V4L2_QCOM_BUF_FLAG_CODECCONFIG 0x00020000
#define V4L2_QCOM_BUF_FLAG_EOS 0x02000000
/*
kernel debugging:
echo 0xff > /sys/module/videobuf2_core/parameters/debug
echo 0x7fffffff > /sys/kernel/debug/msm_vidc/debug_level
echo 0xff > /sys/devices/platform/soc/aa00000.qcom,vidc/video4linux/video33/dev_debug
*/
const int env_debug_encoder = (getenv("DEBUG_ENCODER") != NULL) ? atoi(getenv("DEBUG_ENCODER")) : 0;
static void dequeue_buffer(int fd, v4l2_buf_type buf_type, unsigned int *index=NULL, unsigned int *bytesused=NULL, unsigned int *flags=NULL, struct timeval *timestamp=NULL) {
v4l2_plane plane = {0};
v4l2_buffer v4l_buf = {
.type = buf_type,
.memory = V4L2_MEMORY_USERPTR,
.m = { .planes = &plane, },
.length = 1,
};
util::safe_ioctl(fd, VIDIOC_DQBUF, &v4l_buf, "VIDIOC_DQBUF failed");
if (index) *index = v4l_buf.index;
if (bytesused) *bytesused = v4l_buf.m.planes[0].bytesused;
if (flags) *flags = v4l_buf.flags;
if (timestamp) *timestamp = v4l_buf.timestamp;
assert(v4l_buf.m.planes[0].data_offset == 0);
}
static void queue_buffer(int fd, v4l2_buf_type buf_type, unsigned int index, VisionBuf *buf, struct timeval timestamp={}) {
v4l2_plane plane = {
.length = (unsigned int)buf->len,
.m = { .userptr = (unsigned long)buf->addr, },
.bytesused = (uint32_t)buf->len,
.reserved = {(unsigned int)buf->fd}
};
v4l2_buffer v4l_buf = {
.type = buf_type,
.index = index,
.memory = V4L2_MEMORY_USERPTR,
.m = { .planes = &plane, },
.length = 1,
.flags = V4L2_BUF_FLAG_TIMESTAMP_COPY,
.timestamp = timestamp
};
util::safe_ioctl(fd, VIDIOC_QBUF, &v4l_buf, "VIDIOC_QBUF failed");
}
static void request_buffers(int fd, v4l2_buf_type buf_type, unsigned int count) {
struct v4l2_requestbuffers reqbuf = {
.type = buf_type,
.memory = V4L2_MEMORY_USERPTR,
.count = count
};
util::safe_ioctl(fd, VIDIOC_REQBUFS, &reqbuf, "VIDIOC_REQBUFS failed");
}
void V4LEncoder::dequeue_handler(V4LEncoder *e) {
std::string dequeue_thread_name = "dq-"+std::string(e->encoder_info.publish_name);
util::set_thread_name(dequeue_thread_name.c_str());
e->segment_num++;
uint32_t idx = -1;
bool exit = false;
// POLLIN is capture, POLLOUT is frame
struct pollfd pfd;
pfd.events = POLLIN | POLLOUT;
pfd.fd = e->fd;
// save the header
kj::Array<capnp::byte> header;
while (!exit) {
int rc = poll(&pfd, 1, 1000);
if (rc < 0) {
if (errno != EINTR) {
// TODO: exit encoder?
// ignore the error and keep going
LOGE("poll failed (%d - %d)", rc, errno);
}
continue;
} else if (rc == 0) {
LOGE("encoder dequeue poll timeout");
continue;
}
if (env_debug_encoder >= 2) {
printf("%20s poll %x at %.2f ms\n", e->encoder_info.publish_name, pfd.revents, millis_since_boot());
}
int frame_id = -1;
if (pfd.revents & POLLIN) {
unsigned int bytesused, flags, index;
struct timeval timestamp;
dequeue_buffer(e->fd, V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE, &index, &bytesused, &flags, &timestamp);
e->buf_out[index].sync(VISIONBUF_SYNC_FROM_DEVICE);
uint8_t *buf = (uint8_t*)e->buf_out[index].addr;
int64_t ts = timestamp.tv_sec * 1000000 + timestamp.tv_usec;
// eof packet, we exit
if (flags & V4L2_QCOM_BUF_FLAG_EOS) {
exit = true;
} else if (flags & V4L2_QCOM_BUF_FLAG_CODECCONFIG) {
// save header
header = kj::heapArray<capnp::byte>(buf, bytesused);
} 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 (env_debug_encoder) {
printf("%20s got(%d) %6d bytes flags %8x idx %3d/%4d id %8d ts %ld lat %.2f ms (%lu frames free)\n",
e->encoder_info.publish_name, index, bytesused, flags, e->segment_num, idx, frame_id, ts, millis_since_boot()-(ts/1000.), e->free_buf_in.size());
}
// requeue the buffer
queue_buffer(e->fd, V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE, index, &e->buf_out[index]);
}
if (pfd.revents & POLLOUT) {
unsigned int index;
dequeue_buffer(e->fd, V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE, &index);
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) {
fd = HANDLE_EINTR(open("/dev/v4l/by-path/platform-aa00000.qcom_vidc-video-index1", O_RDWR|O_NONBLOCK));
assert(fd >= 0);
struct v4l2_capability cap;
util::safe_ioctl(fd, VIDIOC_QUERYCAP, &cap, "VIDIOC_QUERYCAP failed");
LOGD("opened encoder device %s %s = %d", cap.driver, cap.card, fd);
assert(strcmp((const char *)cap.driver, "msm_vidc_driver") == 0);
assert(strcmp((const char *)cap.card, "msm_vidc_venc") == 0);
EncoderSettings encoder_settings = encoder_info.get_settings(in_width);
bool is_h265 = encoder_settings.encode_type == cereal::EncodeIndex::Type::FULL_H_E_V_C;
current_bitrate = encoder_settings.bitrate;
adaptive_bitrate = encoder_info.adaptive_bitrate;
struct v4l2_format fmt_out = {
.type = V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE,
.fmt = {
.pix_mp = {
// downscales are free with v4l
.width = (unsigned int)(out_width),
.height = (unsigned int)(out_height),
.pixelformat = is_h265 ? V4L2_PIX_FMT_HEVC : V4L2_PIX_FMT_H264,
.field = V4L2_FIELD_ANY,
.colorspace = V4L2_COLORSPACE_DEFAULT,
}
}
};
util::safe_ioctl(fd, VIDIOC_S_FMT, &fmt_out, "VIDIOC_S_FMT failed");
v4l2_streamparm streamparm = {
.type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE,
.parm = {
.output = {
// TODO: more stuff here? we don't know
.timeperframe = {
.numerator = 1,
.denominator = (unsigned int)encoder_info.fps
}
}
}
};
util::safe_ioctl(fd, VIDIOC_S_PARM, &streamparm, "VIDIOC_S_PARM failed");
struct v4l2_format fmt_in = {
.type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE,
.fmt = {
.pix_mp = {
.width = (unsigned int)in_width,
.height = (unsigned int)in_height,
.pixelformat = V4L2_PIX_FMT_NV12,
.field = V4L2_FIELD_ANY,
.colorspace = V4L2_COLORSPACE_470_SYSTEM_BG,
}
}
};
util::safe_ioctl(fd, VIDIOC_S_FMT, &fmt_in, "VIDIOC_S_FMT failed");
LOGD("in buffer size %d, out buffer size %d",
fmt_in.fmt.pix_mp.plane_fmt[0].sizeimage,
fmt_out.fmt.pix_mp.plane_fmt[0].sizeimage);
// shared ctrls
{
// Livestream (adaptive) and qcamera use CBR so the bitrate target is actually enforced — the
// Venus VBR mode overshoots the target several-fold, which made qcamera.ts balloon. fcamera/
// ecamera/dcamera stay VBR for best recording quality.
const int rate_control = (adaptive_bitrate || encoder_info.cbr)
? V4L2_CID_MPEG_VIDC_VIDEO_RATE_CONTROL_CBR_CFR
: V4L2_CID_MPEG_VIDC_VIDEO_RATE_CONTROL_VBR_CFR;
struct v4l2_control ctrls[] = {
{ .id = V4L2_CID_MPEG_VIDEO_BITRATE, .value = encoder_settings.bitrate},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_NUM_P_FRAMES, .value = encoder_settings.gop_size - encoder_settings.b_frames - 1},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_NUM_B_FRAMES, .value = encoder_settings.b_frames},
{ .id = V4L2_CID_MPEG_VIDEO_HEADER_MODE, .value = V4L2_MPEG_VIDEO_HEADER_MODE_SEPARATE},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_RATE_CONTROL, .value = rate_control},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_PRIORITY, .value = V4L2_MPEG_VIDC_VIDEO_PRIORITY_REALTIME_DISABLE},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_IDR_PERIOD, .value = 1},
};
for (auto ctrl : ctrls) {
util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL failed");
}
}
if (is_h265) {
struct v4l2_control ctrls[] = {
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_HEVC_PROFILE, .value = V4L2_MPEG_VIDC_VIDEO_HEVC_PROFILE_MAIN},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_HEVC_TIER_LEVEL, .value = V4L2_MPEG_VIDC_VIDEO_HEVC_LEVEL_HIGH_TIER_LEVEL_5},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_VUI_TIMING_INFO, .value = V4L2_MPEG_VIDC_VIDEO_VUI_TIMING_INFO_ENABLED},
};
for (auto ctrl : ctrls) {
util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL failed");
}
} else {
struct v4l2_control ctrls[] = {
{ .id = V4L2_CID_MPEG_VIDEO_H264_PROFILE, .value = V4L2_MPEG_VIDEO_H264_PROFILE_HIGH},
{ .id = V4L2_CID_MPEG_VIDEO_H264_LEVEL, .value = V4L2_MPEG_VIDEO_H264_LEVEL_UNKNOWN},
{ .id = V4L2_CID_MPEG_VIDEO_H264_ENTROPY_MODE, .value = V4L2_MPEG_VIDEO_H264_ENTROPY_MODE_CABAC},
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_H264_CABAC_MODEL, .value = V4L2_CID_MPEG_VIDC_VIDEO_H264_CABAC_MODEL_0},
{ .id = V4L2_CID_MPEG_VIDEO_H264_LOOP_FILTER_MODE, .value = 0},
{ .id = V4L2_CID_MPEG_VIDEO_H264_LOOP_FILTER_ALPHA, .value = 0},
{ .id = V4L2_CID_MPEG_VIDEO_H264_LOOP_FILTER_BETA, .value = 0},
{ .id = V4L2_CID_MPEG_VIDEO_MULTI_SLICE_MODE, .value = 0},
};
for (auto ctrl : ctrls) {
util::safe_ioctl(fd, VIDIOC_S_CTRL, &ctrl, "VIDIOC_S_CTRL failed");
}
}
// allocate buffers
request_buffers(fd, V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE, BUF_OUT_COUNT);
request_buffers(fd, V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE, BUF_IN_COUNT);
// start encoder
v4l2_buf_type buf_type = V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE;
util::safe_ioctl(fd, VIDIOC_STREAMON, &buf_type, "VIDIOC_STREAMON failed");
buf_type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;
util::safe_ioctl(fd, VIDIOC_STREAMON, &buf_type, "VIDIOC_STREAMON failed");
// queue up output buffers
for (unsigned int i = 0; i < BUF_OUT_COUNT; i++) {
buf_out[i].allocate(fmt_out.fmt.pix_mp.plane_fmt[0].sizeimage);
queue_buffer(fd, V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE, i, &buf_out[i]);
}
// queue up input buffers
for (unsigned int i = 0; i < BUF_IN_COUNT; i++) {
free_buf_in.push(i);
}
}
void V4LEncoder::encoder_open() {
dequeue_handler_thread = std::thread(V4LEncoder::dequeue_handler, this);
this->is_open = true;
this->counter = 0;
}
void V4LEncoder::set_bitrate(int bitrate) {
// Only adaptive (livestream) encoders accept runtime bitrate changes; skip if unchanged.
if (!adaptive_bitrate || bitrate == current_bitrate) return;
if (bitrate <= 0) {
LOGE("invalid livestream encoder bitrate %d", bitrate);
return;
}
struct v4l2_control ctrl = { .id = V4L2_CID_MPEG_VIDEO_BITRATE, .value = bitrate };
if (HANDLE_EINTR(ioctl(fd, VIDIOC_S_CTRL, &ctrl)) != 0) {
LOGE("failed to update %s bitrate to %d", encoder_info.publish_name, bitrate);
return;
}
current_bitrate = bitrate;
}
void V4LEncoder::request_keyframe() {
// Force an IDR on the next encoded frame so a joining/recovering client gets a decodable frame
// immediately instead of waiting up to a full GOP. Only meaningful for livestream encoders.
if (!adaptive_bitrate) return;
struct v4l2_control ctrl = { .id = V4L2_CID_MPEG_VIDC_VIDEO_REQUEST_IFRAME, .value = 1 };
if (HANDLE_EINTR(ioctl(fd, VIDIOC_S_CTRL, &ctrl)) != 0) {
LOGE("failed to request keyframe for %s", encoder_info.publish_name);
}
}
int V4LEncoder::encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra) {
struct timeval timestamp {
.tv_sec = (long)(extra->timestamp_eof/1000000000),
.tv_usec = (long)((extra->timestamp_eof/1000) % 1000000),
};
// reserve buffer
int buffer_in = free_buf_in.pop();
// push buffer
extras.push(*extra);
//buf->sync(VISIONBUF_SYNC_TO_DEVICE);
queue_buffer(fd, V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE, buffer_in, buf, timestamp);
return this->counter++;
}
void V4LEncoder::encoder_close() {
if (this->is_open) {
// pop all the frames before closing, then put the buffers back
for (int i = 0; i < BUF_IN_COUNT; i++) free_buf_in.pop();
for (int i = 0; i < BUF_IN_COUNT; i++) free_buf_in.push(i);
// no frames, stop the encoder
struct v4l2_encoder_cmd encoder_cmd = { .cmd = V4L2_ENC_CMD_STOP };
util::safe_ioctl(fd, VIDIOC_ENCODER_CMD, &encoder_cmd, "VIDIOC_ENCODER_CMD failed");
// join waits for V4L2_QCOM_BUF_FLAG_EOS
dequeue_handler_thread.join();
assert(extras.empty());
}
this->is_open = false;
}
V4LEncoder::~V4LEncoder() {
encoder_close();
v4l2_buf_type buf_type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;
util::safe_ioctl(fd, VIDIOC_STREAMOFF, &buf_type, "VIDIOC_STREAMOFF failed");
request_buffers(fd, V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE, 0);
buf_type = V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE;
util::safe_ioctl(fd, VIDIOC_STREAMOFF, &buf_type, "VIDIOC_STREAMOFF failed");
request_buffers(fd, V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE, 0);
close(fd);
for (int i = 0; i < BUF_OUT_COUNT; i++) {
if (buf_out[i].free() != 0) {
LOGE("Failed to free buffer");
}
}
}

View File

@@ -0,0 +1,35 @@
#pragma once
#include "common/queue.h"
#include "system/loggerd/encoder/encoder.h"
#define BUF_IN_COUNT 7
#define BUF_OUT_COUNT 6
class V4LEncoder : public VideoEncoder {
public:
V4LEncoder(const EncoderInfo &encoder_info, int in_width, int in_height);
~V4LEncoder();
int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra);
void encoder_open();
void encoder_close();
void set_bitrate(int bitrate) override;
void request_keyframe() override;
private:
int fd;
bool adaptive_bitrate = false;
int current_bitrate = 0;
bool is_open = false;
int segment_num = -1;
int counter = 0;
SafeQueue<VisionIpcBufExtra> extras;
static void dequeue_handler(V4LEncoder *e);
std::thread dequeue_handler_thread;
VisionBuf buf_out[BUF_OUT_COUNT];
SafeQueue<unsigned int> free_buf_in;
};

206
system/loggerd/encoderd.cc Normal file
View File

@@ -0,0 +1,206 @@
#include <algorithm>
#include <cassert>
#include <string>
#include "system/loggerd/loggerd.h"
#include "system/loggerd/encoder/jpeg_encoder.h"
#ifdef __TICI__
#include "system/loggerd/encoder/v4l_encoder.h"
#define Encoder V4LEncoder
#else
#include "system/loggerd/encoder/ffmpeg_encoder.h"
#define Encoder FfmpegEncoder
#endif
ExitHandler do_exit;
struct EncoderdState {
int max_waiting = 0;
// Sync logic for startup
std::atomic<int> encoders_ready = 0;
std::atomic<uint32_t> start_frame_id = 0;
bool camera_ready[VISION_STREAM_WIDE_ROAD + 1] = {};
bool camera_synced[VISION_STREAM_WIDE_ROAD + 1] = {};
};
// Handle initial encoder syncing by waiting for all encoders to reach the same frame id
bool sync_encoders(EncoderdState *s, VisionStreamType cam_type, uint32_t frame_id) {
if (s->camera_synced[cam_type]) return true;
if (s->max_waiting > 1 && s->encoders_ready != s->max_waiting) {
// add a small margin to the start frame id in case one of the encoders already dropped the next frame
update_max_atomic(s->start_frame_id, frame_id + 2);
if (std::exchange(s->camera_ready[cam_type], true) == false) {
++s->encoders_ready;
LOGD("camera %d encoder ready", cam_type);
}
return false;
} else {
if (s->max_waiting == 1) update_max_atomic(s->start_frame_id, frame_id);
bool synced = frame_id >= s->start_frame_id;
s->camera_synced[cam_type] = synced;
if (!synced) LOGD("camera %d waiting for frame %d, cur %d", cam_type, (int)s->start_frame_id, frame_id);
return synced;
}
}
// Push the adaptive bitrate (set by webrtcd's LivestreamBitrateController) to the encoders.
static void apply_livestream_bitrate(std::vector<std::unique_ptr<Encoder>> &encoders) {
static Params params;
std::string val = params.get("LivestreamEncoderBitrate");
if (val.empty()) return;
try {
int bitrate = std::stoi(val);
for (auto &e : encoders) e->set_bitrate(bitrate);
} catch (const std::exception &e) {
LOGE("bad LivestreamEncoderBitrate value '%s': %s", val.c_str(), e.what());
}
}
// Force a keyframe when webrtcd requests one (fast first frame, camera switch, loss recovery).
// webrtcd clears the param once the client has actually received a keyframe, so this keeps
// requesting until an IDR lands rather than firing once and hoping it wasn't dropped.
static void apply_keyframe_request(std::vector<std::unique_ptr<Encoder>> &encoders) {
static Params params;
if (!params.getBool("LivestreamRequestKeyframe")) return;
for (auto &e : encoders) e->request_keyframe();
}
void encoder_thread(EncoderdState *s, const LogCameraInfo &cam_info) {
util::set_thread_name(cam_info.thread_name);
const bool has_adaptive = std::any_of(cam_info.encoder_infos.begin(), cam_info.encoder_infos.end(),
[](const auto &ei) { return ei.adaptive_bitrate; });
std::vector<std::unique_ptr<Encoder>> encoders;
VisionIpcClient vipc_client = VisionIpcClient("camerad", cam_info.stream_type, false);
std::unique_ptr<JpegEncoder> jpeg_encoder;
int cur_seg = 0;
while (!do_exit) {
if (!vipc_client.connect(false)) {
util::sleep_for(5);
continue;
}
// init encoders
if (encoders.empty()) {
const VisionBuf &buf_info = vipc_client.buffers[0];
LOGW("encoder %s init %zux%zu", cam_info.thread_name, buf_info.width, buf_info.height);
assert(buf_info.width > 0 && buf_info.height > 0);
for (const auto &encoder_info : cam_info.encoder_infos) {
auto &e = encoders.emplace_back(new Encoder(encoder_info, buf_info.width, buf_info.height));
e->encoder_open();
}
// Only one thumbnail can be generated per camera stream
if (auto thumbnail_name = cam_info.encoder_infos[0].thumbnail_name) {
jpeg_encoder = std::make_unique<JpegEncoder>(thumbnail_name, buf_info.width / 4, buf_info.height / 4);
}
}
bool lagging = false;
while (!do_exit) {
VisionIpcBufExtra extra;
VisionBuf* buf = vipc_client.recv(&extra);
if (buf == nullptr) continue;
// detect loop around and drop the frames
if (buf->get_frame_id() != extra.frame_id) {
if (!lagging) {
LOGE("encoder %s lag buffer id: %" PRIu64 " extra id: %d", cam_info.thread_name, buf->get_frame_id(), extra.frame_id);
lagging = true;
}
continue;
}
lagging = false;
if (!sync_encoders(s, cam_info.stream_type, extra.frame_id)) {
continue;
}
if (do_exit) break;
// adaptive bitrate + on-demand keyframes: pick up the latest requests from webrtcd
if (has_adaptive) {
apply_livestream_bitrate(encoders);
apply_keyframe_request(encoders);
}
// do rotation if required
const int frames_per_seg = SEGMENT_LENGTH * MAIN_FPS;
if (cur_seg >= 0 && extra.frame_id >= ((cur_seg + 1) * frames_per_seg) + s->start_frame_id) {
for (auto &e : encoders) {
e->encoder_close();
e->encoder_open();
}
++cur_seg;
}
// encode a frame
for (int i = 0; i < encoders.size(); ++i) {
int out_id = encoders[i]->encode_frame(buf, &extra);
if (out_id == -1) {
LOGE("Failed to encode frame. frame_id: %d", extra.frame_id);
}
}
if (jpeg_encoder && (extra.frame_id % 1200 == 100)) {
jpeg_encoder->pushThumbnail(buf, extra);
}
}
}
}
template <size_t N>
void encoderd_thread(const LogCameraInfo (&cameras)[N]) {
EncoderdState s;
std::set<VisionStreamType> streams;
while (!do_exit) {
streams = VisionIpcClient::getAvailableStreams("camerad", false);
if (!streams.empty()) {
break;
}
util::sleep_for(100);
}
if (!streams.empty()) {
std::vector<std::thread> encoder_threads;
for (auto stream : streams) {
auto it = std::find_if(std::begin(cameras), std::end(cameras),
[stream](auto &cam) { return cam.stream_type == stream; });
assert(it != std::end(cameras));
++s.max_waiting;
encoder_threads.push_back(std::thread(encoder_thread, &s, *it));
}
for (auto &t : encoder_threads) t.join();
}
}
int main(int argc, char* argv[]) {
if (!Hardware::PC()) {
int ret;
ret = util::set_realtime_priority(52);
assert(ret == 0);
ret = util::set_core_affinity({3});
assert(ret == 0);
}
if (argc > 1) {
std::string arg1(argv[1]);
if (arg1 == "--stream") {
encoderd_thread(stream_cameras_logged);
} else {
LOGE("Argument '%s' is not supported", arg1.c_str());
}
} else {
encoderd_thread(cameras_logged);
}
return 0;
}

View File

@@ -0,0 +1,83 @@
#pragma once
#include <cstdio>
#include <cstdlib>
#include <ctime>
#include <sys/types.h>
#include <fcntl.h>
#include <unistd.h>
// Bounded write-behind for loggerd's append-only FILE* (rlog/qlog/video): async
// writeback of new bytes (sync_file_range) + drop the now-clean prior range
// (posix_fadvise DONTNEED) so dirty pages don't pile up and force reclaim stalls.
// Never fsyncs, never drops dirty pages. Linux-only, no-op elsewhere, best-effort.
// Disable: LOGGERD_NO_WRITEBACK_TUNING=1. synced/dropped are caller-owned cursors.
static inline void stream_writeback(FILE *f, off_t &synced, off_t &dropped,
off_t chunk = (off_t)4 << 20) {
#if defined(__linux__)
static const bool disabled = getenv("LOGGERD_NO_WRITEBACK_TUNING") != nullptr;
if (disabled || f == nullptr) return;
off_t pos = ftello(f);
if (pos < 0 || pos - synced < chunk) return;
if (fflush(f) != 0) return;
int fd = fileno(f);
sync_file_range(fd, synced, pos - synced, SYNC_FILE_RANGE_WRITE);
if (synced > dropped) {
posix_fadvise(fd, dropped, synced - dropped, POSIX_FADV_DONTNEED);
dropped = synced;
}
synced = pos;
#else
(void)f; (void)synced; (void)dropped; (void)chunk;
#endif
}
static inline double writeback_now_ms() {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return ts.tv_sec * 1000.0 + ts.tv_nsec / 1e6;
}
static inline int durable_fsync(int fd) {
#if defined(__APPLE__)
return fsync(fd);
#else
return fdatasync(fd);
#endif
}
// Durable checkpoint on a caller-owned time cursor: at most once per interval_ms
// (always when force), commit data AND inode size through the journal so every
// byte written before the checkpoint survives a power cut. last_ms == 0 means
// never synced. Returns true if a sync was issued.
static inline bool fd_durable(int fd, double &last_ms, bool force, double interval_ms = 2000.0) {
if (fd < 0) return false;
double now = writeback_now_ms();
if (!force && last_ms != 0.0 && now - last_ms < interval_ms) return false;
durable_fsync(fd);
last_ms = now;
return true;
}
// FILE* variant: flush stdio buffers first.
static inline bool stream_durable(FILE *f, double &last_ms, bool force, double interval_ms = 2000.0) {
if (f == nullptr) return false;
double now = writeback_now_ms();
if (!force && last_ms != 0.0 && now - last_ms < interval_ms) return false;
if (fflush(f) != 0) return false;
durable_fsync(fileno(f));
last_ms = now;
return true;
}
// fsync a directory so entries for newly created files survive a power cut.
static inline void fsync_dir(const char *path) {
int fd = open(path, O_RDONLY);
if (fd >= 0) {
fsync(fd);
close(fd);
}
}

210
system/loggerd/logger.cc Normal file
View File

@@ -0,0 +1,210 @@
// _GNU_SOURCE for sync_file_range() (io_writeback.h); must precede all includes
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include "system/loggerd/logger.h"
#include <fstream>
#include <map>
#include <vector>
#include <iostream>
#include <sstream>
#include <random>
#include "common/params.h"
#include "common/swaglog.h"
#include "system/loggerd/io_writeback.h"
#include "iqpilot/common/version.h"
// ***** log metadata *****
kj::Array<capnp::word> logger_build_init_data() {
uint64_t wall_time = nanos_since_epoch();
MessageBuilder msg;
auto init = msg.initEvent().initInitData();
init.setWallTimeNanos(wall_time);
init.setVersion(IQPILOT_VERSION);
init.setDirty(!getenv("CLEAN"));
init.setDeviceType(Hardware::get_device_type());
// log kernel args
std::ifstream cmdline_stream("/proc/cmdline");
std::vector<std::string> kernel_args;
std::string buf;
while (cmdline_stream >> buf) {
kernel_args.push_back(buf);
}
auto lkernel_args = init.initKernelArgs(kernel_args.size());
for (int i=0; i<kernel_args.size(); i++) {
lkernel_args.set(i, kernel_args[i]);
}
init.setKernelVersion(util::read_file("/proc/version"));
init.setOsVersion(util::read_file("/VERSION"));
// log params
Params params(util::getenv("PARAMS_COPY_PATH", ""));
std::map<std::string, std::string> params_map = params.readAll();
init.setGitCommit(params_map["GitCommit"]);
init.setGitCommitDate(params_map["GitCommitDate"]);
init.setGitBranch(params_map["GitBranch"]);
init.setGitRemote(params_map["GitRemote"]);
init.setPassive(false);
init.setDongleId(params_map["DongleId"]);
// for prebuilt branches
init.setGitSrcCommit(util::read_file("../../git_src_commit"));
init.setGitSrcCommitDate(util::read_file("../../git_src_commit_date"));
auto lparams = init.initParams().initEntries(params_map.size());
int j = 0;
for (auto& [key, value] : params_map) {
auto lentry = lparams[j];
lentry.setKey(key);
if ( !(params.getKeyFlag(key) & DONT_LOG) ) {
lentry.setValue(capnp::Data::Reader((const kj::byte*)value.data(), value.size()));
}
j++;
}
// log commands
std::vector<std::string> log_commands = {
"df -h", // usage for all filesystems
};
auto hw_logs = Hardware::get_init_logs();
auto commands = init.initCommands().initEntries(log_commands.size() + hw_logs.size());
for (int i = 0; i < log_commands.size(); i++) {
auto lentry = commands[i];
lentry.setKey(log_commands[i]);
const std::string result = util::check_output(log_commands[i]);
lentry.setValue(capnp::Data::Reader((const kj::byte*)result.data(), result.size()));
}
int i = log_commands.size();
for (auto &[key, value] : hw_logs) {
auto lentry = commands[i];
lentry.setKey(key);
lentry.setValue(capnp::Data::Reader((const kj::byte*)value.data(), value.size()));
i++;
}
return capnp::messageToFlatArray(msg);
}
std::string logger_get_identifier(std::string key) {
// a log identifier is a 32 bit counter, plus a 10 character unique ID.
// e.g. 000001a3--c20ba54385
Params params;
uint32_t cnt;
try {
cnt = std::stoul(params.get(key));
} catch (std::exception &e) {
cnt = 0;
}
params.put(key, std::to_string(cnt + 1));
std::stringstream ss;
std::random_device rd;
std::mt19937 mt(rd());
std::uniform_int_distribution<int> dist(0, 15);
for (int i = 0; i < 10; ++i) {
ss << std::hex << dist(mt);
}
return util::string_format("%08x--%s", cnt, ss.str().c_str());
}
std::string zstd_decompress(const std::string &in) {
ZSTD_DCtx *dctx = ZSTD_createDCtx();
assert(dctx != nullptr);
// Initialize input and output buffers
ZSTD_inBuffer input = {in.data(), in.size(), 0};
// Estimate and reserve memory for decompressed data
size_t estimatedDecompressedSize = ZSTD_getFrameContentSize(in.data(), in.size());
if (estimatedDecompressedSize == ZSTD_CONTENTSIZE_ERROR || estimatedDecompressedSize == ZSTD_CONTENTSIZE_UNKNOWN) {
estimatedDecompressedSize = in.size() * 2; // Use a fallback size
}
std::string decompressedData;
decompressedData.reserve(estimatedDecompressedSize);
const size_t bufferSize = ZSTD_DStreamOutSize(); // Recommended output buffer size
std::string outputBuffer(bufferSize, '\0');
while (input.pos < input.size) {
ZSTD_outBuffer output = {outputBuffer.data(), bufferSize, 0};
size_t result = ZSTD_decompressStream(dctx, &output, &input);
if (ZSTD_isError(result)) {
break;
}
decompressedData.append(outputBuffer.data(), output.pos);
}
ZSTD_freeDCtx(dctx);
decompressedData.shrink_to_fit();
return decompressedData;
}
static void log_sentinel(LoggerState *log, SentinelType type, int exit_signal = 0) {
MessageBuilder msg;
auto sen = msg.initEvent().initSentinel();
sen.setType(type);
sen.setSignal(exit_signal);
log->write(msg.toBytes(), true);
}
LoggerState::LoggerState(const std::string &log_root) {
route_name = logger_get_identifier("RouteCount");
route_path = log_root + "/" + route_name;
init_data = logger_build_init_data();
}
LoggerState::~LoggerState() {
if (rlog) {
log_sentinel(this, SentinelType::END_OF_ROUTE, exit_signal);
std::remove(lock_file.c_str());
}
}
bool LoggerState::next() {
if (rlog) {
log_sentinel(this, SentinelType::END_OF_SEGMENT);
std::remove(lock_file.c_str());
}
segment_path = route_path + "--" + std::to_string(++part);
bool ret = util::create_directories(segment_path, 0775);
assert(ret == true);
lock_file = segment_path + "/rlog.lock";
std::ofstream{lock_file};
rlog.reset(new ZstdFileWriter(segment_path + "/rlog.zst", LOG_COMPRESSION_LEVEL));
qlog.reset(new ZstdFileWriter(segment_path + "/qlog.zst", LOG_COMPRESSION_LEVEL));
fsync_dir(segment_path.c_str());
// log init data & sentinel type.
write(init_data.asBytes(), true);
log_sentinel(this, part > 0 ? SentinelType::START_OF_SEGMENT : SentinelType::START_OF_ROUTE);
return true;
}
void LoggerState::write(uint8_t* data, size_t size, bool in_qlog) {
rlog->write(data, size);
if (in_qlog) qlog->write(data, size);
}

41
system/loggerd/logger.h Normal file
View File

@@ -0,0 +1,41 @@
#pragma once
#include <cassert>
#include <memory>
#include <string>
#include "cereal/messaging/messaging.h"
#include "common/util.h"
#include "system/hardware/hw.h"
#include "system/loggerd/zstd_writer.h"
constexpr int LOG_COMPRESSION_LEVEL = 10;
typedef cereal::Sentinel::SentinelType SentinelType;
class LoggerState {
public:
LoggerState(const std::string& log_root = Path::log_root());
~LoggerState();
bool next();
void write(uint8_t* data, size_t size, bool in_qlog);
inline int segment() const { return part; }
inline const std::string& segmentPath() const { return segment_path; }
inline const std::string& routeName() const { return route_name; }
inline void write(kj::ArrayPtr<kj::byte> bytes, bool in_qlog) { write(bytes.begin(), bytes.size(), in_qlog); }
inline void setExitSignal(int signal) { exit_signal = signal; }
inline void durableSync(bool force) {
if (rlog) rlog->durable_flush(force);
if (qlog) qlog->durable_flush(force);
}
protected:
int part = -1, exit_signal = 0;
std::string route_path, route_name, segment_path, lock_file;
kj::Array<capnp::word> init_data;
std::unique_ptr<ZstdFileWriter> rlog, qlog;
};
kj::Array<capnp::word> logger_build_init_data();
std::string logger_get_identifier(std::string key);
std::string zstd_decompress(const std::string &in);

403
system/loggerd/loggerd.cc Normal file
View File

@@ -0,0 +1,403 @@
#include <sys/xattr.h>
#include <map>
#include <memory>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
#include "common/params.h"
#include "system/loggerd/encoder/encoder.h"
#include "system/loggerd/loggerd.h"
#include "system/loggerd/video_writer.h"
ExitHandler do_exit;
// Set by the power watcher when the harness 12V input collapses. The main loop
// reacts by force-syncing every open file so the tail survives if the
// supercaps/battery run out before a clean shutdown.
static std::atomic<bool> power_lost{false};
// Harness VIN via the INA231 (same source as Hardware::get_voltage). The PMIC
// never sees the harness input on comma 3/3X, so power_supply uevents are
// useless here; poll the ADC instead. Only arms after power has been seen good
// once, so bench setups without harness power never trigger it.
static void power_loss_watcher() {
if (Hardware::PC()) return;
bool armed = false;
while (!do_exit) {
int mv = Hardware::get_voltage();
if (mv > 5000) {
if (power_lost.exchange(false)) {
LOGW("harness power restored (%d mV), resuming normal flush cadence", mv);
}
armed = true;
} else if (armed && mv < 4000 && !power_lost.exchange(true)) {
LOGE("harness power lost (%d mV), forcing durable flush of all logs", mv);
}
util::sleep_for(50);
}
}
struct LoggerdState {
LoggerState logger;
std::atomic<double> last_camera_seen_tms{0.0};
std::atomic<int> ready_to_rotate{0}; // count of encoders ready to rotate
int max_waiting = 0;
double last_rotate_tms = 0.; // last rotate time in ms
};
void logger_rotate(LoggerdState *s) {
bool ret =s->logger.next();
assert(ret);
s->ready_to_rotate = 0;
s->last_rotate_tms = millis_since_boot();
LOGW((s->logger.segment() == 0) ? "logging to %s" : "rotated to %s", s->logger.segmentPath().c_str());
}
void rotate_if_needed(LoggerdState *s) {
// all encoders ready, trigger rotation
bool all_ready = s->ready_to_rotate == s->max_waiting;
// fallback logic to prevent extremely long segments in the case of camera, encoder, etc. malfunctions
bool timed_out = false;
double tms = millis_since_boot();
double seg_length_secs = (tms - s->last_rotate_tms) / 1000.;
if ((seg_length_secs > SEGMENT_LENGTH) && !LOGGERD_TEST) {
// TODO: might be nice to put these reasons in the sentinel
if ((tms - s->last_camera_seen_tms) > NO_CAMERA_PATIENCE) {
timed_out = true;
LOGE("no camera packets seen. auto rotating");
} else if (seg_length_secs > SEGMENT_LENGTH*1.2) {
timed_out = true;
LOGE("segment too long. auto rotating");
}
}
if (all_ready || timed_out) {
logger_rotate(s);
}
}
struct RemoteEncoder {
std::unique_ptr<VideoWriter> writer;
int encoderd_segment_offset;
int current_segment = -1;
std::vector<Message *> q;
int dropped_frames = 0;
bool recording = false;
bool marked_ready_to_rotate = false;
bool seen_first_packet = false;
bool audio_initialized = false;
};
size_t write_encode_data(LoggerdState *s, cereal::Event::Reader event, RemoteEncoder &re, const EncoderInfo &encoder_info) {
auto edata = (event.*(encoder_info.get_encode_data_func))();
auto idx = edata.getIdx();
auto flags = idx.getFlags();
// if we aren't recording yet, try to start, since we are in the correct segment
if (!re.recording) {
if (flags & V4L2_BUF_FLAG_KEYFRAME) {
// only create on iframe
if (re.dropped_frames) {
// this should only happen for the first segment, maybe
LOGW("%s: dropped %d non iframe packets before init", encoder_info.publish_name, re.dropped_frames);
re.dropped_frames = 0;
}
if (encoder_info.record) {
// write the header
auto header = edata.getHeader();
re.writer->write((uint8_t *)header.begin(), header.size(), idx.getTimestampEof() / 1000, true, false);
}
re.recording = true;
} else {
// this is a sad case when we aren't recording, but don't have an iframe
// nothing we can do but drop the frame
++re.dropped_frames;
return 0;
}
}
// we have to be recording if we are here
assert(re.recording);
// if we are actually writing the video file, do so
if (re.writer) {
auto data = edata.getData();
re.writer->write((uint8_t *)data.begin(), data.size(), idx.getTimestampEof() / 1000, false, flags & V4L2_BUF_FLAG_KEYFRAME);
}
// put it in log stream as the idx packet
MessageBuilder bmsg;
auto evt = bmsg.initEvent(event.getValid());
evt.setLogMonoTime(event.getLogMonoTime());
(evt.*(encoder_info.set_encode_idx_func))(idx);
auto new_msg = bmsg.toBytes();
s->logger.write((uint8_t *)new_msg.begin(), new_msg.size(), true); // always in qlog?
return new_msg.size();
}
int handle_encoder_msg(LoggerdState *s, Message *msg, std::string &name, struct RemoteEncoder &re, const EncoderInfo &encoder_info) {
int bytes_count = 0;
// extract the message
capnp::FlatArrayMessageReader cmsg(kj::ArrayPtr<capnp::word>((capnp::word *)msg->getData(), msg->getSize() / sizeof(capnp::word)));
auto event = cmsg.getRoot<cereal::Event>();
auto edata = (event.*(encoder_info.get_encode_data_func))();
auto idx = edata.getIdx();
// encoderd can have started long before loggerd
if (!re.seen_first_packet) {
re.seen_first_packet = true;
re.encoderd_segment_offset = idx.getSegmentNum();
LOGD("%s: has encoderd offset %d", name.c_str(), re.encoderd_segment_offset);
}
int offset_segment_num = idx.getSegmentNum() - re.encoderd_segment_offset;
if (offset_segment_num == s->logger.segment()) {
// loggerd is now on the segment that matches this packet
// if this is a new segment, we close any possible old segments, move to the new, and process any queued packets
if (re.current_segment != s->logger.segment()) {
// if we aren't actually recording, don't create the writer
if (encoder_info.record) {
assert(encoder_info.filename != NULL);
re.writer.reset(new VideoWriter(s->logger.segmentPath().c_str(),
encoder_info.filename, idx.getType() != cereal::EncodeIndex::Type::FULL_H_E_V_C,
edata.getWidth(), edata.getHeight(), encoder_info.fps, idx.getType()));
re.recording = false;
re.audio_initialized = false;
}
re.current_segment = s->logger.segment();
re.marked_ready_to_rotate = false;
}
if (re.audio_initialized || !encoder_info.include_audio) {
// we are in this segment now, process any queued messages before this one
if (!re.q.empty()) {
for (auto qmsg : re.q) {
capnp::FlatArrayMessageReader reader({(capnp::word *)qmsg->getData(), qmsg->getSize() / sizeof(capnp::word)});
bytes_count += write_encode_data(s, reader.getRoot<cereal::Event>(), re, encoder_info);
delete qmsg;
}
re.q.clear();
}
bytes_count += write_encode_data(s, event, re, encoder_info);
delete msg;
} else if (re.q.size() > MAIN_FPS*10) {
LOGE_100("%s: dropping frame waiting for audio initialization, queue is too large", name.c_str());
delete msg;
} else {
re.q.push_back(msg); // queue up all the new segment messages, they go in after audio is initialized
}
} else if (offset_segment_num > s->logger.segment()) {
// encoderd packet has a newer segment, this means encoderd has rolled over
if (!re.marked_ready_to_rotate) {
re.marked_ready_to_rotate = true;
++s->ready_to_rotate;
LOGD("rotate %d -> %d ready %d/%d for %s",
s->logger.segment(), offset_segment_num,
s->ready_to_rotate.load(), s->max_waiting, name.c_str());
}
// TODO: define this behavior, but for now don't leak
if (re.q.size() > MAIN_FPS*10) {
LOGE_100("%s: dropping frame, queue is too large", name.c_str());
delete msg;
} else {
// queue up all the new segment messages, they go in after the rotate
re.q.push_back(msg);
}
} else {
LOGE("%s: encoderd packet has a older segment!!! idx.getSegmentNum():%d s->logger.segment():%d re.encoderd_segment_offset:%d",
name.c_str(), idx.getSegmentNum(), s->logger.segment(), re.encoderd_segment_offset);
// free the message, it's useless. this should never happen
// actually, this can happen if you restart encoderd
re.encoderd_segment_offset = -s->logger.segment();
delete msg;
}
return bytes_count;
}
void handle_preserve_segment(LoggerdState *s) {
static int prev_segment = -1;
if (s->logger.segment() == prev_segment) return;
LOGW("preserving %s", s->logger.segmentPath().c_str());
#ifdef __APPLE__
int ret = setxattr(s->logger.segmentPath().c_str(), PRESERVE_ATTR_NAME, &PRESERVE_ATTR_VALUE, 1, 0, 0);
#else
int ret = setxattr(s->logger.segmentPath().c_str(), PRESERVE_ATTR_NAME, &PRESERVE_ATTR_VALUE, 1, 0);
#endif
if (ret) {
LOGE("setxattr %s failed for %s: %s", PRESERVE_ATTR_NAME, s->logger.segmentPath().c_str(), strerror(errno));
}
// mark route for uploading
Params params;
std::string routes = params.get("AthenadRecentlyViewedRoutes");
params.put("AthenadRecentlyViewedRoutes", routes + "," + s->logger.routeName());
prev_segment = s->logger.segment();
}
void loggerd_thread() {
// setup messaging
typedef struct ServiceState {
std::string name;
int counter, freq;
bool encoder, preserve_segment, record_audio;
} ServiceState;
std::unordered_map<SubSocket*, ServiceState> service_state;
std::unordered_map<SubSocket*, struct RemoteEncoder> remote_encoders;
std::unique_ptr<Context> ctx(Context::create());
std::unique_ptr<Poller> poller(Poller::create());
// subscribe to all socks
for (const auto& [_, it] : services) {
const bool encoder = util::ends_with(it.name, "EncodeData");
const bool livestream_encoder = util::starts_with(it.name, "livestream");
const bool record_audio = (it.name == "rawAudioData") && Params().getBool("RecordAudio");
if (it.should_log || (encoder && !livestream_encoder) || record_audio) {
LOGD("logging %s", it.name.c_str());
SubSocket * sock = SubSocket::create(ctx.get(), it.name, "127.0.0.1", false, true, it.queue_size);
assert(sock != NULL);
poller->registerSocket(sock);
service_state[sock] = {
.name = it.name,
.counter = 0,
.freq = it.decimation,
.encoder = encoder,
.preserve_segment = (it.name == "userBookmark") || (it.name == "audioFeedback"),
.record_audio = record_audio,
};
}
}
LoggerdState s;
// init logger
logger_rotate(&s);
Params().put("CurrentRoute", s.logger.routeName());
std::map<std::string, EncoderInfo> encoder_infos_dict;
std::vector<RemoteEncoder*> encoders_with_audio;
for (const auto &cam : cameras_logged) {
for (const auto &encoder_info : cam.encoder_infos) {
encoder_infos_dict[encoder_info.publish_name] = encoder_info;
s.max_waiting++;
}
}
for (auto &[sock, service] : service_state) {
auto it = encoder_infos_dict.find(service.name);
if (it != encoder_infos_dict.end() && it->second.include_audio) {
encoders_with_audio.push_back(&remote_encoders[sock]);
}
}
std::thread power_thread(power_loss_watcher);
uint64_t msg_count = 0, bytes_count = 0;
double start_ts = millis_since_boot();
double last_panic_flush_ms = 0.;
while (!do_exit) {
// poll for new messages on all sockets
for (auto sock : poller->poll(1000)) {
if (do_exit) break;
ServiceState &service = service_state[sock];
if (service.preserve_segment) {
handle_preserve_segment(&s);
}
// drain socket
int count = 0;
Message *msg = nullptr;
while (!do_exit && (msg = sock->receive(true))) {
const bool in_qlog = service.freq != -1 && (service.counter++ % service.freq == 0);
if (service.record_audio) {
capnp::FlatArrayMessageReader cmsg(kj::ArrayPtr<capnp::word>((capnp::word *)msg->getData(), msg->getSize() / sizeof(capnp::word)));
auto event = cmsg.getRoot<cereal::Event>();
auto audio_data = event.getRawAudioData().getData();
auto sample_rate = event.getRawAudioData().getSampleRate();
for (auto* encoder : encoders_with_audio) {
if (encoder && encoder->writer) {
encoder->writer->write_audio((uint8_t*)audio_data.begin(), audio_data.size(), event.getLogMonoTime() / 1000, sample_rate);
encoder->audio_initialized = true;
}
}
}
if (service.encoder) {
s.last_camera_seen_tms = millis_since_boot();
bytes_count += handle_encoder_msg(&s, msg, service.name, remote_encoders[sock], encoder_infos_dict[service.name]);
} else {
s.logger.write((uint8_t *)msg->getData(), msg->getSize(), in_qlog);
bytes_count += msg->getSize();
delete msg;
}
rotate_if_needed(&s);
if ((++msg_count % 10000) == 0) {
double seconds = (millis_since_boot() - start_ts) / 1000.0;
LOGD("%" PRIu64 " messages, %.2f msg/sec, %.2f KB/sec", msg_count, msg_count / seconds, bytes_count * 0.001 / seconds);
}
count++;
if (count >= 200) {
LOGD("large volume of '%s' messages", service.name.c_str());
break;
}
}
}
// harness power gone: everything already written must hit storage now, and
// keep hitting it while we're running on residual charge
if (power_lost) {
double now = millis_since_boot();
if (now - last_panic_flush_ms > 450.) {
for (auto &[sock, re] : remote_encoders) {
if (re.writer) re.writer->durable_sync(true);
}
s.logger.durableSync(true);
last_panic_flush_ms = now;
}
}
}
LOGW("closing logger");
s.logger.setExitSignal(do_exit.signal);
if (do_exit.power_failure) {
LOGE("power failure");
sync();
LOGE("sync done");
}
// messaging cleanup
for (auto &[sock, service] : service_state) delete sock;
power_thread.join();
}
int main(int argc, char** argv) {
if (!Hardware::PC()) {
int ret;
ret = util::set_core_affinity({0, 1, 2, 3});
assert(ret == 0);
// TODO: why does this impact camerad timings?
//ret = util::set_realtime_priority(1);
//assert(ret == 0);
}
loggerd_thread();
return 0;
}

198
system/loggerd/loggerd.h Normal file
View File

@@ -0,0 +1,198 @@
#pragma once
#include <cstdlib>
#include <vector>
#include "cereal/messaging/messaging.h"
#include "cereal/services.h"
#include "msgq/visionipc/visionipc_client.h"
#include "system/hardware/hw.h"
#include "common/params.h"
#include "common/swaglog.h"
#include "common/util.h"
#include "system/loggerd/logger.h"
constexpr int MAIN_FPS = 20;
const auto MAIN_ENCODE_TYPE = Hardware::PC() ? cereal::EncodeIndex::Type::BIG_BOX_LOSSLESS : cereal::EncodeIndex::Type::FULL_H_E_V_C;
#define NO_CAMERA_PATIENCE 500 // fall back to time-based rotation if all cameras are dead
#define INIT_ENCODE_FUNCTIONS(encode_type) \
.get_encode_data_func = &cereal::Event::Reader::get##encode_type##Data, \
.set_encode_idx_func = &cereal::Event::Builder::set##encode_type##Idx, \
.init_encode_data_func = &cereal::Event::Builder::init##encode_type##Data
const bool LOGGERD_TEST = getenv("LOGGERD_TEST");
const int SEGMENT_LENGTH = LOGGERD_TEST ? atoi(getenv("LOGGERD_SEGMENT_LENGTH")) : 60;
constexpr char PRESERVE_ATTR_NAME[] = "user.preserve";
constexpr char PRESERVE_ATTR_VALUE = '1';
constexpr int QCAM_WIDTH = 1316;
constexpr int QCAM_HEIGHT = 826;
constexpr int MICI_QCAM_WIDTH = 1210;
constexpr int MICI_QCAM_HEIGHT = 760;
static_assert(QCAM_WIDTH % 2 == 0 && QCAM_HEIGHT % 2 == 0 &&
MICI_QCAM_WIDTH % 2 == 0 && MICI_QCAM_HEIGHT % 2 == 0,
"qcamera dimensions must be even");
inline bool is_mici() {
return Hardware::get_device_type() == cereal::InitData::DeviceType::MICI;
}
struct EncoderSettings {
cereal::EncodeIndex::Type encode_type;
int bitrate;
int gop_size;
int b_frames = 0; // we don't use b frames
static EncoderSettings MainEncoderSettings(int in_width) {
if (in_width <= 1344) {
return EncoderSettings{.encode_type = MAIN_ENCODE_TYPE, .bitrate = 5'000'000, .gop_size = 20};
} else {
return EncoderSettings{.encode_type = MAIN_ENCODE_TYPE, .bitrate = 10'000'000, .gop_size = 30};
}
}
static EncoderSettings QcamEncoderSettings() {
// Keep Konn3kt route uploads useful without turning each 60-second qcamera
// segment into a 26 MB file. This is still a substantial improvement over
// stock qcam, while CBR keeps storage and upload usage predictable.
int _qcam_bitrate = getenv("QCAM_BITRATE") ? atoi(getenv("QCAM_BITRATE")) : 2'400'000;
return EncoderSettings{.encode_type = cereal::EncodeIndex::Type::QCAMERA_H264, .bitrate = _qcam_bitrate, .gop_size = 15};
}
static EncoderSettings StreamEncoderSettings() {
// Startup bitrate for the livestream encoder. The adaptive bitrate controller
// (webrtcd LivestreamBitrateController) overrides this at runtime via the
// LivestreamEncoderBitrate param based on measured RTCP loss/RTT. Start at a
// conservative mid level so a poor link doesn't choke the first seconds, then ramp up.
int _stream_bitrate = getenv("STREAM_BITRATE") ? atoi(getenv("STREAM_BITRATE")) : 1'500'000;
return EncoderSettings{.encode_type = cereal::EncodeIndex::Type::QCAMERA_H264, .bitrate = _stream_bitrate , .gop_size = 30};
}
};
class EncoderInfo {
public:
const char *publish_name;
const char *thumbnail_name = NULL;
const char *filename = NULL;
bool record = true;
bool include_audio = false;
bool adaptive_bitrate = false; // allow runtime bitrate updates (livestream ABR)
bool cbr = false; // force constant bitrate so file size = bitrate * duration (qcamera)
int frame_width = -1;
int frame_height = -1;
int fps = MAIN_FPS;
std::function<EncoderSettings(int)> get_settings;
::cereal::EncodeData::Reader (cereal::Event::Reader::*get_encode_data_func)() const;
void (cereal::Event::Builder::*set_encode_idx_func)(::cereal::EncodeIndex::Reader);
cereal::EncodeData::Builder (cereal::Event::Builder::*init_encode_data_func)();
};
class LogCameraInfo {
public:
const char *thread_name;
int fps = MAIN_FPS;
VisionStreamType stream_type;
std::vector<EncoderInfo> encoder_infos;
};
const EncoderInfo main_road_encoder_info = {
.publish_name = "roadEncodeData",
.thumbnail_name = "thumbnail",
.filename = "fcamera.hevc",
.get_settings = [](int in_width){return EncoderSettings::MainEncoderSettings(in_width);},
INIT_ENCODE_FUNCTIONS(RoadEncode),
};
const EncoderInfo main_wide_road_encoder_info = {
.publish_name = "wideRoadEncodeData",
.filename = "ecamera.hevc",
.get_settings = [](int in_width){return EncoderSettings::MainEncoderSettings(in_width);},
INIT_ENCODE_FUNCTIONS(WideRoadEncode),
};
const EncoderInfo main_driver_encoder_info = {
.publish_name = "driverEncodeData",
.filename = "dcamera.hevc",
.record = Params().getBool("RecordFront"),
.get_settings = [](int in_width){return EncoderSettings::MainEncoderSettings(in_width);},
INIT_ENCODE_FUNCTIONS(DriverEncode),
};
const EncoderInfo stream_road_encoder_info = {
.publish_name = "livestreamRoadEncodeData",
//.thumbnail_name = "thumbnail",
.record = false,
.adaptive_bitrate = true,
.get_settings = [](int){return EncoderSettings::StreamEncoderSettings();},
INIT_ENCODE_FUNCTIONS(LivestreamRoadEncode),
};
const EncoderInfo stream_wide_road_encoder_info = {
.publish_name = "livestreamWideRoadEncodeData",
.record = false,
.adaptive_bitrate = true,
.get_settings = [](int){return EncoderSettings::StreamEncoderSettings();},
INIT_ENCODE_FUNCTIONS(LivestreamWideRoadEncode),
};
const EncoderInfo stream_driver_encoder_info = {
.publish_name = "livestreamDriverEncodeData",
.record = false,
.adaptive_bitrate = true,
.get_settings = [](int){return EncoderSettings::StreamEncoderSettings();},
INIT_ENCODE_FUNCTIONS(LivestreamDriverEncode),
};
const EncoderInfo qcam_encoder_info = {
.publish_name = "qRoadEncodeData",
.filename = "qcamera.ts",
.cbr = true, // enforce the bitrate so upload size stays predictable (no VBR overshoot)
.get_settings = [](int){return EncoderSettings::QcamEncoderSettings();},
.frame_width = is_mici() ? MICI_QCAM_WIDTH : QCAM_WIDTH,
.frame_height = is_mici() ? MICI_QCAM_HEIGHT : QCAM_HEIGHT,
.include_audio = Params().getBool("RecordAudio"),
INIT_ENCODE_FUNCTIONS(QRoadEncode),
};
const LogCameraInfo road_camera_info{
.thread_name = "road_cam_encoder",
.stream_type = VISION_STREAM_ROAD,
.encoder_infos = {main_road_encoder_info, qcam_encoder_info}
};
const LogCameraInfo wide_road_camera_info{
.thread_name = "wide_road_cam_encoder",
.stream_type = VISION_STREAM_WIDE_ROAD,
.encoder_infos = {main_wide_road_encoder_info}
};
const LogCameraInfo driver_camera_info{
.thread_name = "driver_cam_encoder",
.stream_type = VISION_STREAM_DRIVER,
.encoder_infos = {main_driver_encoder_info}
};
const LogCameraInfo stream_road_camera_info{
.thread_name = "road_cam_encoder",
.stream_type = VISION_STREAM_ROAD,
.encoder_infos = {stream_road_encoder_info}
};
const LogCameraInfo stream_wide_road_camera_info{
.thread_name = "wide_road_cam_encoder",
.stream_type = VISION_STREAM_WIDE_ROAD,
.encoder_infos = {stream_wide_road_encoder_info}
};
const LogCameraInfo stream_driver_camera_info{
.thread_name = "driver_cam_encoder",
.stream_type = VISION_STREAM_DRIVER,
.encoder_infos = {stream_driver_encoder_info}
};
const LogCameraInfo cameras_logged[] = {road_camera_info, wide_road_camera_info, driver_camera_info};
const LogCameraInfo stream_cameras_logged[] = {stream_road_camera_info, stream_wide_road_camera_info, stream_driver_camera_info};

View File

View File

@@ -0,0 +1,90 @@
import os
import random
from pathlib import Path
import openpilot.system.loggerd.deleter as deleter
import konn3kt_private.uploaderd.iquploaderd as uploader
from openpilot.common.params import Params
from openpilot.system.hardware.hw import Paths
from openpilot.system.loggerd.xattr_cache import setxattr
def create_random_file(file_path: Path, size_mb: float, lock: bool = False, upload_xattr: bytes | None = None) -> None:
file_path.parent.mkdir(parents=True, exist_ok=True)
if lock:
lock_path = str(file_path) + ".lock"
os.close(os.open(lock_path, os.O_CREAT | os.O_EXCL))
chunks = 128
chunk_bytes = int(size_mb * 1024 * 1024 / chunks)
data = os.urandom(chunk_bytes)
with open(file_path, "wb") as f:
for _ in range(chunks):
f.write(data)
if upload_xattr is not None:
setxattr(str(file_path), uploader.UPLOAD_ATTR_NAME, upload_xattr)
class MockResponse:
def __init__(self, text, status_code):
self.text = text
self.status_code = status_code
class MockApi:
def __init__(self, dongle_id):
pass
def get(self, *args, **kwargs):
return MockResponse('{"url": "http://localhost/does/not/exist", "headers": {}}', 200)
def get_token(self):
return "fake-token"
class MockApiIgnore:
def __init__(self, dongle_id):
pass
def get(self, *args, **kwargs):
return MockResponse('', 412)
def get_token(self):
return "fake-token"
class UploaderTestCase:
f_type = "UNKNOWN"
root: Path
seg_num: int
seg_format: str
seg_format2: str
seg_dir: str
def set_ignore(self):
uploader.Api = MockApiIgnore
def setup_method(self):
uploader.Api = MockApi
uploader.fake_upload = True
uploader.force_wifi = True
uploader.allow_sleep = False
self.seg_num = random.randint(1, 300)
self.seg_format = "00000004--0ac3964c96--{}"
self.seg_format2 = "00000005--4c4e99b08b--{}"
self.seg_dir = self.seg_format.format(self.seg_num)
self.params = Params()
self.params.put("IsOffroad", True)
self.params.put("DongleId", "0000000000000000")
def make_file_with_data(self, f_dir: str, fn: str, size_mb: float = .1, lock: bool = False,
upload_xattr: bytes | None = None, preserve_xattr: bytes | None = None) -> Path:
file_path = Path(Paths.log_root()) / f_dir / fn
create_random_file(file_path, size_mb, lock, upload_xattr)
if preserve_xattr is not None:
setxattr(str(file_path.parent), deleter.PRESERVE_ATTR_NAME, preserve_xattr)
return file_path

View File

@@ -0,0 +1,117 @@
import time
import threading
from collections import namedtuple
from pathlib import Path
from collections.abc import Sequence
import openpilot.system.loggerd.deleter as deleter
from openpilot.common.timeout import Timeout, TimeoutException
from openpilot.system.loggerd.tests.loggerd_tests_common import UploaderTestCase
Stats = namedtuple("Stats", ['f_bavail', 'f_blocks', 'f_frsize'])
class TestDeleter(UploaderTestCase):
def fake_statvfs(self, d):
return self.fake_stats
def setup_method(self):
self.f_type = "fcamera.hevc"
super().setup_method()
self.fake_stats = Stats(f_bavail=0, f_blocks=10, f_frsize=4096)
deleter.os.statvfs = self.fake_statvfs
def start_thread(self):
self.end_event = threading.Event()
self.del_thread = threading.Thread(target=deleter.deleter_thread, args=[self.end_event])
self.del_thread.daemon = True
self.del_thread.start()
def join_thread(self):
self.end_event.set()
self.del_thread.join()
def test_delete(self):
f_path = self.make_file_with_data(self.seg_dir, self.f_type, 1)
self.start_thread()
try:
with Timeout(2, "Timeout waiting for file to be deleted"):
while f_path.exists():
time.sleep(0.01)
finally:
self.join_thread()
def assertDeleteOrder(self, f_paths: Sequence[Path], timeout: int = 5) -> None:
deleted_order = []
self.start_thread()
try:
with Timeout(timeout, "Timeout waiting for files to be deleted"):
while True:
for f in f_paths:
if not f.exists() and f not in deleted_order:
deleted_order.append(f)
if len(deleted_order) == len(f_paths):
break
time.sleep(0.01)
except TimeoutException:
print("Not deleted:", [f for f in f_paths if f not in deleted_order])
raise
finally:
self.join_thread()
assert deleted_order == f_paths, "Files not deleted in expected order"
def test_delete_order(self):
self.assertDeleteOrder([
self.make_file_with_data(self.seg_format.format(0), self.f_type),
self.make_file_with_data(self.seg_format.format(1), self.f_type),
self.make_file_with_data(self.seg_format2.format(0), self.f_type),
])
def test_delete_many_preserved(self):
self.assertDeleteOrder([
self.make_file_with_data(self.seg_format.format(0), self.f_type),
self.make_file_with_data(self.seg_format.format(1), self.f_type, preserve_xattr=deleter.PRESERVE_ATTR_VALUE),
self.make_file_with_data(self.seg_format.format(2), self.f_type),
] + [
self.make_file_with_data(self.seg_format2.format(i), self.f_type, preserve_xattr=deleter.PRESERVE_ATTR_VALUE)
for i in range(5)
])
def test_delete_last(self):
self.assertDeleteOrder([
self.make_file_with_data(self.seg_format.format(1), self.f_type),
self.make_file_with_data(self.seg_format2.format(0), self.f_type),
self.make_file_with_data(self.seg_format.format(0), self.f_type, preserve_xattr=deleter.PRESERVE_ATTR_VALUE),
self.make_file_with_data("boot", self.seg_format[:-4]),
self.make_file_with_data("crash", self.seg_format2[:-4]),
])
def test_no_delete_when_available_space(self):
f_path = self.make_file_with_data(self.seg_dir, self.f_type)
block_size = 4096
available = (10 * 1024 * 1024 * 1024) / block_size # 10GB free
self.fake_stats = Stats(f_bavail=available, f_blocks=10, f_frsize=block_size)
self.start_thread()
start_time = time.monotonic()
while f_path.exists() and time.monotonic() - start_time < 2:
time.sleep(0.01)
self.join_thread()
assert f_path.exists(), "File deleted with available space"
def test_no_delete_with_lock_file(self):
f_path = self.make_file_with_data(self.seg_dir, self.f_type, lock=True)
self.start_thread()
start_time = time.monotonic()
while f_path.exists() and time.monotonic() - start_time < 2:
time.sleep(0.01)
self.join_thread()
assert f_path.exists(), "File deleted when locked"

View File

@@ -0,0 +1,152 @@
import math
import os
import pytest
import random
import shutil
import subprocess
import time
from pathlib import Path
from parameterized import parameterized
from tqdm import trange
from openpilot.common.params import Params
from openpilot.common.timeout import Timeout
from openpilot.system.hardware import TICI
from openpilot.system.manager.process_config import managed_processes
from openpilot.tools.lib.logreader import LogReader
from openpilot.system.hardware.hw import Paths
SEGMENT_LENGTH = 2
FULL_SIZE = 2507572
def hevc_size(w): return FULL_SIZE // 2 if w <= 1344 else FULL_SIZE
CAMERAS = [
("fcamera.hevc", 20, hevc_size, "roadEncodeIdx"),
("dcamera.hevc", 20, hevc_size, "driverEncodeIdx"),
("ecamera.hevc", 20, hevc_size, "wideRoadEncodeIdx"),
("qcamera.ts", 20, lambda x: 130000, None),
]
# we check frame count, so we don't have to be too strict on size
FILE_SIZE_TOLERANCE = 0.7
@pytest.mark.tici # TODO: all of loggerd should work on PC
class TestEncoder:
def setup_method(self):
self._clear_logs()
os.environ["LOGGERD_TEST"] = "1"
os.environ["LOGGERD_SEGMENT_LENGTH"] = str(SEGMENT_LENGTH)
def teardown_method(self):
self._clear_logs()
def _clear_logs(self):
if os.path.exists(Paths.log_root()):
shutil.rmtree(Paths.log_root())
def _get_latest_segment_path(self):
last_route = sorted(Path(Paths.log_root()).iterdir())[-1]
return os.path.join(Paths.log_root(), last_route)
# TODO: this should run faster than real time
@parameterized.expand([(True, ), (False, )])
def test_log_rotation(self, record_front):
Params().put_bool("RecordFront", record_front)
managed_processes['sensord'].start()
managed_processes['loggerd'].start()
managed_processes['encoderd'].start()
time.sleep(1.0)
managed_processes['camerad'].start()
num_segments = int(os.getenv("SEGMENTS", random.randint(2, 8)))
# wait for loggerd to make the dir for first segment
route_prefix_path = None
with Timeout(int(SEGMENT_LENGTH*3)):
while route_prefix_path is None:
try:
route_prefix_path = self._get_latest_segment_path().rsplit("--", 1)[0]
except Exception:
time.sleep(0.1)
def check_seg(i):
# check each camera file size
counts = []
first_frames = []
for camera, fps, size_lambda, encode_idx_name in CAMERAS:
if not record_front and "dcamera" in camera:
continue
file_path = f"{route_prefix_path}--{i}/{camera}"
# check file exists
assert os.path.exists(file_path), f"segment #{i}: '{file_path}' missing"
# TODO: this ffprobe call is really slow
# get width and check frame count
cmd = f"ffprobe -v error -select_streams v:0 -count_packets -show_entries stream=nb_read_packets,width -of csv=p=0 {file_path}"
if TICI:
cmd = "LD_LIBRARY_PATH=/usr/local/lib " + cmd
expected_frames = fps * SEGMENT_LENGTH
probe = subprocess.check_output(cmd, shell=True, encoding='utf8').split('\n')[0].strip().split(',')
frame_width, frame_count = int(probe[0]), int(probe[1])
counts.append(frame_count)
assert frame_count == expected_frames, \
f"segment #{i}: {camera} failed frame count check: expected {expected_frames}, got {frame_count}"
# sanity check file size
file_size = os.path.getsize(file_path)
target_size = size_lambda(frame_width)
assert math.isclose(file_size, target_size, rel_tol=FILE_SIZE_TOLERANCE), \
f"{file_path} size {file_size} isn't close to target size {target_size}"
# Check encodeIdx
if encode_idx_name is not None:
rlog_path = f"{route_prefix_path}--{i}/rlog.zst"
msgs = [m for m in LogReader(rlog_path) if m.which() == encode_idx_name]
encode_msgs = [getattr(m, encode_idx_name) for m in msgs]
valid = [m.valid for m in msgs]
segment_idxs = [m.segmentId for m in encode_msgs]
encode_idxs = [m.encodeId for m in encode_msgs]
frame_idxs = [m.frameId for m in encode_msgs]
# Check frame count
assert frame_count == len(segment_idxs)
assert frame_count == len(encode_idxs)
# Check for duplicates or skips
assert 0 == segment_idxs[0]
assert len(set(segment_idxs)) == len(segment_idxs)
assert all(valid)
assert expected_frames * i == encode_idxs[0]
first_frames.append(frame_idxs[0])
assert len(set(encode_idxs)) == len(encode_idxs)
assert 1 == len(set(first_frames))
if TICI:
expected_frames = fps * SEGMENT_LENGTH
assert min(counts) == expected_frames
shutil.rmtree(f"{route_prefix_path}--{i}")
try:
for i in trange(num_segments):
# poll for next segment
with Timeout(int(SEGMENT_LENGTH*10), error_msg=f"timed out waiting for segment {i}"):
while Path(f"{route_prefix_path}--{i+1}") not in Path(Paths.log_root()).iterdir():
time.sleep(0.1)
check_seg(i)
finally:
managed_processes['loggerd'].stop()
managed_processes['encoderd'].stop()
managed_processes['camerad'].stop()
managed_processes['sensord'].stop()

View File

@@ -0,0 +1,75 @@
#include "catch2/catch.hpp"
#include "system/loggerd/logger.h"
typedef cereal::Sentinel::SentinelType SentinelType;
void verify_segment(const std::string &route_path, int segment, int max_segment, int required_event_cnt) {
const std::string segment_path = route_path + "--" + std::to_string(segment);
SentinelType begin_sentinel = segment == 0 ? SentinelType::START_OF_ROUTE : SentinelType::START_OF_SEGMENT;
SentinelType end_sentinel = segment == max_segment - 1 ? SentinelType::END_OF_ROUTE : SentinelType::END_OF_SEGMENT;
REQUIRE(!util::file_exists(segment_path + "/rlog.lock"));
for (const char *fn : {"/rlog.zst", "/qlog.zst"}) {
const std::string log_file = segment_path + fn;
std::string log = util::read_file(log_file);
REQUIRE(!log.empty());
std::string decompressed_log = zstd_decompress(log);
int event_cnt = 0, i = 0;
kj::ArrayPtr<const capnp::word> words((capnp::word *)decompressed_log.data(), decompressed_log.size() / sizeof(capnp::word));
while (words.size() > 0) {
try {
capnp::FlatArrayMessageReader reader(words);
auto event = reader.getRoot<cereal::Event>();
words = kj::arrayPtr(reader.getEnd(), words.end());
if (i == 0) {
REQUIRE(event.which() == cereal::Event::INIT_DATA);
} else if (i == 1) {
REQUIRE(event.which() == cereal::Event::SENTINEL);
REQUIRE(event.getSentinel().getType() == begin_sentinel);
REQUIRE(event.getSentinel().getSignal() == 0);
} else if (words.size() > 0) {
REQUIRE(event.which() == cereal::Event::CLOCKS);
++event_cnt;
} else {
// the last event must be SENTINEL
REQUIRE(event.which() == cereal::Event::SENTINEL);
REQUIRE(event.getSentinel().getType() == end_sentinel);
REQUIRE(event.getSentinel().getSignal() == (end_sentinel == SentinelType::END_OF_ROUTE ? 1 : 0));
}
++i;
} catch (const kj::Exception &ex) {
INFO("failed parse " << i << " exception :" << ex.getDescription());
REQUIRE(0);
break;
}
}
REQUIRE(event_cnt == required_event_cnt);
}
}
void write_msg(LoggerState *logger) {
MessageBuilder msg;
msg.initEvent().initClocks();
logger->write(msg.toBytes(), true);
}
TEST_CASE("logger") {
const int segment_cnt = 100;
const std::string log_root = "/tmp/test_logger";
system(("rm " + log_root + " -rf").c_str());
std::string route_name;
{
LoggerState logger(log_root);
route_name = logger.routeName();
for (int i = 0; i < segment_cnt; ++i) {
REQUIRE(logger.next());
REQUIRE(util::file_exists(logger.segmentPath() + "/rlog.lock"));
REQUIRE(logger.segment() == i);
write_msg(&logger);
}
logger.setExitSignal(1);
}
for (int i = 0; i < segment_cnt; ++i) {
verify_segment(log_root + "/" + route_name, i, segment_cnt, 1);
}
}

View File

@@ -0,0 +1,355 @@
import numpy as np
import os
import re
import random
import string
import subprocess
import time
from collections import defaultdict
from pathlib import Path
import pytest
import cereal.messaging as messaging
from cereal import log
from cereal.services import SERVICE_LIST
from openpilot.common.basedir import BASEDIR
from openpilot.common.params import Params
from openpilot.common.timeout import Timeout
from openpilot.system.hardware.hw import Paths
from openpilot.system.hardware import TICI
from openpilot.system.loggerd.xattr_cache import getxattr
from openpilot.system.loggerd.deleter import PRESERVE_ATTR_NAME, PRESERVE_ATTR_VALUE
from openpilot.system.manager.process_config import managed_processes
from openpilot.system.version import get_version
from openpilot.tools.lib.helpers import RE
from openpilot.tools.lib.logreader import LogReader
from msgq.visionipc import VisionIpcServer, VisionStreamType
SentinelType = log.Sentinel.SentinelType
CEREAL_SERVICES = [f for f in log.Event.schema.union_fields if f in SERVICE_LIST
and SERVICE_LIST[f].should_log and "encode" not in f.lower()]
class TestLoggerd:
def _get_latest_log_dir(self):
log_dirs = sorted(Path(Paths.log_root()).iterdir(), key=lambda f: f.stat().st_mtime)
return log_dirs[-1]
def _get_log_dir(self, x):
for l in x.splitlines():
for p in l.split(' '):
path = Path(p.strip())
if path.is_dir():
return path
return None
def _get_log_fn(self, x):
for l in x.splitlines():
for p in l.split(' '):
path = Path(p.strip())
if path.is_file():
return path
return None
def _gen_bootlog(self):
with Timeout(5):
out = subprocess.check_output("./bootlog", cwd=os.path.join(BASEDIR, "system/loggerd"), encoding='utf-8')
log_fn = self._get_log_fn(out)
# check existence
assert log_fn is not None
return log_fn
def _check_init_data(self, msgs):
msg = msgs[0]
assert msg.which() == 'initData'
def _check_sentinel(self, msgs, route):
start_type = SentinelType.startOfRoute if route else SentinelType.startOfSegment
assert msgs[1].sentinel.type == start_type
end_type = SentinelType.endOfRoute if route else SentinelType.endOfSegment
assert msgs[-1].sentinel.type == end_type
def _publish_random_messages(self, services: list[str]) -> dict[str, list]:
pm = messaging.PubMaster(services)
managed_processes["loggerd"].start()
for s in services:
assert pm.wait_for_readers_to_update(s, timeout=5)
sent_msgs = defaultdict(list)
for _ in range(random.randint(2, 10) * 100):
for s in services:
try:
m = messaging.new_message(s)
except Exception:
m = messaging.new_message(s, random.randint(2, 10))
pm.send(s, m)
sent_msgs[s].append(m)
for s in services:
assert pm.wait_for_readers_to_update(s, timeout=5)
managed_processes["loggerd"].stop()
return sent_msgs
def _publish_camera_and_audio_messages(self, num_segs=1, segment_length=5):
# Use small frame sizes for testing (width, height, size, stride, uv_offset)
# NV12 format: size = stride * height * 1.5, uv_offset = stride * height
w, h = 320, 240
frame_spec = (w, h, w * h * 3 // 2, w, w * h)
streams = [
(VisionStreamType.VISION_STREAM_ROAD, frame_spec, "roadCameraState"),
(VisionStreamType.VISION_STREAM_DRIVER, frame_spec, "driverCameraState"),
(VisionStreamType.VISION_STREAM_WIDE_ROAD, frame_spec, "wideRoadCameraState"),
]
sm = messaging.SubMaster(["roadEncodeData"])
pm = messaging.PubMaster([s for _, _, s in streams] + ["rawAudioData"])
vipc_server = VisionIpcServer("camerad")
for stream_type, frame_spec, _ in streams:
vipc_server.create_buffers_with_sizes(stream_type, 40, *(frame_spec))
vipc_server.start_listener()
os.environ["LOGGERD_TEST"] = "1"
os.environ["LOGGERD_SEGMENT_LENGTH"] = str(segment_length)
managed_processes["loggerd"].start()
managed_processes["encoderd"].start()
assert pm.wait_for_readers_to_update("roadCameraState", timeout=5)
fps = 20
for n in range(1, int(num_segs * segment_length * fps) + 1):
# send video
for stream_type, frame_spec, state in streams:
dat = np.empty(frame_spec[2], dtype=np.uint8)
vipc_server.send(stream_type, dat[:].flatten().tobytes(), n, n / fps, n / fps)
camera_state = messaging.new_message(state)
frame = getattr(camera_state, state)
frame.frameId = n
pm.send(state, camera_state)
# send audio
msg = messaging.new_message('rawAudioData')
msg.rawAudioData.data = bytes(800 * 2) # 800 samples of int16
msg.rawAudioData.sampleRate = 16000
pm.send('rawAudioData', msg)
for _, _, state in streams:
assert pm.wait_for_readers_to_update(state, timeout=5, dt=0.001)
sm.update(100) # wait for encode data publish
managed_processes["loggerd"].stop()
managed_processes["encoderd"].stop()
def test_init_data_values(self):
os.environ["CLEAN"] = random.choice(["0", "1"])
dongle = ''.join(random.choice(string.printable) for n in range(random.randint(1, 100)))
fake_params = [
# param, initData field, value
("DongleId", "dongleId", dongle),
("GitCommit", "gitCommit", "commit"),
("GitCommitDate", "gitCommitDate", "date"),
("GitBranch", "gitBranch", "branch"),
("GitRemote", "gitRemote", "remote"),
]
params = Params()
for k, _, v in fake_params:
params.put(k, v)
params.put("AccessToken", "abc")
lr = list(LogReader(str(self._gen_bootlog())))
initData = lr[0].initData
assert initData.dirty != bool(os.environ["CLEAN"])
assert initData.version == get_version()
if os.path.isfile("/proc/cmdline"):
with open("/proc/cmdline") as f:
assert list(initData.kernelArgs) == f.read().strip().split(" ")
with open("/proc/version") as f:
assert initData.kernelVersion == f.read()
# check params
logged_params = {entry.key: entry.value for entry in initData.params.entries}
expected_params = {k for k, _, __ in fake_params} | {'AccessToken', 'BootCount'}
assert set(logged_params.keys()) == expected_params, set(logged_params.keys()) ^ expected_params
assert logged_params['AccessToken'] == b'', f"DONT_LOG param value was logged: {repr(logged_params['AccessToken'])}"
for param_key, initData_key, v in fake_params:
assert getattr(initData, initData_key) == v
assert logged_params[param_key].decode() == v
@pytest.mark.xdist_group("camera_encoder_tests") # setting xdist group ensures tests are run in same worker, prevents encoderd from crashing
def test_rotation(self):
Params().put("RecordFront", True)
expected_files = {"rlog.zst", "qlog.zst", "qcamera.ts", "fcamera.hevc", "dcamera.hevc", "ecamera.hevc"}
num_segs = random.randint(2, 3)
length = random.randint(4, 5) # H264 encoder uses 40 lookahead frames and does B-frame reordering, so minimum 3 seconds before qcam output
self._publish_camera_and_audio_messages(num_segs=num_segs, segment_length=length)
route_path = str(self._get_latest_log_dir()).rsplit("--", 1)[0]
for n in range(num_segs):
p = Path(f"{route_path}--{n}")
logged = {f.name for f in p.iterdir() if f.is_file()}
diff = logged ^ expected_files
assert len(diff) == 0, f"didn't get all expected files. seg={n} {route_path=}, {diff=}\n{logged=} {expected_files=}"
def test_bootlog(self):
# generate bootlog with fake launch log
launch_log = ''.join(str(random.choice(string.printable)) for _ in range(100))
with open("/tmp/launch_log", "w") as f:
f.write(launch_log)
bootlog_path = self._gen_bootlog()
lr = list(LogReader(str(bootlog_path)))
# check length
assert len(lr) == 2 # boot + initData
self._check_init_data(lr)
# check msgs
bootlog_msgs = [m for m in lr if m.which() == 'boot']
assert len(bootlog_msgs) == 1
# sanity check values
boot = bootlog_msgs.pop().boot
assert abs(boot.wallTimeNanos - time.time_ns()) < 5*1e9 # within 5s
assert boot.launchLog == launch_log
if TICI:
for fn in ["console-ramoops", "pmsg-ramoops-0"]:
path = Path(os.path.join("/sys/fs/pstore/", fn))
if path.is_file():
with open(path, "rb") as f:
expected_val = f.read()
bootlog_val = [e.value for e in boot.pstore.entries if e.key == fn][0]
assert expected_val == bootlog_val
else:
assert len(boot.pstore.entries) == 0
# next one should increment by one
bl1 = re.match(RE.LOG_ID_V2, bootlog_path.name)
bl2 = re.match(RE.LOG_ID_V2, self._gen_bootlog().name)
assert bl1.group('uid') != bl2.group('uid')
assert int(bl1.group('count')) == 0 and int(bl2.group('count')) == 1
def test_qlog(self):
qlog_services = [s for s in CEREAL_SERVICES if SERVICE_LIST[s].decimation is not None]
no_qlog_services = [s for s in CEREAL_SERVICES if SERVICE_LIST[s].decimation is None]
services = random.sample(qlog_services, random.randint(2, min(10, len(qlog_services)))) + \
random.sample(no_qlog_services, random.randint(2, min(10, len(no_qlog_services))))
sent_msgs = self._publish_random_messages(services)
qlog_path = os.path.join(self._get_latest_log_dir(), "qlog.zst")
lr = list(LogReader(qlog_path))
# check initData and sentinel
self._check_init_data(lr)
self._check_sentinel(lr, True)
recv_msgs = defaultdict(list)
for m in lr:
recv_msgs[m.which()].append(m)
for s, msgs in sent_msgs.items():
recv_cnt = len(recv_msgs[s])
if s in no_qlog_services:
# check services with no specific decimation aren't in qlog
assert recv_cnt == 0, f"got {recv_cnt} {s} msgs in qlog"
else:
# check logged message count matches decimation
expected_cnt = (len(msgs) - 1) // SERVICE_LIST[s].decimation + 1
assert recv_cnt == expected_cnt, f"expected {expected_cnt} msgs for {s}, got {recv_cnt}"
def test_rlog(self):
services = random.sample(CEREAL_SERVICES, random.randint(5, 10))
sent_msgs = self._publish_random_messages(services)
lr = list(LogReader(os.path.join(self._get_latest_log_dir(), "rlog.zst")))
# check initData and sentinel
self._check_init_data(lr)
self._check_sentinel(lr, True)
# check all messages were logged and in order
lr = lr[2:-1] # slice off initData and both sentinels
for m in lr:
sent = sent_msgs[m.which()].pop(0)
sent.clear_write_flag()
assert sent.to_bytes() == m.as_builder().to_bytes()
def test_preserving_bookmarked_segments(self):
services = set(random.sample(CEREAL_SERVICES, random.randint(5, 10))) | {"userBookmark"}
self._publish_random_messages(services)
segment_dir = self._get_latest_log_dir()
assert getxattr(segment_dir, PRESERVE_ATTR_NAME) == PRESERVE_ATTR_VALUE
def test_not_preserving_nonbookmarked_segments(self):
services = set(random.sample(CEREAL_SERVICES, random.randint(5, 10))) - {"userBookmark", "audioFeedback"}
self._publish_random_messages(services)
segment_dir = self._get_latest_log_dir()
assert getxattr(segment_dir, PRESERVE_ATTR_NAME) is None
@pytest.mark.xdist_group("camera_encoder_tests") # setting xdist group ensures tests are run in same worker, prevents encoderd from crashing
@pytest.mark.parametrize("record_front", [True, False])
def test_record_front(self, record_front):
params = Params()
params.put_bool("RecordFront", record_front)
self._publish_camera_and_audio_messages()
dcamera_hevc_exists = os.path.exists(os.path.join(self._get_latest_log_dir(), 'dcamera.hevc'))
assert dcamera_hevc_exists == record_front
@pytest.mark.xdist_group("camera_encoder_tests") # setting xdist group ensures tests are run in same worker, prevents encoderd from crashing
@pytest.mark.parametrize("record_audio", [True, False])
def test_record_audio(self, record_audio):
params = Params()
params.put_bool("RecordAudio", record_audio)
self._publish_camera_and_audio_messages()
qcamera_ts_path = os.path.join(self._get_latest_log_dir(), 'qcamera.ts')
ffprobe_cmd = f"ffprobe -i {qcamera_ts_path} -show_streams -select_streams a -loglevel error"
has_audio_stream = subprocess.run(ffprobe_cmd, shell=True, capture_output=True).stdout.strip() != b''
assert has_audio_stream == record_audio
raw_audio_in_rlog = any(m.which() == 'rawAudioData' for m in LogReader(os.path.join(self._get_latest_log_dir(), 'rlog.zst')))
assert raw_audio_in_rlog == record_audio
@pytest.mark.xdist_group("camera_encoder_tests") # setting xdist group ensures tests are run in same worker, prevents encoderd from crashing
def test_record_audio_init_failure_fails_open(self):
params = Params()
params.put_bool("RecordAudio", True)
os.environ["LOGGERD_TEST_AUDIO_INIT_FAIL"] = "1"
try:
self._publish_camera_and_audio_messages()
finally:
os.environ.pop("LOGGERD_TEST_AUDIO_INIT_FAIL", None)
latest_log_dir = self._get_latest_log_dir()
qcamera_ts_path = os.path.join(latest_log_dir, 'qcamera.ts')
assert os.path.exists(qcamera_ts_path)
ffprobe_cmd = f"ffprobe -i {qcamera_ts_path} -show_streams -select_streams a -loglevel error"
has_audio_stream = subprocess.run(ffprobe_cmd, shell=True, capture_output=True).stdout.strip() != b''
assert has_audio_stream is False
raw_audio_in_rlog = any(m.which() == 'rawAudioData' for m in LogReader(os.path.join(latest_log_dir, 'rlog.zst')))
assert raw_audio_in_rlog is True

View File

@@ -0,0 +1,2 @@
#define CATCH_CONFIG_MAIN
#include "catch2/catch.hpp"

View File

@@ -0,0 +1,184 @@
import os
import time
import threading
import logging
import json
from pathlib import Path
from openpilot.system.hardware.hw import Paths
from openpilot.common.swaglog import cloudlog
from konn3kt_private.uploaderd.iquploaderd import main, UPLOAD_ATTR_NAME, UPLOAD_ATTR_VALUE
from openpilot.system.loggerd.tests.loggerd_tests_common import UploaderTestCase
class FakeLogHandler(logging.Handler):
def __init__(self):
logging.Handler.__init__(self)
self.reset()
def reset(self):
self.upload_order = list()
self.upload_ignored = list()
def emit(self, record):
try:
j = json.loads(record.getMessage())
if j["event"] == "upload_success":
self.upload_order.append(j["key"])
if j["event"] == "upload_ignored":
self.upload_ignored.append(j["key"])
except Exception:
pass
log_handler = FakeLogHandler()
cloudlog.addHandler(log_handler)
class TestUploader(UploaderTestCase):
def setup_method(self):
super().setup_method()
log_handler.reset()
def start_thread(self):
self.end_event = threading.Event()
self.up_thread = threading.Thread(target=main, args=[self.end_event])
self.up_thread.daemon = True
self.up_thread.start()
def join_thread(self):
self.end_event.set()
self.up_thread.join()
def gen_files(self, lock=False, xattr: bytes | None = None, boot=True) -> list[Path]:
f_paths = []
for t in ["qlog", "rlog", "dcamera.hevc", "fcamera.hevc"]:
f_paths.append(self.make_file_with_data(self.seg_dir, t, 1, lock=lock, upload_xattr=xattr))
if boot:
f_paths.append(self.make_file_with_data("boot", f"{self.seg_dir}", 1, lock=lock, upload_xattr=xattr))
return f_paths
def gen_order(self, seg1: list[int], seg2: list[int], boot=True) -> list[str]:
keys = []
if boot:
keys += [f"boot/{self.seg_format.format(i)}.zst" for i in seg1]
keys += [f"boot/{self.seg_format2.format(i)}.zst" for i in seg2]
keys += [f"{self.seg_format.format(i)}/qlog.zst" for i in seg1]
keys += [f"{self.seg_format2.format(i)}/qlog.zst" for i in seg2]
return keys
def test_upload(self):
self.gen_files(lock=False)
self.start_thread()
# allow enough time that files could upload twice if there is a bug in the logic
time.sleep(1)
self.join_thread()
exp_order = self.gen_order([self.seg_num], [])
assert len(log_handler.upload_ignored) == 0, "Some files were ignored"
assert not len(log_handler.upload_order) < len(exp_order), "Some files failed to upload"
assert not len(log_handler.upload_order) > len(exp_order), "Some files were uploaded twice"
for f_path in exp_order:
assert os.getxattr((Path(Paths.log_root()) / f_path).with_suffix(""), UPLOAD_ATTR_NAME) == UPLOAD_ATTR_VALUE, "All files not uploaded"
assert log_handler.upload_order == exp_order, "Files uploaded in wrong order"
def test_upload_with_wrong_xattr(self):
self.gen_files(lock=False, xattr=b'0')
self.start_thread()
# allow enough time that files could upload twice if there is a bug in the logic
time.sleep(1)
self.join_thread()
exp_order = self.gen_order([self.seg_num], [])
assert len(log_handler.upload_ignored) == 0, "Some files were ignored"
assert not len(log_handler.upload_order) < len(exp_order), "Some files failed to upload"
assert not len(log_handler.upload_order) > len(exp_order), "Some files were uploaded twice"
for f_path in exp_order:
assert os.getxattr((Path(Paths.log_root()) / f_path).with_suffix(""), UPLOAD_ATTR_NAME) == UPLOAD_ATTR_VALUE, "All files not uploaded"
assert log_handler.upload_order == exp_order, "Files uploaded in wrong order"
def test_upload_ignored(self):
self.set_ignore()
self.gen_files(lock=False)
self.start_thread()
# allow enough time that files could upload twice if there is a bug in the logic
time.sleep(1)
self.join_thread()
exp_order = self.gen_order([self.seg_num], [])
assert len(log_handler.upload_order) == 0, "Some files were not ignored"
assert not len(log_handler.upload_ignored) < len(exp_order), "Some files failed to ignore"
assert not len(log_handler.upload_ignored) > len(exp_order), "Some files were ignored twice"
for f_path in exp_order:
assert os.getxattr((Path(Paths.log_root()) / f_path).with_suffix(""), UPLOAD_ATTR_NAME) == UPLOAD_ATTR_VALUE, "All files not ignored"
assert log_handler.upload_ignored == exp_order, "Files ignored in wrong order"
def test_upload_files_in_create_order(self):
seg1_nums = [0, 1, 2, 10, 20]
for i in seg1_nums:
self.seg_dir = self.seg_format.format(i)
self.gen_files(boot=False)
seg2_nums = [5, 50, 51]
for i in seg2_nums:
self.seg_dir = self.seg_format2.format(i)
self.gen_files(boot=False)
exp_order = self.gen_order(seg1_nums, seg2_nums, boot=False)
self.start_thread()
# allow enough time that files could upload twice if there is a bug in the logic
time.sleep(1)
self.join_thread()
assert len(log_handler.upload_ignored) == 0, "Some files were ignored"
assert not len(log_handler.upload_order) < len(exp_order), "Some files failed to upload"
assert not len(log_handler.upload_order) > len(exp_order), "Some files were uploaded twice"
for f_path in exp_order:
assert os.getxattr((Path(Paths.log_root()) / f_path).with_suffix(""), UPLOAD_ATTR_NAME) == UPLOAD_ATTR_VALUE, "All files not uploaded"
assert log_handler.upload_order == exp_order, "Files uploaded in wrong order"
def test_no_upload_with_lock_file(self):
self.start_thread()
time.sleep(0.25)
f_paths = self.gen_files(lock=True, boot=False)
# allow enough time that files should have been uploaded if they would be uploaded
time.sleep(1)
self.join_thread()
for f_path in f_paths:
fn = f_path.with_suffix(f_path.suffix.replace(".zst", ""))
uploaded = UPLOAD_ATTR_NAME in os.listxattr(fn) and os.getxattr(fn, UPLOAD_ATTR_NAME) == UPLOAD_ATTR_VALUE
assert not uploaded, "File upload when locked"
def test_no_upload_with_xattr(self):
self.gen_files(lock=False, xattr=UPLOAD_ATTR_VALUE)
self.start_thread()
# allow enough time that files could upload twice if there is a bug in the logic
time.sleep(1)
self.join_thread()
assert len(log_handler.upload_order) == 0, "File uploaded again"
def test_clear_locks_on_startup(self):
f_paths = self.gen_files(lock=True, boot=False)
self.start_thread()
time.sleep(0.25)
self.join_thread()
for f_path in f_paths:
lock_path = f_path.with_suffix(f_path.suffix + ".lock")
assert not lock_path.is_file(), "File lock not cleared on startup"

View File

@@ -0,0 +1,44 @@
#include <zstd.h>
#include <catch2/catch.hpp>
#include <cstring>
#include <vector>
#include "common/util.h"
#include "system/loggerd/logger.h"
#include "system/loggerd/zstd_writer.h"
TEST_CASE("ZstdFileWriter writes and compresses data correctly in loops", "[ZstdFileWriter]") {
const std::string filename = "test_zstd_file.zst";
const int iterations = 100;
const size_t dataSize = 1024;
std::string totalTestData;
// Step 1: Write compressed data to file in a loop
{
ZstdFileWriter writer(filename, LOG_COMPRESSION_LEVEL);
// Write various data sizes including edge cases
std::vector<size_t> testSizes = {dataSize, 1, 0, dataSize * 2}; // Normal, minimal, empty, large
for (int i = 0; i < iterations; ++i) {
size_t currentSize = testSizes[i % testSizes.size()];
std::string testData = util::random_string(currentSize);
totalTestData.append(testData);
writer.write((void *)testData.c_str(), testData.size());
}
}
// Step 2: Decompress the file and verify the data
auto compressedContent = util::read_file(filename);
REQUIRE(compressedContent.size() > 0);
REQUIRE(compressedContent.size() < totalTestData.size());
std::string decompressedData = zstd_decompress(compressedContent);
// Step 3: Verify that the decompressed data matches the original accumulated data
REQUIRE(decompressedData.size() == totalTestData.size());
REQUIRE(std::memcmp(decompressedData.data(), totalTestData.c_str(), totalTestData.size()) == 0);
// Clean up the test file
std::remove(filename.c_str());
}

View File

@@ -0,0 +1,13 @@
#!/usr/bin/env bash
set -e
cd /sys/kernel/debug/tracing
echo "" > trace
echo 1 > tracing_on
echo 1 > /sys/kernel/debug/tracing/events/msm_vidc/enable
echo 0xff > /sys/module/videobuf2_core/parameters/debug
echo 0x7fffffff > /sys/kernel/debug/msm_vidc/debug_level
echo 0xff > /sys/devices/platform/soc/aa00000.qcom,vidc/video4linux/video33/dev_debug
cat /sys/kernel/debug/tracing/trace_pipe

7
system/loggerd/uploader.py Executable file
View File

@@ -0,0 +1,7 @@
#!/usr/bin/env python3
from openpilot.iqpilot._proprietary_loader import load_private_module
load_private_module(__name__, "iqpilot_private.konn3kt.uploaderd.iquploaderd")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,21 @@
#!/usr/bin/env python3
import os
from openpilot.common.swaglog import cloudlog
def get_directory_sort(d: str) -> list[str]:
prefix = ["0"] if d.startswith("2024-") else ["1"]
return prefix + [s.rjust(10, "0") for s in d.rsplit("--", 1)]
def listdir_by_creation(d: str) -> list[str]:
if not os.path.isdir(d):
return []
try:
paths = [f for f in os.listdir(d) if os.path.isdir(os.path.join(d, f))]
return sorted(paths, key=get_directory_sort)
except OSError:
cloudlog.exception("uploader_common.listdir_by_creation_failed")
return []

View File

@@ -0,0 +1,479 @@
// _GNU_SOURCE for sync_file_range() (io_writeback.h); must precede all includes
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <algorithm>
#include <cassert>
#include <cmath>
#include "system/loggerd/video_writer.h"
#include "common/swaglog.h"
#include "common/util.h"
#include "system/loggerd/io_writeback.h"
static void configure_video_stream(AVStream *stream, const AVCodecContext *codec_ctx, uint8_t *extradata, int extradata_size) {
if (stream->codecpar == nullptr) {
stream->codecpar = avcodec_parameters_alloc();
assert(stream->codecpar != nullptr);
}
AVCodecParameters *codecpar = stream->codecpar;
codecpar->codec_type = codec_ctx->codec_type;
codecpar->codec_id = codec_ctx->codec_id;
codecpar->format = codec_ctx->pix_fmt;
codecpar->width = codec_ctx->width;
codecpar->height = codec_ctx->height;
codecpar->bit_rate = codec_ctx->bit_rate;
stream->time_base = codec_ctx->time_base;
stream->avg_frame_rate = AVRational{codec_ctx->time_base.den, codec_ctx->time_base.num};
if (extradata != nullptr && extradata_size > 0) {
codecpar->extradata = (uint8_t *)av_mallocz(extradata_size + AV_INPUT_BUFFER_PADDING_SIZE);
assert(codecpar->extradata != nullptr);
codecpar->extradata_size = extradata_size;
memcpy(codecpar->extradata, extradata, extradata_size);
}
}
static const AVSampleFormat kPreferredAudioSampleFormats[] = {
AV_SAMPLE_FMT_FLTP,
AV_SAMPLE_FMT_FLT,
AV_SAMPLE_FMT_S16P,
AV_SAMPLE_FMT_S16,
};
static const AVSampleFormat *get_supported_audio_sample_formats(const AVCodec *codec) {
if (codec == nullptr) {
return nullptr;
}
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wdeprecated-declarations"
return codec->sample_fmts;
#pragma clang diagnostic pop
}
static AVSampleFormat pick_audio_sample_format(const AVCodec *codec) {
const AVSampleFormat *sample_fmts = get_supported_audio_sample_formats(codec);
if (sample_fmts == nullptr) {
return AV_SAMPLE_FMT_NONE;
}
for (AVSampleFormat preferred_fmt : kPreferredAudioSampleFormats) {
for (const AVSampleFormat *fmt = sample_fmts; *fmt != AV_SAMPLE_FMT_NONE; ++fmt) {
if (*fmt == preferred_fmt) {
return preferred_fmt;
}
}
}
return sample_fmts[0];
}
static std::string audio_sample_format_name(AVSampleFormat fmt) {
const char *name = av_get_sample_fmt_name(fmt);
return name != nullptr ? name : "unknown";
}
static void disable_audio_stream(AVCodecContext **audio_codec_ctx, AVFrame **audio_frame, AVStream **audio_stream,
int *audio_stream_index, std::deque<float> *audio_buffer, bool *audio_failed) {
if (*audio_frame != nullptr) {
av_frame_free(audio_frame);
}
if (*audio_codec_ctx != nullptr) {
avcodec_free_context(audio_codec_ctx);
}
*audio_stream = nullptr;
*audio_stream_index = -1;
audio_buffer->clear();
*audio_failed = true;
}
static void configure_audio_stream(AVStream *stream, const AVCodecContext *codec_ctx) {
AVCodecParameters *codecpar = stream->codecpar;
codecpar->codec_type = codec_ctx->codec_type;
codecpar->codec_id = codec_ctx->codec_id;
codecpar->format = codec_ctx->sample_fmt;
codecpar->bit_rate = codec_ctx->bit_rate;
codecpar->sample_rate = codec_ctx->sample_rate;
#if LIBAVUTIL_VERSION_MAJOR >= 57
codecpar->ch_layout = codec_ctx->ch_layout;
#else
codecpar->channel_layout = codec_ctx->channel_layout;
codecpar->channels = codec_ctx->channels;
#endif
if (codec_ctx->extradata != nullptr && codec_ctx->extradata_size > 0) {
codecpar->extradata = (uint8_t *)av_mallocz(codec_ctx->extradata_size + AV_INPUT_BUFFER_PADDING_SIZE);
assert(codecpar->extradata != nullptr);
codecpar->extradata_size = codec_ctx->extradata_size;
memcpy(codecpar->extradata, codec_ctx->extradata, codec_ctx->extradata_size);
}
stream->time_base = codec_ctx->time_base;
}
static bool fill_audio_frame_samples(AVFrame *frame, const std::deque<float> &audio_buffer, int sample_count) {
auto clamp_to_i16 = [](float sample) -> int16_t {
const float clamped = std::clamp(sample, -1.0f, 1.0f);
return static_cast<int16_t>(std::lrint(clamped * 32767.0f));
};
switch (static_cast<AVSampleFormat>(frame->format)) {
case AV_SAMPLE_FMT_FLTP:
case AV_SAMPLE_FMT_FLT: {
float *samples = reinterpret_cast<float *>(frame->data[0]);
std::copy(audio_buffer.begin(), audio_buffer.begin() + sample_count, samples);
return true;
}
case AV_SAMPLE_FMT_S16P:
case AV_SAMPLE_FMT_S16: {
int16_t *samples = reinterpret_cast<int16_t *>(frame->data[0]);
std::transform(audio_buffer.begin(), audio_buffer.begin() + sample_count, samples, clamp_to_i16);
return true;
}
default:
return false;
}
}
#if LIBAVUTIL_VERSION_MAJOR >= 57
static void set_mono_channel_layout(AVCodecContext *codec_ctx) {
codec_ctx->ch_layout.order = AV_CHANNEL_ORDER_NATIVE;
codec_ctx->ch_layout.nb_channels = 1;
codec_ctx->ch_layout.u.mask = AV_CH_LAYOUT_MONO;
}
static void set_frame_channel_layout(AVFrame *frame, const AVCodecContext *codec_ctx) {
frame->ch_layout = codec_ctx->ch_layout;
}
#else
static void set_mono_channel_layout(AVCodecContext *codec_ctx) {
codec_ctx->channel_layout = AV_CH_LAYOUT_MONO;
}
static void set_frame_channel_layout(AVFrame *frame, const AVCodecContext *codec_ctx) {
frame->channel_layout = codec_ctx->channel_layout;
}
#endif
VideoWriter::VideoWriter(const char *path, const char *filename, bool remuxing, int width, int height, int fps, cereal::EncodeIndex::Type codec)
: remuxing(remuxing) {
dir_path = path;
vid_path = util::string_format("%s/%s", path, filename);
lock_path = util::string_format("%s/%s.lock", path, filename);
int lock_fd = HANDLE_EINTR(open(lock_path.c_str(), O_RDWR | O_CREAT, 0664));
assert(lock_fd >= 0);
close(lock_fd);
LOGD("encoder_open %s remuxing:%d", this->vid_path.c_str(), this->remuxing);
if (this->remuxing) {
bool raw = (codec == cereal::EncodeIndex::Type::BIG_BOX_LOSSLESS);
avformat_alloc_output_context2(&this->ofmt_ctx, NULL, raw ? "matroska" : NULL, this->vid_path.c_str());
assert(this->ofmt_ctx);
// set codec correctly. needed?
assert(codec != cereal::EncodeIndex::Type::FULL_H_E_V_C);
const AVCodec *avcodec = avcodec_find_encoder(raw ? AV_CODEC_ID_FFVHUFF : AV_CODEC_ID_H264);
assert(avcodec);
this->codec_ctx = avcodec_alloc_context3(avcodec);
assert(this->codec_ctx);
this->codec_ctx->width = width;
this->codec_ctx->height = height;
this->codec_ctx->pix_fmt = AV_PIX_FMT_YUV420P;
this->codec_ctx->time_base = (AVRational){ 1, fps };
if (codec == cereal::EncodeIndex::Type::BIG_BOX_LOSSLESS) {
// without this, there's just noise
int err = avcodec_open2(this->codec_ctx, avcodec, NULL);
assert(err >= 0);
}
this->out_stream = avformat_new_stream(this->ofmt_ctx, raw ? avcodec : NULL);
assert(this->out_stream);
this->out_stream_index = 0;
int err = avio_open(&this->ofmt_ctx->pb, this->vid_path.c_str(), AVIO_FLAG_WRITE);
assert(err >= 0);
// second fd on the same inode: lets us fdatasync what avio has written
// without hooking ffmpeg's IO (Linux allows fsync through a read-only fd)
this->durable_fd = HANDLE_EINTR(open(this->vid_path.c_str(), O_RDONLY));
} else {
this->of = util::safe_fopen(this->vid_path.c_str(), "wb");
assert(this->of);
}
fsync_dir(dir_path.c_str());
}
void VideoWriter::initialize_audio(int sample_rate) {
if (this->audio_failed || this->audio_initialized) {
return;
}
if (this->ofmt_ctx->oformat->audio_codec == AV_CODEC_ID_NONE) {
LOGW("audio mux unavailable for %s, continuing without audio", this->vid_path.c_str());
this->audio_failed = true;
return;
}
const AVCodec *audio_avcodec = avcodec_find_encoder(AV_CODEC_ID_AAC);
if (audio_avcodec == nullptr) {
LOGW("audio init disabled for %s: AAC encoder unavailable, continuing without audio", this->vid_path.c_str());
this->audio_failed = true;
return;
}
this->audio_codec_ctx = avcodec_alloc_context3(audio_avcodec);
if (this->audio_codec_ctx == nullptr) {
LOGW("audio init disabled for %s: failed to allocate AAC codec context, continuing without audio", this->vid_path.c_str());
this->audio_failed = true;
return;
}
this->audio_codec_ctx->sample_fmt = pick_audio_sample_format(audio_avcodec);
if (this->audio_codec_ctx->sample_fmt == AV_SAMPLE_FMT_NONE) {
LOGW("audio init disabled for %s: no supported AAC sample format, continuing without audio", this->vid_path.c_str());
disable_audio_stream(&this->audio_codec_ctx, &this->audio_frame, &this->audio_stream, &this->audio_stream_index,
&this->audio_buffer, &this->audio_failed);
return;
}
this->audio_codec_ctx->sample_rate = sample_rate;
set_mono_channel_layout(this->audio_codec_ctx);
this->audio_codec_ctx->bit_rate = 32000;
this->audio_codec_ctx->flags |= AV_CODEC_FLAG_GLOBAL_HEADER;
this->audio_codec_ctx->time_base = (AVRational){1, audio_codec_ctx->sample_rate};
if (util::getenv("LOGGERD_TEST_AUDIO_INIT_FAIL", 0) == 1) {
LOGW("audio init disabled for %s: forced test failure, continuing without audio", this->vid_path.c_str());
disable_audio_stream(&this->audio_codec_ctx, &this->audio_frame, &this->audio_stream, &this->audio_stream_index,
&this->audio_buffer, &this->audio_failed);
return;
}
int err = avcodec_open2(this->audio_codec_ctx, audio_avcodec, NULL);
if (err < 0) {
LOGW("audio init disabled for %s: AAC encoder rejected sample format %s (err=%d), continuing without audio",
this->vid_path.c_str(), audio_sample_format_name(this->audio_codec_ctx->sample_fmt).c_str(), err);
disable_audio_stream(&this->audio_codec_ctx, &this->audio_frame, &this->audio_stream, &this->audio_stream_index,
&this->audio_buffer, &this->audio_failed);
return;
}
av_log_set_level(AV_LOG_WARNING); // hide "QAvg" info msgs at the end of every segment
this->audio_stream = avformat_new_stream(this->ofmt_ctx, NULL);
if (this->audio_stream == nullptr) {
LOGW("audio init disabled for %s: failed to create audio stream, continuing without audio", this->vid_path.c_str());
disable_audio_stream(&this->audio_codec_ctx, &this->audio_frame, &this->audio_stream, &this->audio_stream_index,
&this->audio_buffer, &this->audio_failed);
return;
}
this->audio_stream_index = 1;
configure_audio_stream(this->audio_stream, this->audio_codec_ctx);
this->audio_frame = av_frame_alloc();
if (this->audio_frame == nullptr) {
LOGW("audio init disabled for %s: failed to allocate audio frame, continuing without audio", this->vid_path.c_str());
disable_audio_stream(&this->audio_codec_ctx, &this->audio_frame, &this->audio_stream, &this->audio_stream_index,
&this->audio_buffer, &this->audio_failed);
return;
}
this->audio_frame->format = this->audio_codec_ctx->sample_fmt;
set_frame_channel_layout(this->audio_frame, this->audio_codec_ctx);
this->audio_frame->sample_rate = this->audio_codec_ctx->sample_rate;
this->audio_frame->nb_samples = this->audio_codec_ctx->frame_size;
err = av_frame_get_buffer(this->audio_frame, 0);
if (err < 0) {
LOGW("audio init disabled for %s: failed to allocate audio frame buffer (err=%d), continuing without audio",
this->vid_path.c_str(), err);
disable_audio_stream(&this->audio_codec_ctx, &this->audio_frame, &this->audio_stream, &this->audio_stream_index,
&this->audio_buffer, &this->audio_failed);
return;
}
this->audio_initialized = true;
}
void VideoWriter::write(uint8_t *data, int len, long long timestamp, bool codecconfig, bool keyframe) {
if (of && data) {
size_t written = util::safe_fwrite(data, 1, len, of);
if (written != len) {
LOGE("failed to write file.errno=%d", errno);
}
stream_writeback(of, wb_synced_, wb_dropped_);
}
if (remuxing) {
if (codecconfig) {
if (len > 0) {
codec_ctx->extradata = (uint8_t*)av_mallocz(len + AV_INPUT_BUFFER_PADDING_SIZE);
assert(codec_ctx->extradata != nullptr);
codec_ctx->extradata_size = len;
memcpy(codec_ctx->extradata, data, len);
}
configure_video_stream(out_stream, codec_ctx, codec_ctx->extradata, codec_ctx->extradata_size);
// if there is an audio stream, it must be initialized before this point
int err = avformat_write_header(ofmt_ctx, NULL);
assert(err >= 0);
header_written = true;
} else {
// input timestamps are in microseconds
AVRational in_timebase = {1, 1000000};
AVPacket pkt = {};
pkt.data = data;
pkt.size = len;
pkt.stream_index = this->out_stream_index;
enum AVRounding rnd = static_cast<enum AVRounding>(AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
pkt.pts = pkt.dts = av_rescale_q_rnd(timestamp, in_timebase, out_stream->time_base, rnd);
pkt.duration = av_rescale_q(50*1000, in_timebase, out_stream->time_base);
if (keyframe) {
pkt.flags |= AV_PKT_FLAG_KEY;
}
// TODO: can use av_write_frame for non raw?
int err = av_interleaved_write_frame(ofmt_ctx, &pkt);
if (err < 0) { LOGW("ts encoder write issue len: %d ts: %lld", len, timestamp); }
av_packet_unref(&pkt);
}
}
durable_sync(false);
}
void VideoWriter::durable_sync(bool force) {
if (of) {
stream_durable(of, last_durable_ms, force);
} else if (remuxing && header_written && durable_fd >= 0) {
double now = writeback_now_ms();
if (!force && last_durable_ms != 0.0 && now - last_durable_ms < 2000.0) return;
avio_flush(ofmt_ctx->pb);
durable_fsync(durable_fd);
last_durable_ms = now;
}
}
void VideoWriter::write_audio(uint8_t *data, int len, long long timestamp, int sample_rate) {
if (!remuxing || audio_failed) return;
if (!audio_initialized) {
initialize_audio(sample_rate);
}
if (!audio_initialized || !audio_codec_ctx || !audio_frame) return;
// sync logMonoTime of first audio packet with the timestampEof of first video packet
if (audio_pts == 0) {
audio_pts = (timestamp * audio_codec_ctx->sample_rate) / 1000000ULL;
}
// convert s16le samples to fltp and add to buffer
const int16_t *raw_samples = reinterpret_cast<const int16_t*>(data);
int sample_count = len / sizeof(int16_t);
constexpr float normalizer = 1.0f / 32768.0f;
const size_t max_buffer_size = sample_rate * 10; // 10 seconds
if (audio_buffer.size() + sample_count > max_buffer_size) {
size_t samples_to_drop = (audio_buffer.size() + sample_count) - max_buffer_size;
LOGE("Audio buffer overflow, dropping %zu oldest samples", samples_to_drop);
audio_buffer.erase(audio_buffer.begin(), audio_buffer.begin() + samples_to_drop);
audio_pts += samples_to_drop;
}
// Add new samples to the buffer
const size_t original_size = audio_buffer.size();
audio_buffer.resize(original_size + sample_count);
std::transform(raw_samples, raw_samples + sample_count, audio_buffer.begin() + original_size,
[](int16_t sample) { return sample * normalizer; });
if (!header_written) return; // header not written yet, process audio frame after header is written
while (audio_buffer.size() >= audio_codec_ctx->frame_size) {
audio_frame->pts = audio_pts;
if (!fill_audio_frame_samples(audio_frame, audio_buffer, audio_codec_ctx->frame_size)) {
LOGW("audio init disabled for %s: unsupported audio sample format %s during frame write, continuing without audio",
this->vid_path.c_str(), audio_sample_format_name(static_cast<AVSampleFormat>(audio_frame->format)).c_str());
disable_audio_stream(&this->audio_codec_ctx, &this->audio_frame, &this->audio_stream, &this->audio_stream_index,
&this->audio_buffer, &this->audio_failed);
return;
}
audio_buffer.erase(audio_buffer.begin(), audio_buffer.begin() + audio_codec_ctx->frame_size);
encode_and_write_audio_frame(audio_frame);
}
}
void VideoWriter::encode_and_write_audio_frame(AVFrame* frame) {
if (!remuxing || !audio_codec_ctx) return;
int send_result = avcodec_send_frame(audio_codec_ctx, frame); // encode frame
if (send_result >= 0) {
AVPacket *pkt = av_packet_alloc();
while (avcodec_receive_packet(audio_codec_ctx, pkt) == 0) {
av_packet_rescale_ts(pkt, audio_codec_ctx->time_base, audio_stream->time_base);
pkt->stream_index = audio_stream_index;
int err = av_interleaved_write_frame(ofmt_ctx, pkt); // write encoded frame
if (err < 0) {
LOGW("AUDIO: Write frame failed - error: %d", err);
}
av_packet_unref(pkt);
}
av_packet_free(&pkt);
} else {
LOGW("AUDIO: Failed to send audio frame to encoder: %d", send_result);
}
audio_pts += audio_codec_ctx->frame_size;
}
void VideoWriter::process_remaining_audio() {
if (audio_codec_ctx == nullptr || audio_frame == nullptr) return;
// Process remaining audio samples by padding with silence
if (audio_buffer.size() > 0 && audio_buffer.size() < audio_codec_ctx->frame_size) {
audio_buffer.resize(audio_codec_ctx->frame_size, 0.0f);
// Encode final frame
audio_frame->pts = audio_pts;
if (!fill_audio_frame_samples(audio_frame, audio_buffer, audio_codec_ctx->frame_size)) {
LOGW("audio init disabled for %s: unsupported audio sample format %s during flush, dropping remaining audio",
this->vid_path.c_str(), audio_sample_format_name(static_cast<AVSampleFormat>(audio_frame->format)).c_str());
return;
}
encode_and_write_audio_frame(audio_frame);
}
}
VideoWriter::~VideoWriter() {
if (this->remuxing) {
if (this->audio_codec_ctx) {
if (this->header_written) {
process_remaining_audio();
encode_and_write_audio_frame(NULL); // flush encoder
}
avcodec_free_context(&this->audio_codec_ctx);
}
// a writer that never saw a keyframe (e.g. torn down mid-rotation) has no
// header; av_write_trailer on a header-less muxer crashes inside ffmpeg
if (this->header_written) {
int err = av_write_trailer(this->ofmt_ctx);
if (err != 0) LOGE("av_write_trailer failed %d", err);
} else {
LOGW("closing %s without header, no trailer written", this->vid_path.c_str());
}
avcodec_free_context(&this->codec_ctx);
if (this->audio_frame) av_frame_free(&this->audio_frame);
int err = avio_closep(&this->ofmt_ctx->pb);
if (err != 0) LOGE("avio_closep failed %d", err);
avformat_free_context(this->ofmt_ctx);
if (this->durable_fd >= 0) {
durable_fsync(this->durable_fd);
close(this->durable_fd);
this->durable_fd = -1;
}
} else {
util::safe_fflush(this->of);
durable_fsync(fileno(this->of));
fclose(this->of);
this->of = nullptr;
}
unlink(this->lock_path.c_str());
fsync_dir(this->dir_path.c_str());
}

View File

@@ -0,0 +1,52 @@
#pragma once
#include <string>
#include <deque>
extern "C" {
#include <libavformat/avformat.h>
#include <libavcodec/avcodec.h>
#include <libavutil/channel_layout.h>
#include <libavutil/samplefmt.h>
}
#include "cereal/messaging/messaging.h"
class VideoWriter {
public:
VideoWriter(const char *path, const char *filename, bool remuxing, int width, int height, int fps, cereal::EncodeIndex::Type codec);
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);
void durable_sync(bool force);
~VideoWriter();
private:
void initialize_audio(int sample_rate);
void encode_and_write_audio_frame(AVFrame* frame);
void process_remaining_audio();
std::string vid_path, lock_path, dir_path;
FILE *of = nullptr;
off_t wb_synced_ = 0; // io_writeback.h cursors
off_t wb_dropped_ = 0;
int durable_fd = -1; // remux path: second fd on vid_path for fdatasync
double last_durable_ms = 0.; // io_writeback.h durable-checkpoint cursor
AVCodecContext *codec_ctx;
AVFormatContext *ofmt_ctx;
AVStream *out_stream;
int out_stream_index = -1;
bool audio_initialized = false;
bool audio_failed = false;
bool header_written = false;
AVStream *audio_stream = nullptr;
int audio_stream_index = -1;
AVCodecContext *audio_codec_ctx = nullptr;
AVFrame *audio_frame = nullptr;
uint64_t audio_pts = 0;
std::deque<float> audio_buffer;
bool remuxing;
};

View File

@@ -0,0 +1,23 @@
import errno
import xattr
_cached_attributes: dict[tuple, bytes | None] = {}
def getxattr(path: str, attr_name: str) -> bytes | None:
key = (path, attr_name)
if key not in _cached_attributes:
try:
response = xattr.getxattr(path, attr_name)
except OSError as e:
# ENODATA (Linux) or ENOATTR (macOS) means attribute hasn't been set
if e.errno == errno.ENODATA or (hasattr(errno, 'ENOATTR') and e.errno == errno.ENOATTR):
response = None
else:
raise
_cached_attributes[key] = response
return _cached_attributes[key]
def setxattr(path: str, attr_name: str, attr_value: bytes) -> None:
_cached_attributes.pop((path, attr_name), None)
xattr.setxattr(path, attr_name, attr_value)

View File

@@ -0,0 +1,86 @@
// _GNU_SOURCE for sync_file_range() (io_writeback.h); must precede all includes
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include "system/loggerd/zstd_writer.h"
#include <cassert>
#include "common/util.h"
#include "system/loggerd/io_writeback.h"
// Constructor: Initializes compression stream and opens file
ZstdFileWriter::ZstdFileWriter(const std::string& filename, int compression_level) {
// Create the compression stream
cstream_ = ZSTD_createCStream();
assert(cstream_);
size_t initResult = ZSTD_initCStream(cstream_, compression_level);
assert(!ZSTD_isError(initResult));
input_cache_capacity_ = ZSTD_CStreamInSize();
input_cache_.reserve(input_cache_capacity_);
output_buffer_.resize(ZSTD_CStreamOutSize());
file_ = util::safe_fopen(filename.c_str(), "wb");
assert(file_ != nullptr);
}
// Destructor: Finalizes compression and closes file
ZstdFileWriter::~ZstdFileWriter() {
flushCache(ZSTD_e_end);
util::safe_fflush(file_);
durable_fsync(fileno(file_));
int err = fclose(file_);
assert(err == 0);
ZSTD_freeCStream(cstream_);
}
// Compresses and writes data to file
void ZstdFileWriter::write(void* data, size_t size) {
// Add data to the input cache
input_cache_.insert(input_cache_.end(), (uint8_t*)data, (uint8_t*)data + size);
// If the cache is full, compress and write to the file
if (input_cache_.size() >= input_cache_capacity_) {
flushCache(ZSTD_e_continue);
}
durable_flush(false);
}
// Durable checkpoint: end a zstd block (stream stays decodable up to here even
// if the file is truncated later), then commit it to storage.
void ZstdFileWriter::durable_flush(bool force) {
double now = writeback_now_ms();
if (!force && last_durable_ms_ != 0.0 && now - last_durable_ms_ < 2000.0) return;
flushCache(ZSTD_e_flush);
if (fflush(file_) == 0) durable_fsync(fileno(file_));
last_durable_ms_ = now;
}
// Compress and flush the input cache to the file
void ZstdFileWriter::flushCache(ZSTD_EndDirective mode) {
ZSTD_inBuffer input = {input_cache_.data(), input_cache_.size(), 0};
int finished = 0;
do {
ZSTD_outBuffer output = {output_buffer_.data(), output_buffer_.size(), 0};
size_t remaining = ZSTD_compressStream2(cstream_, &output, &input, mode);
assert(!ZSTD_isError(remaining));
size_t written = util::safe_fwrite(output_buffer_.data(), 1, output.pos, file_);
assert(written == output.pos);
finished = (mode == ZSTD_e_continue) ? (input.pos == input.size) : (remaining == 0);
} while (!finished);
input_cache_.clear(); // Clear cache after compression
if (mode == ZSTD_e_continue) {
stream_writeback(file_, wb_synced_, wb_dropped_);
}
}

View File

@@ -0,0 +1,29 @@
#pragma once
#include <zstd.h>
#include <sys/types.h>
#include <string>
#include <vector>
#include <capnp/common.h>
class ZstdFileWriter {
public:
ZstdFileWriter(const std::string &filename, int compression_level);
~ZstdFileWriter();
void write(void* data, size_t size);
inline void write(kj::ArrayPtr<capnp::byte> array) { write(array.begin(), array.size()); }
void durable_flush(bool force);
private:
void flushCache(ZSTD_EndDirective mode);
size_t input_cache_capacity_ = 0;
std::vector<char> input_cache_;
std::vector<char> output_buffer_;
ZSTD_CStream *cstream_;
FILE* file_ = nullptr;
off_t wb_synced_ = 0; // io_writeback.h cursors
off_t wb_dropped_ = 0;
double last_durable_ms_ = 0.; // io_writeback.h durable-checkpoint cursor
};