mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-21 00:03:45 +08:00
loggerd: move to system/ (#27534)
old-commit-hash: 94eb2159802d3dba99620db0b08731c68a7e4733
This commit is contained in:
Adeeb Shihadeh
@@ -0,0 +1,4 @@
|
||||
loggerd
|
||||
encoderd
|
||||
bootlog
|
||||
tests/test_logger
|
||||
@@ -0,0 +1,30 @@
|
||||
# loggerd
|
||||
|
||||
openpilot records routes in one minute chunks called segments. A route starts on the rising edge of ignition and ends on the falling edge.
|
||||
|
||||
Check out our [python library](https://github.com/commaai/openpilot/blob/master/tools/lib/logreader.py) for reading openpilot logs. Also checkout our [tools](https://github.com/commaai/openpilot/tree/master/tools) to replay and view your data. These are the same tools we use to debug and develop openpilot.
|
||||
|
||||
## log types
|
||||
|
||||
For each segment, openpilot records the following log types:
|
||||
|
||||
## rlog.bz2
|
||||
|
||||
rlogs contain all the messages passed amongst openpilot's processes. See [cereal/services.py](https://github.com/commaai/cereal/blob/master/services.py) for a list of all the logged services. They're a bzip2 archive of the serialized capnproto messages.
|
||||
|
||||
## {f,e,d}camera.hevc
|
||||
|
||||
Each camera stream is H.265 encoded and written to its respective file.
|
||||
* fcamera.hevc is the road camera
|
||||
* ecamera.hevc is the wide road camera
|
||||
* dcamera.hevc is the driver camera
|
||||
|
||||
## qlog.bz2 & qcamera.ts
|
||||
|
||||
qlogs are a decimated subset of the rlogs. Check out [cereal/services.py](https://github.com/commaai/cereal/blob/master/services.py) for the decimation.
|
||||
|
||||
|
||||
qcameras are H.264 encoded, lower res versions of the fcamera.hevc. The video shown in [comma connect](https://connect.comma.ai/) is from the qcameras.
|
||||
|
||||
|
||||
qlogs and qcameras are designed to be small enough to upload instantly on slow internet and store forever, yet useful enough for most analysis and debugging.
|
||||
@@ -0,0 +1,27 @@
|
||||
Import('env', 'arch', 'cereal', 'messaging', 'common', 'visionipc')
|
||||
|
||||
libs = [common, cereal, messaging, visionipc,
|
||||
'zmq', 'capnp', 'kj', 'z',
|
||||
'avformat', 'avcodec', 'swscale', 'avutil',
|
||||
'yuv', 'OpenCL', 'pthread']
|
||||
|
||||
src = ['logger.cc', 'video_writer.cc', 'encoder/encoder.cc', 'encoder/v4l_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)
|
||||
env.Program('bootlog.cc', LIBS=libs)
|
||||
|
||||
if GetOption('test'):
|
||||
env.Program('tests/test_logger', ['tests/test_runner.cc', 'tests/test_logger.cc'], LIBS=libs + ['curl', 'crypto'])
|
||||
@@ -0,0 +1,65 @@
|
||||
#include <cassert>
|
||||
#include <string>
|
||||
|
||||
#include "cereal/messaging/messaging.h"
|
||||
#include "common/swaglog.h"
|
||||
#include "system/loggerd/logger.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",
|
||||
};
|
||||
|
||||
if (Hardware::TICI()) {
|
||||
bootlog_commands.push_back("[ -e /dev/nvme0 ] && sudo nvme smart-log --output-format=json /dev/nvme0");
|
||||
}
|
||||
|
||||
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 path = LOG_ROOT + "/boot/" + logger_get_route_name();
|
||||
LOGW("bootlog to %s", path.c_str());
|
||||
|
||||
// Open bootlog
|
||||
bool r = util::create_directories(LOG_ROOT + "/boot/", 0775);
|
||||
assert(r);
|
||||
|
||||
RawFile file(path.c_str());
|
||||
// Write initdata
|
||||
file.write(logger_build_init_data().asBytes());
|
||||
// Write bootlog
|
||||
file.write(build_boot_log().asBytes());
|
||||
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
from system.hardware import PC
|
||||
|
||||
if os.environ.get('LOG_ROOT', False):
|
||||
ROOT = os.environ['LOG_ROOT']
|
||||
elif PC:
|
||||
ROOT = os.path.join(str(Path.home()), ".comma", "media", "0", "realdata")
|
||||
else:
|
||||
ROOT = '/data/media/0/realdata/'
|
||||
|
||||
|
||||
CAMERA_FPS = 20
|
||||
SEGMENT_LENGTH = 60
|
||||
|
||||
STATS_DIR_FILE_LIMIT = 10000
|
||||
STATS_SOCKET = "ipc:///tmp/stats"
|
||||
if PC:
|
||||
STATS_DIR = os.path.join(str(Path.home()), ".comma", "stats")
|
||||
else:
|
||||
STATS_DIR = "/data/stats/"
|
||||
STATS_FLUSH_TIME_S = 60
|
||||
|
||||
def get_available_percent(default=None):
|
||||
try:
|
||||
statvfs = os.statvfs(ROOT)
|
||||
available_percent = 100.0 * statvfs.f_bavail / statvfs.f_blocks
|
||||
except OSError:
|
||||
available_percent = default
|
||||
|
||||
return available_percent
|
||||
|
||||
|
||||
def get_available_bytes(default=None):
|
||||
try:
|
||||
statvfs = os.statvfs(ROOT)
|
||||
available_bytes = statvfs.f_bavail * statvfs.f_frsize
|
||||
except OSError:
|
||||
available_bytes = default
|
||||
|
||||
return available_bytes
|
||||
@@ -0,0 +1,48 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
import shutil
|
||||
import threading
|
||||
from system.swaglog import cloudlog
|
||||
from system.loggerd.config import ROOT, get_available_bytes, get_available_percent
|
||||
from system.loggerd.uploader import listdir_by_creation
|
||||
|
||||
MIN_BYTES = 5 * 1024 * 1024 * 1024
|
||||
MIN_PERCENT = 10
|
||||
|
||||
DELETE_LAST = ['boot', 'crash']
|
||||
|
||||
|
||||
def deleter_thread(exit_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:
|
||||
# remove the earliest directory we can
|
||||
dirs = sorted(listdir_by_creation(ROOT), key=lambda x: x in DELETE_LAST)
|
||||
for delete_dir in dirs:
|
||||
delete_path = os.path.join(ROOT, delete_dir)
|
||||
|
||||
if any(name.endswith(".lock") for name in os.listdir(delete_path)):
|
||||
continue
|
||||
|
||||
try:
|
||||
cloudlog.info(f"deleting {delete_path}")
|
||||
if os.path.isfile(delete_path):
|
||||
os.remove(delete_path)
|
||||
else:
|
||||
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()
|
||||
@@ -0,0 +1,82 @@
|
||||
#include <cassert>
|
||||
#include "system/loggerd/encoder/encoder.h"
|
||||
|
||||
VideoEncoder::~VideoEncoder() {}
|
||||
|
||||
void VideoEncoder::publisher_init() {
|
||||
// publish
|
||||
service_name = this->type == DriverCam ? "driverEncodeData" :
|
||||
(this->type == WideRoadCam ? "wideRoadEncodeData" :
|
||||
(this->in_width == this->out_width ? "roadEncodeData" : "qRoadEncodeData"));
|
||||
pm.reset(new PubMaster({service_name}));
|
||||
}
|
||||
|
||||
void VideoEncoder::publisher_publish(VideoEncoder *e, int segment_num, uint32_t idx, VisionIpcBufExtra &extra,
|
||||
unsigned int flags, kj::ArrayPtr<capnp::byte> header, kj::ArrayPtr<capnp::byte> dat) {
|
||||
// broadcast packet
|
||||
MessageBuilder msg;
|
||||
auto event = msg.initEvent(true);
|
||||
auto edat = (e->type == DriverCam) ? event.initDriverEncodeData() :
|
||||
((e->type == WideRoadCam) ? event.initWideRoadEncodeData() :
|
||||
(e->in_width == e->out_width ? event.initRoadEncodeData() : event.initQRoadEncodeData()));
|
||||
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(e->codec);
|
||||
edata.setEncodeId(e->cnt++);
|
||||
edata.setSegmentNum(segment_num);
|
||||
edata.setSegmentId(idx);
|
||||
edata.setFlags(flags);
|
||||
edata.setLen(dat.size());
|
||||
edat.setData(dat);
|
||||
if (flags & V4L2_BUF_FLAG_KEYFRAME) edat.setHeader(header);
|
||||
|
||||
auto words = new kj::Array<capnp::word>(capnp::messageToFlatArray(msg));
|
||||
auto bytes = words->asBytes();
|
||||
e->pm->send(e->service_name, bytes.begin(), bytes.size());
|
||||
if (e->write) {
|
||||
e->to_write.push(words);
|
||||
} else {
|
||||
delete words;
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: writing should be moved to loggerd
|
||||
void VideoEncoder::write_handler(VideoEncoder *e, const char *path) {
|
||||
VideoWriter writer(path, e->filename, e->codec != cereal::EncodeIndex::Type::FULL_H_E_V_C, e->out_width, e->out_height, e->fps, e->codec);
|
||||
|
||||
bool first = true;
|
||||
kj::Array<capnp::word>* out_buf;
|
||||
while ((out_buf = e->to_write.pop())) {
|
||||
capnp::FlatArrayMessageReader cmsg(*out_buf);
|
||||
cereal::Event::Reader event = cmsg.getRoot<cereal::Event>();
|
||||
|
||||
auto edata = (e->type == DriverCam) ? event.getDriverEncodeData() :
|
||||
((e->type == WideRoadCam) ? event.getWideRoadEncodeData() :
|
||||
(e->in_width == e->out_width ? event.getRoadEncodeData() : event.getQRoadEncodeData()));
|
||||
auto idx = edata.getIdx();
|
||||
auto flags = idx.getFlags();
|
||||
|
||||
if (first) {
|
||||
assert(flags & V4L2_BUF_FLAG_KEYFRAME);
|
||||
auto header = edata.getHeader();
|
||||
writer.write((uint8_t *)header.begin(), header.size(), idx.getTimestampEof()/1000, true, false);
|
||||
first = false;
|
||||
}
|
||||
|
||||
// dangerous cast from const, but should be fine
|
||||
auto data = edata.getData();
|
||||
if (data.size() > 0) {
|
||||
writer.write((uint8_t *)data.begin(), data.size(), idx.getTimestampEof()/1000, false, flags & V4L2_BUF_FLAG_KEYFRAME);
|
||||
}
|
||||
|
||||
// free the data
|
||||
delete out_buf;
|
||||
}
|
||||
|
||||
// VideoWriter is freed on out of scope
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
#pragma once
|
||||
|
||||
#include <cassert>
|
||||
#include <cstdint>
|
||||
#include <thread>
|
||||
|
||||
#include "cereal/messaging/messaging.h"
|
||||
#include "cereal/visionipc/visionipc.h"
|
||||
#include "common/queue.h"
|
||||
#include "system/loggerd/video_writer.h"
|
||||
#include "system/camerad/cameras/camera_common.h"
|
||||
|
||||
#define V4L2_BUF_FLAG_KEYFRAME 8
|
||||
|
||||
class VideoEncoder {
|
||||
public:
|
||||
VideoEncoder(const char* filename, CameraType type, int in_width, int in_height, int fps,
|
||||
int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height, bool write)
|
||||
: filename(filename), type(type), in_width(in_width), in_height(in_height), fps(fps),
|
||||
bitrate(bitrate), codec(codec), out_width(out_width), out_height(out_height), write(write) { }
|
||||
virtual ~VideoEncoder();
|
||||
virtual int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra) = 0;
|
||||
virtual void encoder_open(const char* path) = 0;
|
||||
virtual void encoder_close() = 0;
|
||||
|
||||
void publisher_init();
|
||||
static void publisher_publish(VideoEncoder *e, int segment_num, uint32_t idx, VisionIpcBufExtra &extra, unsigned int flags, kj::ArrayPtr<capnp::byte> header, kj::ArrayPtr<capnp::byte> dat);
|
||||
|
||||
void writer_open(const char* path) {
|
||||
if (this->write) write_handler_thread = std::thread(VideoEncoder::write_handler, this, path);
|
||||
}
|
||||
|
||||
void writer_close() {
|
||||
if (this->write) {
|
||||
to_write.push(NULL);
|
||||
write_handler_thread.join();
|
||||
}
|
||||
assert(to_write.empty());
|
||||
}
|
||||
|
||||
protected:
|
||||
bool write;
|
||||
const char* filename;
|
||||
int in_width, in_height;
|
||||
int out_width, out_height, fps;
|
||||
int bitrate;
|
||||
cereal::EncodeIndex::Type codec;
|
||||
CameraType type;
|
||||
|
||||
private:
|
||||
// total frames encoded
|
||||
int cnt = 0;
|
||||
|
||||
// publishing
|
||||
std::unique_ptr<PubMaster> pm;
|
||||
const char *service_name;
|
||||
|
||||
// writing support
|
||||
static void write_handler(VideoEncoder *e, const char *path);
|
||||
std::thread write_handler_thread;
|
||||
SafeQueue<kj::Array<capnp::word>* > to_write;
|
||||
};
|
||||
@@ -0,0 +1,153 @@
|
||||
#pragma clang diagnostic ignored "-Wdeprecated-declarations"
|
||||
|
||||
#include "system/loggerd/encoder/ffmpeg_encoder.h"
|
||||
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <cassert>
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
|
||||
#define __STDC_CONSTANT_MACROS
|
||||
|
||||
#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;
|
||||
|
||||
void FfmpegEncoder::encoder_init() {
|
||||
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);
|
||||
}
|
||||
|
||||
publisher_init();
|
||||
}
|
||||
|
||||
FfmpegEncoder::~FfmpegEncoder() {
|
||||
encoder_close();
|
||||
av_frame_free(&frame);
|
||||
}
|
||||
|
||||
void FfmpegEncoder::encoder_open(const char* path) {
|
||||
const AVCodec *codec = avcodec_find_encoder(AV_CODEC_ID_FFVHUFF);
|
||||
|
||||
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, fps };
|
||||
int err = avcodec_open2(this->codec_ctx, codec, NULL);
|
||||
assert(err >= 0);
|
||||
|
||||
writer_open(path);
|
||||
is_open = true;
|
||||
segment_num++;
|
||||
counter = 0;
|
||||
}
|
||||
|
||||
void FfmpegEncoder::encoder_close() {
|
||||
if (!is_open) return;
|
||||
|
||||
writer_close();
|
||||
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;
|
||||
av_init_packet(&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", this->filename, pkt.size, pkt.flags, counter, extra->frame_id);
|
||||
}
|
||||
|
||||
publisher_publish(this, 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;
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
#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 char* filename, CameraType type, int in_width, int in_height, int fps,
|
||||
int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height, bool write) :
|
||||
VideoEncoder(filename, type, in_width, in_height, fps, bitrate, cereal::EncodeIndex::Type::BIG_BOX_LOSSLESS, out_width, out_height, write) { encoder_init(); }
|
||||
~FfmpegEncoder();
|
||||
void encoder_init();
|
||||
int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra);
|
||||
void encoder_open(const char* path);
|
||||
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;
|
||||
};
|
||||
@@ -0,0 +1,311 @@
|
||||
#include <cassert>
|
||||
#include <sys/ioctl.h>
|
||||
#include <poll.h>
|
||||
|
||||
#include "system/loggerd/encoder/v4l_encoder.h"
|
||||
#include "common/util.h"
|
||||
#include "common/timing.h"
|
||||
|
||||
#include "libyuv.h"
|
||||
#include "msm_media_info.h"
|
||||
|
||||
// has to be in this order
|
||||
#include "v4l2-controls.h"
|
||||
#include <linux/videodev2.h>
|
||||
#define V4L2_QCOM_BUF_FLAG_CODECCONFIG 0x00020000
|
||||
#define V4L2_QCOM_BUF_FLAG_EOS 0x02000000
|
||||
|
||||
// echo 0x7fffffff > /sys/kernel/debug/msm_vidc/debug_level
|
||||
const int env_debug_encoder = (getenv("DEBUG_ENCODER") != NULL) ? atoi(getenv("DEBUG_ENCODER")) : 0;
|
||||
|
||||
#define checked_ioctl(x,y,z) { int _ret = HANDLE_EINTR(ioctl(x,y,z)); if (_ret!=0) { LOGE("checked_ioctl failed %d %lx %p", x, y, z); } assert(_ret==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,
|
||||
};
|
||||
checked_ioctl(fd, VIDIOC_DQBUF, &v4l_buf);
|
||||
|
||||
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
|
||||
};
|
||||
|
||||
checked_ioctl(fd, VIDIOC_QBUF, &v4l_buf);
|
||||
}
|
||||
|
||||
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
|
||||
};
|
||||
checked_ioctl(fd, VIDIOC_REQBUFS, &reqbuf);
|
||||
}
|
||||
|
||||
void V4LEncoder::dequeue_handler(V4LEncoder *e) {
|
||||
std::string dequeue_thread_name = "dq-"+std::string(e->filename);
|
||||
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) { LOGE("encoder dequeue poll timeout"); continue; }
|
||||
|
||||
if (env_debug_encoder >= 2) {
|
||||
printf("%20s poll %x at %.2f ms\n", e->filename, 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, ×tamp);
|
||||
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, 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->filename, 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void V4LEncoder::encoder_init() {
|
||||
fd = open("/dev/v4l/by-path/platform-aa00000.qcom_vidc-video-index1", O_RDWR|O_NONBLOCK);
|
||||
assert(fd >= 0);
|
||||
|
||||
struct v4l2_capability cap;
|
||||
checked_ioctl(fd, VIDIOC_QUERYCAP, &cap);
|
||||
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);
|
||||
|
||||
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 = (codec == cereal::EncodeIndex::Type::FULL_H_E_V_C) ? V4L2_PIX_FMT_HEVC : V4L2_PIX_FMT_H264,
|
||||
.field = V4L2_FIELD_ANY,
|
||||
.colorspace = V4L2_COLORSPACE_DEFAULT,
|
||||
}
|
||||
}
|
||||
};
|
||||
checked_ioctl(fd, VIDIOC_S_FMT, &fmt_out);
|
||||
|
||||
v4l2_streamparm streamparm = {
|
||||
.type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE,
|
||||
.parm = {
|
||||
.output = {
|
||||
// TODO: more stuff here? we don't know
|
||||
.timeperframe = {
|
||||
.numerator = 1,
|
||||
.denominator = 20
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
checked_ioctl(fd, VIDIOC_S_PARM, &streamparm);
|
||||
|
||||
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,
|
||||
}
|
||||
}
|
||||
};
|
||||
checked_ioctl(fd, VIDIOC_S_FMT, &fmt_in);
|
||||
|
||||
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
|
||||
{
|
||||
struct v4l2_control ctrls[] = {
|
||||
{ .id = V4L2_CID_MPEG_VIDEO_HEADER_MODE, .value = V4L2_MPEG_VIDEO_HEADER_MODE_SEPARATE},
|
||||
{ .id = V4L2_CID_MPEG_VIDEO_BITRATE, .value = bitrate},
|
||||
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_RATE_CONTROL, .value = V4L2_CID_MPEG_VIDC_VIDEO_RATE_CONTROL_VBR_CFR},
|
||||
{ .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) {
|
||||
checked_ioctl(fd, VIDIOC_S_CTRL, &ctrl);
|
||||
}
|
||||
}
|
||||
|
||||
if (codec == cereal::EncodeIndex::Type::FULL_H_E_V_C) {
|
||||
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_NUM_P_FRAMES, .value = 29},
|
||||
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_NUM_B_FRAMES, .value = 0},
|
||||
};
|
||||
for (auto ctrl : ctrls) {
|
||||
checked_ioctl(fd, VIDIOC_S_CTRL, &ctrl);
|
||||
}
|
||||
} 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_VIDC_VIDEO_NUM_P_FRAMES, .value = 14},
|
||||
{ .id = V4L2_CID_MPEG_VIDC_VIDEO_NUM_B_FRAMES, .value = 0},
|
||||
{ .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) {
|
||||
checked_ioctl(fd, VIDIOC_S_CTRL, &ctrl);
|
||||
}
|
||||
}
|
||||
|
||||
// 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;
|
||||
checked_ioctl(fd, VIDIOC_STREAMON, &buf_type);
|
||||
buf_type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;
|
||||
checked_ioctl(fd, VIDIOC_STREAMON, &buf_type);
|
||||
|
||||
// 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);
|
||||
}
|
||||
|
||||
publisher_init();
|
||||
}
|
||||
|
||||
void V4LEncoder::encoder_open(const char* path) {
|
||||
dequeue_handler_thread = std::thread(V4LEncoder::dequeue_handler, this);
|
||||
writer_open(path);
|
||||
this->is_open = true;
|
||||
this->counter = 0;
|
||||
}
|
||||
|
||||
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 };
|
||||
checked_ioctl(fd, VIDIOC_ENCODER_CMD, &encoder_cmd);
|
||||
// join waits for V4L2_QCOM_BUF_FLAG_EOS
|
||||
dequeue_handler_thread.join();
|
||||
assert(extras.empty());
|
||||
writer_close();
|
||||
}
|
||||
this->is_open = false;
|
||||
}
|
||||
|
||||
V4LEncoder::~V4LEncoder() {
|
||||
encoder_close();
|
||||
v4l2_buf_type buf_type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;
|
||||
checked_ioctl(fd, VIDIOC_STREAMOFF, &buf_type);
|
||||
request_buffers(fd, V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE, 0);
|
||||
buf_type = V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE;
|
||||
checked_ioctl(fd, VIDIOC_STREAMOFF, &buf_type);
|
||||
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");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
#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 char* filename, CameraType type, int in_width, int in_height, int fps,
|
||||
int bitrate, cereal::EncodeIndex::Type codec, int out_width, int out_height, bool write) :
|
||||
VideoEncoder(filename, type, in_width, in_height, fps, bitrate, codec, out_width, out_height, write) { encoder_init(); }
|
||||
~V4LEncoder();
|
||||
void encoder_init();
|
||||
int encode_frame(VisionBuf* buf, VisionIpcBufExtra *extra);
|
||||
void encoder_open(const char* path);
|
||||
void encoder_close();
|
||||
private:
|
||||
int fd;
|
||||
|
||||
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;
|
||||
};
|
||||
@@ -0,0 +1,149 @@
|
||||
#include "system/loggerd/loggerd.h"
|
||||
|
||||
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[WideRoadCam + 1] = {};
|
||||
bool camera_synced[WideRoadCam + 1] = {};
|
||||
};
|
||||
|
||||
// Handle initial encoder syncing by waiting for all encoders to reach the same frame id
|
||||
bool sync_encoders(EncoderdState *s, CameraType 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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
void encoder_thread(EncoderdState *s, const LogCameraInfo &cam_info) {
|
||||
util::set_thread_name(cam_info.filename);
|
||||
|
||||
std::vector<Encoder *> encoders;
|
||||
VisionIpcClient vipc_client = VisionIpcClient("camerad", cam_info.stream_type, false);
|
||||
|
||||
int cur_seg = 0;
|
||||
while (!do_exit) {
|
||||
if (!vipc_client.connect(false)) {
|
||||
util::sleep_for(5);
|
||||
continue;
|
||||
}
|
||||
|
||||
// init encoders
|
||||
if (encoders.empty()) {
|
||||
VisionBuf buf_info = vipc_client.buffers[0];
|
||||
LOGW("encoder %s init %dx%d", cam_info.filename, buf_info.width, buf_info.height);
|
||||
|
||||
if (buf_info.width > 0 && buf_info.height > 0) {
|
||||
// main encoder
|
||||
encoders.push_back(new Encoder(cam_info.filename, cam_info.type, buf_info.width, buf_info.height,
|
||||
cam_info.fps, cam_info.bitrate,
|
||||
cam_info.is_h265 ? cereal::EncodeIndex::Type::FULL_H_E_V_C : cereal::EncodeIndex::Type::QCAMERA_H264,
|
||||
buf_info.width, buf_info.height, false));
|
||||
// qcamera encoder
|
||||
if (cam_info.has_qcamera) {
|
||||
encoders.push_back(new Encoder(qcam_info.filename, cam_info.type, buf_info.width, buf_info.height,
|
||||
qcam_info.fps, qcam_info.bitrate,
|
||||
qcam_info.is_h265 ? cereal::EncodeIndex::Type::FULL_H_E_V_C : cereal::EncodeIndex::Type::QCAMERA_H264,
|
||||
qcam_info.frame_width, qcam_info.frame_height, false));
|
||||
}
|
||||
} else {
|
||||
LOGE("not initting empty encoder");
|
||||
s->max_waiting--;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
for (int i = 0; i < encoders.size(); ++i) {
|
||||
encoders[i]->encoder_open(NULL);
|
||||
}
|
||||
|
||||
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: %d extra id: %d", cam_info.filename, buf->get_frame_id(), extra.frame_id);
|
||||
lagging = true;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
lagging = false;
|
||||
|
||||
if (!sync_encoders(s, cam_info.type, extra.frame_id)) {
|
||||
continue;
|
||||
}
|
||||
if (do_exit) break;
|
||||
|
||||
// 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(NULL);
|
||||
}
|
||||
++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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
LOG("encoder destroy");
|
||||
for(auto &e : encoders) {
|
||||
e->encoder_close();
|
||||
delete e;
|
||||
}
|
||||
}
|
||||
|
||||
void encoderd_thread() {
|
||||
EncoderdState s;
|
||||
|
||||
std::vector<std::thread> encoder_threads;
|
||||
for (const auto &cam : cameras_logged) {
|
||||
encoder_threads.push_back(std::thread(encoder_thread, &s, cam));
|
||||
s.max_waiting++;
|
||||
}
|
||||
for (auto &t : encoder_threads) t.join();
|
||||
}
|
||||
|
||||
int main() {
|
||||
if (!Hardware::PC()) {
|
||||
int ret;
|
||||
ret = util::set_realtime_priority(52);
|
||||
assert(ret == 0);
|
||||
ret = util::set_core_affinity({3});
|
||||
assert(ret == 0);
|
||||
}
|
||||
encoderd_thread();
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,250 @@
|
||||
#include "system/loggerd/logger.h"
|
||||
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
#include <ftw.h>
|
||||
|
||||
#include <cassert>
|
||||
#include <cerrno>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <ctime>
|
||||
#include <fstream>
|
||||
#include <iostream>
|
||||
#include <streambuf>
|
||||
|
||||
#include "common/params.h"
|
||||
#include "common/swaglog.h"
|
||||
#include "common/version.h"
|
||||
|
||||
// ***** log metadata *****
|
||||
kj::Array<capnp::word> logger_build_init_data() {
|
||||
MessageBuilder msg;
|
||||
auto init = msg.initEvent().initInitData();
|
||||
|
||||
init.setVersion(COMMA_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
|
||||
auto params = Params();
|
||||
std::map<std::string, std::string> params_map = params.readAll();
|
||||
|
||||
init.setGitCommit(params_map["GitCommit"]);
|
||||
init.setGitBranch(params_map["GitBranch"]);
|
||||
init.setGitRemote(params_map["GitRemote"]);
|
||||
init.setPassive(params.getBool("Passive"));
|
||||
init.setDongleId(params_map["DongleId"]);
|
||||
|
||||
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.getKeyType(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 commands = init.initCommands().initEntries(log_commands.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()));
|
||||
}
|
||||
|
||||
return capnp::messageToFlatArray(msg);
|
||||
}
|
||||
|
||||
std::string logger_get_route_name() {
|
||||
char route_name[64] = {'\0'};
|
||||
time_t rawtime = time(NULL);
|
||||
struct tm timeinfo;
|
||||
localtime_r(&rawtime, &timeinfo);
|
||||
strftime(route_name, sizeof(route_name), "%Y-%m-%d--%H-%M-%S", &timeinfo);
|
||||
return route_name;
|
||||
}
|
||||
|
||||
void log_init_data(LoggerState *s) {
|
||||
auto bytes = s->init_data.asBytes();
|
||||
logger_log(s, bytes.begin(), bytes.size(), s->has_qlog);
|
||||
}
|
||||
|
||||
|
||||
static void lh_log_sentinel(LoggerHandle *h, SentinelType type) {
|
||||
MessageBuilder msg;
|
||||
auto sen = msg.initEvent().initSentinel();
|
||||
sen.setType(type);
|
||||
sen.setSignal(h->exit_signal);
|
||||
auto bytes = msg.toBytes();
|
||||
|
||||
lh_log(h, bytes.begin(), bytes.size(), true);
|
||||
}
|
||||
|
||||
// ***** logging functions *****
|
||||
|
||||
void logger_init(LoggerState *s, bool has_qlog) {
|
||||
pthread_mutex_init(&s->lock, NULL);
|
||||
|
||||
s->part = -1;
|
||||
s->has_qlog = has_qlog;
|
||||
s->route_name = logger_get_route_name();
|
||||
s->init_data = logger_build_init_data();
|
||||
}
|
||||
|
||||
static LoggerHandle* logger_open(LoggerState *s, const char* root_path) {
|
||||
LoggerHandle *h = NULL;
|
||||
for (int i=0; i<LOGGER_MAX_HANDLES; i++) {
|
||||
if (s->handles[i].refcnt == 0) {
|
||||
h = &s->handles[i];
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert(h);
|
||||
|
||||
snprintf(h->segment_path, sizeof(h->segment_path),
|
||||
"%s/%s--%d", root_path, s->route_name.c_str(), s->part);
|
||||
|
||||
snprintf(h->log_path, sizeof(h->log_path), "%s/rlog", h->segment_path);
|
||||
snprintf(h->qlog_path, sizeof(h->qlog_path), "%s/qlog", h->segment_path);
|
||||
snprintf(h->lock_path, sizeof(h->lock_path), "%s.lock", h->log_path);
|
||||
h->end_sentinel_type = SentinelType::END_OF_SEGMENT;
|
||||
h->exit_signal = 0;
|
||||
|
||||
if (!util::create_directories(h->segment_path, 0775)) return nullptr;
|
||||
|
||||
FILE* lock_file = fopen(h->lock_path, "wb");
|
||||
if (lock_file == NULL) return NULL;
|
||||
fclose(lock_file);
|
||||
|
||||
h->log = std::make_unique<RawFile>(h->log_path);
|
||||
if (s->has_qlog) {
|
||||
h->q_log = std::make_unique<RawFile>(h->qlog_path);
|
||||
}
|
||||
|
||||
pthread_mutex_init(&h->lock, NULL);
|
||||
h->refcnt++;
|
||||
return h;
|
||||
}
|
||||
|
||||
int logger_next(LoggerState *s, const char* root_path,
|
||||
char* out_segment_path, size_t out_segment_path_len,
|
||||
int* out_part) {
|
||||
bool is_start_of_route = !s->cur_handle;
|
||||
|
||||
pthread_mutex_lock(&s->lock);
|
||||
s->part++;
|
||||
|
||||
LoggerHandle* next_h = logger_open(s, root_path);
|
||||
if (!next_h) {
|
||||
pthread_mutex_unlock(&s->lock);
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (s->cur_handle) {
|
||||
lh_close(s->cur_handle);
|
||||
}
|
||||
s->cur_handle = next_h;
|
||||
|
||||
if (out_segment_path) {
|
||||
snprintf(out_segment_path, out_segment_path_len, "%s", next_h->segment_path);
|
||||
}
|
||||
if (out_part) {
|
||||
*out_part = s->part;
|
||||
}
|
||||
|
||||
pthread_mutex_unlock(&s->lock);
|
||||
|
||||
// write beginning of log metadata
|
||||
log_init_data(s);
|
||||
lh_log_sentinel(s->cur_handle, is_start_of_route ? SentinelType::START_OF_ROUTE : SentinelType::START_OF_SEGMENT);
|
||||
return 0;
|
||||
}
|
||||
|
||||
LoggerHandle* logger_get_handle(LoggerState *s) {
|
||||
pthread_mutex_lock(&s->lock);
|
||||
LoggerHandle* h = s->cur_handle;
|
||||
if (h) {
|
||||
pthread_mutex_lock(&h->lock);
|
||||
h->refcnt++;
|
||||
pthread_mutex_unlock(&h->lock);
|
||||
}
|
||||
pthread_mutex_unlock(&s->lock);
|
||||
return h;
|
||||
}
|
||||
|
||||
void logger_log(LoggerState *s, uint8_t* data, size_t data_size, bool in_qlog) {
|
||||
pthread_mutex_lock(&s->lock);
|
||||
if (s->cur_handle) {
|
||||
lh_log(s->cur_handle, data, data_size, in_qlog);
|
||||
}
|
||||
pthread_mutex_unlock(&s->lock);
|
||||
}
|
||||
|
||||
void logger_close(LoggerState *s, ExitHandler *exit_handler) {
|
||||
pthread_mutex_lock(&s->lock);
|
||||
if (s->cur_handle) {
|
||||
s->cur_handle->exit_signal = exit_handler && exit_handler->signal.load();
|
||||
s->cur_handle->end_sentinel_type = SentinelType::END_OF_ROUTE;
|
||||
lh_close(s->cur_handle);
|
||||
}
|
||||
pthread_mutex_unlock(&s->lock);
|
||||
}
|
||||
|
||||
void lh_log(LoggerHandle* h, uint8_t* data, size_t data_size, bool in_qlog) {
|
||||
pthread_mutex_lock(&h->lock);
|
||||
assert(h->refcnt > 0);
|
||||
h->log->write(data, data_size);
|
||||
if (in_qlog && h->q_log) {
|
||||
h->q_log->write(data, data_size);
|
||||
}
|
||||
pthread_mutex_unlock(&h->lock);
|
||||
}
|
||||
|
||||
void lh_close(LoggerHandle* h) {
|
||||
pthread_mutex_lock(&h->lock);
|
||||
assert(h->refcnt > 0);
|
||||
if (h->refcnt == 1) {
|
||||
// a very ugly hack. only here can guarantee sentinel is the last msg
|
||||
pthread_mutex_unlock(&h->lock);
|
||||
lh_log_sentinel(h, h->end_sentinel_type);
|
||||
pthread_mutex_lock(&h->lock);
|
||||
}
|
||||
h->refcnt--;
|
||||
if (h->refcnt == 0) {
|
||||
h->log.reset(nullptr);
|
||||
h->q_log.reset(nullptr);
|
||||
unlink(h->lock_path);
|
||||
pthread_mutex_unlock(&h->lock);
|
||||
pthread_mutex_destroy(&h->lock);
|
||||
return;
|
||||
}
|
||||
pthread_mutex_unlock(&h->lock);
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
#pragma once
|
||||
|
||||
#include <cassert>
|
||||
#include <pthread.h>
|
||||
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <memory>
|
||||
|
||||
#include <capnp/serialize.h>
|
||||
#include <kj/array.h>
|
||||
|
||||
#include "cereal/messaging/messaging.h"
|
||||
#include "common/util.h"
|
||||
#include "common/swaglog.h"
|
||||
#include "system/hardware/hw.h"
|
||||
|
||||
const std::string LOG_ROOT = Path::log_root();
|
||||
|
||||
#define LOGGER_MAX_HANDLES 16
|
||||
|
||||
class RawFile {
|
||||
public:
|
||||
RawFile(const char* path) {
|
||||
file = util::safe_fopen(path, "wb");
|
||||
assert(file != nullptr);
|
||||
}
|
||||
~RawFile() {
|
||||
util::safe_fflush(file);
|
||||
int err = fclose(file);
|
||||
assert(err == 0);
|
||||
}
|
||||
inline void write(void* data, size_t size) {
|
||||
int written = util::safe_fwrite(data, 1, size, file);
|
||||
assert(written == size);
|
||||
}
|
||||
inline void write(kj::ArrayPtr<capnp::byte> array) { write(array.begin(), array.size()); }
|
||||
|
||||
private:
|
||||
FILE* file = nullptr;
|
||||
};
|
||||
|
||||
typedef cereal::Sentinel::SentinelType SentinelType;
|
||||
|
||||
typedef struct LoggerHandle {
|
||||
pthread_mutex_t lock;
|
||||
SentinelType end_sentinel_type;
|
||||
int exit_signal;
|
||||
int refcnt;
|
||||
char segment_path[4096];
|
||||
char log_path[4096];
|
||||
char qlog_path[4096];
|
||||
char lock_path[4096];
|
||||
std::unique_ptr<RawFile> log, q_log;
|
||||
} LoggerHandle;
|
||||
|
||||
typedef struct LoggerState {
|
||||
pthread_mutex_t lock;
|
||||
int part;
|
||||
kj::Array<capnp::word> init_data;
|
||||
std::string route_name;
|
||||
char log_name[64];
|
||||
bool has_qlog;
|
||||
|
||||
LoggerHandle handles[LOGGER_MAX_HANDLES];
|
||||
LoggerHandle* cur_handle;
|
||||
} LoggerState;
|
||||
|
||||
kj::Array<capnp::word> logger_build_init_data();
|
||||
std::string logger_get_route_name();
|
||||
void logger_init(LoggerState *s, bool has_qlog);
|
||||
int logger_next(LoggerState *s, const char* root_path,
|
||||
char* out_segment_path, size_t out_segment_path_len,
|
||||
int* out_part);
|
||||
LoggerHandle* logger_get_handle(LoggerState *s);
|
||||
void logger_close(LoggerState *s, ExitHandler *exit_handler=nullptr);
|
||||
void logger_log(LoggerState *s, uint8_t* data, size_t data_size, bool in_qlog);
|
||||
|
||||
void lh_log(LoggerHandle* h, uint8_t* data, size_t data_size, bool in_qlog);
|
||||
void lh_close(LoggerHandle* h);
|
||||
@@ -0,0 +1,286 @@
|
||||
#include "system/loggerd/loggerd.h"
|
||||
#include "system/loggerd/video_writer.h"
|
||||
|
||||
ExitHandler do_exit;
|
||||
|
||||
struct LoggerdState {
|
||||
LoggerState logger = {};
|
||||
char segment_path[4096];
|
||||
std::atomic<int> rotate_segment;
|
||||
std::atomic<double> last_camera_seen_tms;
|
||||
std::atomic<int> ready_to_rotate; // 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) {
|
||||
int segment = -1;
|
||||
int err = logger_next(&s->logger, LOG_ROOT.c_str(), s->segment_path, sizeof(s->segment_path), &segment);
|
||||
assert(err == 0);
|
||||
s->rotate_segment = segment;
|
||||
s->ready_to_rotate = 0;
|
||||
s->last_rotate_tms = millis_since_boot();
|
||||
LOGW((s->logger.part == 0) ? "logging to %s" : "rotated to %s", s->segment_path);
|
||||
}
|
||||
|
||||
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;
|
||||
};
|
||||
|
||||
int handle_encoder_msg(LoggerdState *s, Message *msg, std::string &name, struct RemoteEncoder &re) {
|
||||
const LogCameraInfo &cam_info = (name == "driverEncodeData") ? cameras_logged[1] :
|
||||
((name == "wideRoadEncodeData") ? cameras_logged[2] :
|
||||
((name == "qRoadEncodeData") ? qcam_info : cameras_logged[0]));
|
||||
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 = (name == "driverEncodeData") ? event.getDriverEncodeData() :
|
||||
((name == "wideRoadEncodeData") ? event.getWideRoadEncodeData() :
|
||||
((name == "qRoadEncodeData") ? event.getQRoadEncodeData() : event.getRoadEncodeData()));
|
||||
auto idx = edata.getIdx();
|
||||
auto flags = idx.getFlags();
|
||||
|
||||
// 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->rotate_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->rotate_segment) {
|
||||
if (re.recording) {
|
||||
re.writer.reset();
|
||||
re.recording = false;
|
||||
}
|
||||
re.current_segment = s->rotate_segment;
|
||||
re.marked_ready_to_rotate = false;
|
||||
// we are in this segment now, process any queued messages before this one
|
||||
if (!re.q.empty()) {
|
||||
for (auto &qmsg: re.q) {
|
||||
bytes_count += handle_encoder_msg(s, qmsg, name, re);
|
||||
}
|
||||
re.q.clear();
|
||||
}
|
||||
}
|
||||
|
||||
// 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", name.c_str(), re.dropped_frames);
|
||||
re.dropped_frames = 0;
|
||||
}
|
||||
// if we aren't actually recording, don't create the writer
|
||||
if (cam_info.record) {
|
||||
re.writer.reset(new VideoWriter(s->segment_path,
|
||||
cam_info.filename, idx.getType() != cereal::EncodeIndex::Type::FULL_H_E_V_C,
|
||||
cam_info.frame_width, cam_info.frame_height, cam_info.fps, idx.getType()));
|
||||
// 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
|
||||
delete msg;
|
||||
++re.dropped_frames;
|
||||
return bytes_count;
|
||||
}
|
||||
}
|
||||
|
||||
// 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());
|
||||
if (name == "driverEncodeData") { evt.setDriverEncodeIdx(idx); }
|
||||
if (name == "wideRoadEncodeData") { evt.setWideRoadEncodeIdx(idx); }
|
||||
if (name == "qRoadEncodeData") { evt.setQRoadEncodeIdx(idx); }
|
||||
if (name == "roadEncodeData") { evt.setRoadEncodeIdx(idx); }
|
||||
auto new_msg = bmsg.toBytes();
|
||||
logger_log(&s->logger, (uint8_t *)new_msg.begin(), new_msg.size(), true); // always in qlog?
|
||||
bytes_count += new_msg.size();
|
||||
|
||||
// free the message, we used it
|
||||
delete msg;
|
||||
} else if (offset_segment_num > s->rotate_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->rotate_segment.load(), offset_segment_num,
|
||||
s->ready_to_rotate.load(), s->max_waiting, name.c_str());
|
||||
}
|
||||
// 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->rotate_segment:%d re.encoderd_segment_offset:%d",
|
||||
name.c_str(), idx.getSegmentNum(), s->rotate_segment.load(), 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->rotate_segment.load();
|
||||
delete msg;
|
||||
}
|
||||
|
||||
return bytes_count;
|
||||
}
|
||||
|
||||
void loggerd_thread() {
|
||||
// setup messaging
|
||||
typedef struct QlogState {
|
||||
std::string name;
|
||||
int counter, freq;
|
||||
bool encoder;
|
||||
} QlogState;
|
||||
std::unordered_map<SubSocket*, QlogState> qlog_states;
|
||||
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 = strcmp(it.name+strlen(it.name)-strlen("EncodeData"), "EncodeData") == 0;
|
||||
if (!it.should_log && !encoder) continue;
|
||||
LOGD("logging %s (on port %d)", it.name, it.port);
|
||||
|
||||
SubSocket * sock = SubSocket::create(ctx.get(), it.name);
|
||||
assert(sock != NULL);
|
||||
poller->registerSocket(sock);
|
||||
qlog_states[sock] = {
|
||||
.name = it.name,
|
||||
.counter = 0,
|
||||
.freq = it.decimation,
|
||||
.encoder = encoder,
|
||||
};
|
||||
}
|
||||
|
||||
LoggerdState s;
|
||||
// init logger
|
||||
logger_init(&s.logger, true);
|
||||
logger_rotate(&s);
|
||||
Params().put("CurrentRoute", s.logger.route_name);
|
||||
|
||||
// init encoders
|
||||
s.last_camera_seen_tms = millis_since_boot();
|
||||
for (const auto &cam : cameras_logged) {
|
||||
s.max_waiting++;
|
||||
if (cam.has_qcamera) { s.max_waiting++; }
|
||||
}
|
||||
|
||||
uint64_t msg_count = 0, bytes_count = 0;
|
||||
double start_ts = millis_since_boot();
|
||||
while (!do_exit) {
|
||||
// poll for new messages on all sockets
|
||||
for (auto sock : poller->poll(1000)) {
|
||||
if (do_exit) break;
|
||||
|
||||
// drain socket
|
||||
int count = 0;
|
||||
QlogState &qs = qlog_states[sock];
|
||||
Message *msg = nullptr;
|
||||
while (!do_exit && (msg = sock->receive(true))) {
|
||||
const bool in_qlog = qs.freq != -1 && (qs.counter++ % qs.freq == 0);
|
||||
|
||||
if (qs.encoder) {
|
||||
s.last_camera_seen_tms = millis_since_boot();
|
||||
bytes_count += handle_encoder_msg(&s, msg, qs.name, remote_encoders[sock]);
|
||||
} else {
|
||||
logger_log(&s.logger, (uint8_t *)msg->getData(), msg->getSize(), in_qlog);
|
||||
bytes_count += msg->getSize();
|
||||
delete msg;
|
||||
}
|
||||
|
||||
rotate_if_needed(&s);
|
||||
|
||||
if ((++msg_count % 1000) == 0) {
|
||||
double seconds = (millis_since_boot() - start_ts) / 1000.0;
|
||||
LOGD("%lu 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", qs.name.c_str());
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
LOGW("closing logger");
|
||||
logger_close(&s.logger, &do_exit);
|
||||
|
||||
if (do_exit.power_failure) {
|
||||
LOGE("power failure");
|
||||
sync();
|
||||
LOGE("sync done");
|
||||
}
|
||||
|
||||
// messaging cleanup
|
||||
for (auto &[sock, qs] : qlog_states) delete sock;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
#pragma once
|
||||
|
||||
#include <unistd.h>
|
||||
|
||||
#include <atomic>
|
||||
#include <cassert>
|
||||
#include <cerrno>
|
||||
#include <condition_variable>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <unordered_map>
|
||||
#include <utility>
|
||||
|
||||
#include "cereal/messaging/messaging.h"
|
||||
#include "cereal/services.h"
|
||||
#include "cereal/visionipc/visionipc.h"
|
||||
#include "cereal/visionipc/visionipc_client.h"
|
||||
#include "system/camerad/cameras/camera_common.h"
|
||||
#include "common/params.h"
|
||||
#include "common/swaglog.h"
|
||||
#include "common/timing.h"
|
||||
#include "common/util.h"
|
||||
#include "system/hardware/hw.h"
|
||||
|
||||
#include "system/loggerd/encoder/encoder.h"
|
||||
#include "system/loggerd/logger.h"
|
||||
#ifdef QCOM2
|
||||
#include "system/loggerd/encoder/v4l_encoder.h"
|
||||
#define Encoder V4LEncoder
|
||||
#else
|
||||
#include "system/loggerd/encoder/ffmpeg_encoder.h"
|
||||
#define Encoder FfmpegEncoder
|
||||
#endif
|
||||
|
||||
constexpr int MAIN_FPS = 20;
|
||||
const int MAIN_BITRATE = 10000000;
|
||||
const int DCAM_BITRATE = MAIN_BITRATE;
|
||||
|
||||
#define NO_CAMERA_PATIENCE 500 // fall back to time-based rotation if all cameras are dead
|
||||
|
||||
const bool LOGGERD_TEST = getenv("LOGGERD_TEST");
|
||||
const int SEGMENT_LENGTH = LOGGERD_TEST ? atoi(getenv("LOGGERD_SEGMENT_LENGTH")) : 60;
|
||||
|
||||
struct LogCameraInfo {
|
||||
CameraType type;
|
||||
const char *filename;
|
||||
VisionStreamType stream_type;
|
||||
int frame_width, frame_height;
|
||||
int fps;
|
||||
int bitrate;
|
||||
bool is_h265;
|
||||
bool has_qcamera;
|
||||
bool record;
|
||||
};
|
||||
|
||||
const LogCameraInfo cameras_logged[] = {
|
||||
{
|
||||
.type = RoadCam,
|
||||
.stream_type = VISION_STREAM_ROAD,
|
||||
.filename = "fcamera.hevc",
|
||||
.fps = MAIN_FPS,
|
||||
.bitrate = MAIN_BITRATE,
|
||||
.is_h265 = true,
|
||||
.has_qcamera = true,
|
||||
.record = true,
|
||||
.frame_width = 1928,
|
||||
.frame_height = 1208,
|
||||
},
|
||||
{
|
||||
.type = DriverCam,
|
||||
.stream_type = VISION_STREAM_DRIVER,
|
||||
.filename = "dcamera.hevc",
|
||||
.fps = MAIN_FPS,
|
||||
.bitrate = DCAM_BITRATE,
|
||||
.is_h265 = true,
|
||||
.has_qcamera = false,
|
||||
.record = Params().getBool("RecordFront"),
|
||||
.frame_width = 1928,
|
||||
.frame_height = 1208,
|
||||
},
|
||||
{
|
||||
.type = WideRoadCam,
|
||||
.stream_type = VISION_STREAM_WIDE_ROAD,
|
||||
.filename = "ecamera.hevc",
|
||||
.fps = MAIN_FPS,
|
||||
.bitrate = MAIN_BITRATE,
|
||||
.is_h265 = true,
|
||||
.has_qcamera = false,
|
||||
.record = true,
|
||||
.frame_width = 1928,
|
||||
.frame_height = 1208,
|
||||
},
|
||||
};
|
||||
const LogCameraInfo qcam_info = {
|
||||
.filename = "qcamera.ts",
|
||||
.fps = MAIN_FPS,
|
||||
.bitrate = 256000,
|
||||
.is_h265 = false,
|
||||
.record = true,
|
||||
.frame_width = 526,
|
||||
.frame_height = 330,
|
||||
};
|
||||
Executable
+26
@@ -0,0 +1,26 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Script to fill up EON with fake data"""
|
||||
|
||||
import os
|
||||
|
||||
from system.loggerd.config import ROOT, get_available_percent
|
||||
from system.loggerd.tests.loggerd_tests_common import create_random_file
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
segment_idx = 0
|
||||
while True:
|
||||
seg_name = "1970-01-01--00-00-00--%d" % segment_idx
|
||||
seg_path = os.path.join(ROOT, seg_name)
|
||||
|
||||
print(seg_path)
|
||||
|
||||
create_random_file(os.path.join(seg_path, 'fcamera.hevc'), 36)
|
||||
create_random_file(os.path.join(seg_path, 'rlog.bz2'), 2)
|
||||
|
||||
segment_idx += 1
|
||||
|
||||
# Fill up to 99 percent
|
||||
available_percent = get_available_percent()
|
||||
if available_percent < 1.0:
|
||||
break
|
||||
@@ -0,0 +1,102 @@
|
||||
import os
|
||||
import errno
|
||||
import shutil
|
||||
import random
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
import system.loggerd.uploader as uploader
|
||||
|
||||
def create_random_file(file_path, size_mb, lock=False):
|
||||
try:
|
||||
os.mkdir(os.path.dirname(file_path))
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
lock_path = file_path + ".lock"
|
||||
if 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)
|
||||
|
||||
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 MockParams():
|
||||
def __init__(self):
|
||||
self.params = {
|
||||
"DongleId": b"0000000000000000",
|
||||
"IsOffroad": b"1",
|
||||
}
|
||||
|
||||
def get(self, k, block=False, encoding=None):
|
||||
val = self.params[k]
|
||||
|
||||
if encoding is not None:
|
||||
return val.decode(encoding)
|
||||
else:
|
||||
return val
|
||||
|
||||
def get_bool(self, k):
|
||||
val = self.params[k]
|
||||
return (val == b'1')
|
||||
|
||||
class UploaderTestCase(unittest.TestCase):
|
||||
f_type = "UNKNOWN"
|
||||
|
||||
def set_ignore(self):
|
||||
uploader.Api = MockApiIgnore
|
||||
|
||||
def setUp(self):
|
||||
self.root = tempfile.mkdtemp()
|
||||
uploader.ROOT = self.root # Monkey patch root dir
|
||||
uploader.Api = MockApi
|
||||
uploader.Params = MockParams
|
||||
uploader.fake_upload = True
|
||||
uploader.force_wifi = True
|
||||
uploader.allow_sleep = False
|
||||
self.seg_num = random.randint(1, 300)
|
||||
self.seg_format = "2019-04-18--12-52-54--{}"
|
||||
self.seg_format2 = "2019-05-18--11-22-33--{}"
|
||||
self.seg_dir = self.seg_format.format(self.seg_num)
|
||||
|
||||
def tearDown(self):
|
||||
try:
|
||||
shutil.rmtree(self.root)
|
||||
except OSError as e:
|
||||
if e.errno != errno.ENOENT:
|
||||
raise
|
||||
|
||||
def make_file_with_data(self, f_dir, fn, size_mb=.1, lock=False):
|
||||
file_path = os.path.join(self.root, f_dir, fn)
|
||||
create_random_file(file_path, size_mb, lock)
|
||||
|
||||
return file_path
|
||||
Executable
+105
@@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
import time
|
||||
import threading
|
||||
import unittest
|
||||
from collections import namedtuple
|
||||
|
||||
from common.timeout import Timeout, TimeoutException
|
||||
import system.loggerd.deleter as deleter
|
||||
from 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(self):
|
||||
self.f_type = "fcamera.hevc"
|
||||
super().setUp()
|
||||
self.fake_stats = Stats(f_bavail=0, f_blocks=10, f_frsize=4096)
|
||||
deleter.os.statvfs = self.fake_statvfs
|
||||
deleter.ROOT = self.root
|
||||
|
||||
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()
|
||||
|
||||
with Timeout(5, "Timeout waiting for file to be deleted"):
|
||||
while os.path.exists(f_path):
|
||||
time.sleep(0.01)
|
||||
self.join_thread()
|
||||
|
||||
self.assertFalse(os.path.exists(f_path), "File not deleted")
|
||||
|
||||
def test_delete_files_in_create_order(self):
|
||||
f_path_1 = self.make_file_with_data(self.seg_dir, self.f_type)
|
||||
time.sleep(1)
|
||||
self.seg_num += 1
|
||||
self.seg_dir = self.seg_format.format(self.seg_num)
|
||||
f_path_2 = self.make_file_with_data(self.seg_dir, self.f_type)
|
||||
|
||||
self.start_thread()
|
||||
|
||||
with Timeout(5, "Timeout waiting for file to be deleted"):
|
||||
while os.path.exists(f_path_1) and os.path.exists(f_path_2):
|
||||
time.sleep(0.01)
|
||||
|
||||
self.join_thread()
|
||||
|
||||
self.assertFalse(os.path.exists(f_path_1), "Older file not deleted")
|
||||
|
||||
self.assertTrue(os.path.exists(f_path_2), "Newer file deleted before older file")
|
||||
|
||||
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()
|
||||
|
||||
try:
|
||||
with Timeout(2, "Timeout waiting for file to be deleted"):
|
||||
while os.path.exists(f_path):
|
||||
time.sleep(0.01)
|
||||
except TimeoutException:
|
||||
pass
|
||||
finally:
|
||||
self.join_thread()
|
||||
|
||||
self.assertTrue(os.path.exists(f_path), "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()
|
||||
|
||||
try:
|
||||
with Timeout(2, "Timeout waiting for file to be deleted"):
|
||||
while os.path.exists(f_path):
|
||||
time.sleep(0.01)
|
||||
except TimeoutException:
|
||||
pass
|
||||
finally:
|
||||
self.join_thread()
|
||||
|
||||
self.assertTrue(os.path.exists(f_path), "File deleted when locked")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Executable
+160
@@ -0,0 +1,160 @@
|
||||
#!/usr/bin/env python3
|
||||
import math
|
||||
import os
|
||||
import random
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from parameterized import parameterized
|
||||
from tqdm import trange
|
||||
|
||||
from common.params import Params
|
||||
from common.timeout import Timeout
|
||||
from system.hardware import TICI
|
||||
from system.loggerd.config import ROOT
|
||||
from selfdrive.manager.process_config import managed_processes
|
||||
from tools.lib.logreader import LogReader
|
||||
|
||||
SEGMENT_LENGTH = 2
|
||||
FULL_SIZE = 2507572
|
||||
CAMERAS = [
|
||||
("fcamera.hevc", 20, FULL_SIZE, "roadEncodeIdx"),
|
||||
("dcamera.hevc", 20, FULL_SIZE, "driverEncodeIdx"),
|
||||
("ecamera.hevc", 20, FULL_SIZE, "wideRoadEncodeIdx"),
|
||||
("qcamera.ts", 20, 130000, None),
|
||||
]
|
||||
|
||||
# we check frame count, so we don't have to be too strict on size
|
||||
FILE_SIZE_TOLERANCE = 0.5
|
||||
|
||||
|
||||
class TestEncoder(unittest.TestCase):
|
||||
|
||||
# TODO: all of loggerd should work on PC
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if not TICI:
|
||||
raise unittest.SkipTest
|
||||
|
||||
def setUp(self):
|
||||
self._clear_logs()
|
||||
os.environ["LOGGERD_TEST"] = "1"
|
||||
os.environ["LOGGERD_SEGMENT_LENGTH"] = str(SEGMENT_LENGTH)
|
||||
|
||||
def tearDown(self):
|
||||
self._clear_logs()
|
||||
|
||||
def _clear_logs(self):
|
||||
if os.path.exists(ROOT):
|
||||
shutil.rmtree(ROOT)
|
||||
|
||||
def _get_latest_segment_path(self):
|
||||
last_route = sorted(Path(ROOT).iterdir())[-1]
|
||||
return os.path.join(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(10, 15)))
|
||||
|
||||
# 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, 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
|
||||
self.assertTrue(os.path.exists(file_path), f"segment #{i}: '{file_path}' missing")
|
||||
|
||||
# TODO: this ffprobe call is really slow
|
||||
# check frame count
|
||||
cmd = f"ffprobe -v error -select_streams v:0 -count_packets -show_entries stream=nb_read_packets -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')
|
||||
frame_count = int(probe.split('\n')[0].strip())
|
||||
counts.append(frame_count)
|
||||
|
||||
self.assertEqual(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)
|
||||
self.assertTrue(math.isclose(file_size, size, rel_tol=FILE_SIZE_TOLERANCE),
|
||||
f"{file_path} size {file_size} isn't close to target size {size}")
|
||||
|
||||
# Check encodeIdx
|
||||
if encode_idx_name is not None:
|
||||
rlog_path = f"{route_prefix_path}--{i}/rlog"
|
||||
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
|
||||
self.assertEqual(frame_count, len(segment_idxs))
|
||||
self.assertEqual(frame_count, len(encode_idxs))
|
||||
|
||||
# Check for duplicates or skips
|
||||
self.assertEqual(0, segment_idxs[0])
|
||||
self.assertEqual(len(set(segment_idxs)), len(segment_idxs))
|
||||
|
||||
self.assertTrue(all(valid))
|
||||
|
||||
self.assertEqual(expected_frames * i, encode_idxs[0])
|
||||
first_frames.append(frame_idxs[0])
|
||||
self.assertEqual(len(set(encode_idxs)), len(encode_idxs))
|
||||
|
||||
self.assertEqual(1, len(set(first_frames)))
|
||||
|
||||
if TICI:
|
||||
expected_frames = fps * SEGMENT_LENGTH
|
||||
self.assertEqual(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(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()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,147 @@
|
||||
#include <sys/stat.h>
|
||||
|
||||
#include <climits>
|
||||
#include <condition_variable>
|
||||
#include <sstream>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
|
||||
#include "catch2/catch.hpp"
|
||||
#include "cereal/messaging/messaging.h"
|
||||
#include "common/util.h"
|
||||
#include "system/loggerd/logger.h"
|
||||
#include "tools/replay/util.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", "/qlog"}) {
|
||||
const std::string log_file = segment_path + fn;
|
||||
std::string log = util::read_file(log_file);
|
||||
REQUIRE(!log.empty());
|
||||
int event_cnt = 0, i = 0;
|
||||
kj::ArrayPtr<const capnp::word> words((capnp::word *)log.data(), 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(LoggerHandle *logger) {
|
||||
MessageBuilder msg;
|
||||
msg.initEvent().initClocks();
|
||||
auto bytes = msg.toBytes();
|
||||
lh_log(logger, bytes.begin(), bytes.size(), true);
|
||||
}
|
||||
|
||||
TEST_CASE("logger") {
|
||||
const std::string log_root = "/tmp/test_logger";
|
||||
system(("rm " + log_root + " -rf").c_str());
|
||||
|
||||
ExitHandler do_exit;
|
||||
|
||||
LoggerState logger = {};
|
||||
logger_init(&logger, true);
|
||||
char segment_path[PATH_MAX] = {};
|
||||
int segment = -1;
|
||||
|
||||
SECTION("single thread logging & rotation(100 segments, one thread)") {
|
||||
const int segment_cnt = 100;
|
||||
for (int i = 0; i < segment_cnt; ++i) {
|
||||
REQUIRE(logger_next(&logger, log_root.c_str(), segment_path, sizeof(segment_path), &segment) == 0);
|
||||
REQUIRE(util::file_exists(std::string(segment_path) + "/rlog.lock"));
|
||||
REQUIRE(segment == i);
|
||||
write_msg(logger.cur_handle);
|
||||
}
|
||||
do_exit = true;
|
||||
do_exit.signal = 1;
|
||||
logger_close(&logger, &do_exit);
|
||||
for (int i = 0; i < segment_cnt; ++i) {
|
||||
verify_segment(log_root + "/" + logger.route_name, i, segment_cnt, 1);
|
||||
}
|
||||
}
|
||||
SECTION("multiple threads logging & rotation(100 segments, 10 threads") {
|
||||
const int segment_cnt = 100, thread_cnt = 10;
|
||||
std::atomic<int> event_cnt[segment_cnt] = {};
|
||||
std::atomic<int> main_segment = -1;
|
||||
|
||||
auto logging_thread = [&]() -> void {
|
||||
LoggerHandle *lh = logger_get_handle(&logger);
|
||||
REQUIRE(lh != nullptr);
|
||||
int segment = main_segment;
|
||||
int delayed_cnt = 0;
|
||||
while (!do_exit) {
|
||||
// write 2 more messages in the current segment and then rotate to the new segment.
|
||||
if (main_segment > segment && ++delayed_cnt == 2) {
|
||||
lh_close(lh);
|
||||
lh = logger_get_handle(&logger);
|
||||
segment = main_segment;
|
||||
delayed_cnt = 0;
|
||||
}
|
||||
write_msg(lh);
|
||||
event_cnt[segment] += 1;
|
||||
usleep(1);
|
||||
}
|
||||
lh_close(lh);
|
||||
};
|
||||
|
||||
// start logging
|
||||
std::vector<std::thread> threads;
|
||||
for (int i = 0; i < segment_cnt; ++i) {
|
||||
REQUIRE(logger_next(&logger, log_root.c_str(), segment_path, sizeof(segment_path), &segment) == 0);
|
||||
REQUIRE(segment == i);
|
||||
main_segment = segment;
|
||||
if (i == 0) {
|
||||
for (int j = 0; j < thread_cnt; ++j) {
|
||||
threads.push_back(std::thread(logging_thread));
|
||||
}
|
||||
}
|
||||
for (int j = 0; j < 100; ++j) {
|
||||
write_msg(logger.cur_handle);
|
||||
usleep(1);
|
||||
}
|
||||
event_cnt[segment] += 100;
|
||||
}
|
||||
|
||||
// end logging
|
||||
for (auto &t : threads) t.join();
|
||||
do_exit = true;
|
||||
do_exit.signal = 1;
|
||||
logger_close(&logger, &do_exit);
|
||||
REQUIRE(logger.cur_handle->refcnt == 0);
|
||||
|
||||
for (int i = 0; i < segment_cnt; ++i) {
|
||||
verify_segment(log_root + "/" + logger.route_name, i, segment_cnt, event_cnt[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
Executable
+280
@@ -0,0 +1,280 @@
|
||||
#!/usr/bin/env python3
|
||||
import numpy as np
|
||||
import os
|
||||
import random
|
||||
import string
|
||||
import subprocess
|
||||
import time
|
||||
import unittest
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
import cereal.messaging as messaging
|
||||
from cereal import log
|
||||
from cereal.services import service_list
|
||||
from common.basedir import BASEDIR
|
||||
from common.params import Params
|
||||
from common.timeout import Timeout
|
||||
from system.loggerd.config import ROOT
|
||||
from selfdrive.manager.process_config import managed_processes
|
||||
from system.version import get_version
|
||||
from tools.lib.logreader import LogReader
|
||||
from cereal.visionipc import VisionIpcServer, VisionStreamType
|
||||
from common.transformations.camera import tici_f_frame_size, tici_d_frame_size, tici_e_frame_size
|
||||
|
||||
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(unittest.TestCase):
|
||||
def _get_latest_log_dir(self):
|
||||
log_dirs = sorted(Path(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]
|
||||
self.assertEqual(msg.which(), 'initData')
|
||||
|
||||
def _check_sentinel(self, msgs, route):
|
||||
start_type = SentinelType.startOfRoute if route else SentinelType.startOfSegment
|
||||
self.assertTrue(msgs[1].sentinel.type == start_type)
|
||||
|
||||
end_type = SentinelType.endOfRoute if route else SentinelType.endOfSegment
|
||||
self.assertTrue(msgs[-1].sentinel.type == end_type)
|
||||
|
||||
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"),
|
||||
("GitBranch", "gitBranch", "branch"),
|
||||
("GitRemote", "gitRemote", "remote"),
|
||||
]
|
||||
params = Params()
|
||||
params.clear_all()
|
||||
for k, _, v in fake_params:
|
||||
params.put(k, v)
|
||||
params.put("LaikadEphemerisV2", "abc")
|
||||
|
||||
lr = list(LogReader(str(self._gen_bootlog())))
|
||||
initData = lr[0].initData
|
||||
|
||||
self.assertTrue(initData.dirty != bool(os.environ["CLEAN"]))
|
||||
self.assertEqual(initData.version, get_version())
|
||||
|
||||
if os.path.isfile("/proc/cmdline"):
|
||||
with open("/proc/cmdline") as f:
|
||||
self.assertEqual(list(initData.kernelArgs), f.read().strip().split(" "))
|
||||
|
||||
with open("/proc/version") as f:
|
||||
self.assertEqual(initData.kernelVersion, f.read())
|
||||
|
||||
# check params
|
||||
logged_params = {entry.key: entry.value for entry in initData.params.entries}
|
||||
expected_params = set(k for k, _, __ in fake_params) | {'LaikadEphemerisV2'}
|
||||
assert set(logged_params.keys()) == expected_params, set(logged_params.keys()) ^ expected_params
|
||||
assert logged_params['LaikadEphemerisV2'] == b'', f"DONT_LOG param value was logged: {repr(logged_params['LaikadEphemerisV2'])}"
|
||||
for param_key, initData_key, v in fake_params:
|
||||
self.assertEqual(getattr(initData, initData_key), v)
|
||||
self.assertEqual(logged_params[param_key].decode(), v)
|
||||
|
||||
params.put("LaikadEphemerisV2", "")
|
||||
|
||||
def test_rotation(self):
|
||||
os.environ["LOGGERD_TEST"] = "1"
|
||||
Params().put("RecordFront", "1")
|
||||
|
||||
expected_files = {"rlog", "qlog", "qcamera.ts", "fcamera.hevc", "dcamera.hevc", "ecamera.hevc"}
|
||||
streams = [(VisionStreamType.VISION_STREAM_ROAD, (*tici_f_frame_size, 2048*2346, 2048, 2048*1216), "roadCameraState"),
|
||||
(VisionStreamType.VISION_STREAM_DRIVER, (*tici_d_frame_size, 2048*2346, 2048, 2048*1216), "driverCameraState"),
|
||||
(VisionStreamType.VISION_STREAM_WIDE_ROAD, (*tici_e_frame_size, 2048*2346, 2048, 2048*1216), "wideRoadCameraState")]
|
||||
|
||||
pm = messaging.PubMaster(["roadCameraState", "driverCameraState", "wideRoadCameraState"])
|
||||
vipc_server = VisionIpcServer("camerad")
|
||||
for stream_type, frame_spec, _ in streams:
|
||||
vipc_server.create_buffers_with_sizes(stream_type, 40, False, *(frame_spec))
|
||||
vipc_server.start_listener()
|
||||
|
||||
for _ in range(5):
|
||||
num_segs = random.randint(2, 5)
|
||||
length = random.randint(1, 3)
|
||||
os.environ["LOGGERD_SEGMENT_LENGTH"] = str(length)
|
||||
managed_processes["loggerd"].start()
|
||||
managed_processes["encoderd"].start()
|
||||
|
||||
fps = 20.0
|
||||
for n in range(1, int(num_segs*length*fps)+1):
|
||||
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)
|
||||
time.sleep(1.0/fps)
|
||||
|
||||
managed_processes["loggerd"].stop()
|
||||
managed_processes["encoderd"].stop()
|
||||
|
||||
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
|
||||
self.assertEqual(len(diff), 0, f"didn't get all expected files. run={_} 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
|
||||
|
||||
for fn in ["console-ramoops", "pmsg-ramoops-0"]:
|
||||
path = Path(os.path.join("/sys/fs/pstore/", fn))
|
||||
if path.is_file():
|
||||
expected_val = open(path, "rb").read()
|
||||
bootlog_val = [e.value for e in boot.pstore.entries if e.key == fn][0]
|
||||
self.assertEqual(expected_val, bootlog_val)
|
||||
|
||||
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))))
|
||||
|
||||
pm = messaging.PubMaster(services)
|
||||
|
||||
# sleep enough for the first poll to time out
|
||||
# TODO: fix loggerd bug dropping the msgs from the first poll
|
||||
managed_processes["loggerd"].start()
|
||||
for s in services:
|
||||
while not pm.all_readers_updated(s):
|
||||
time.sleep(0.1)
|
||||
|
||||
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)
|
||||
time.sleep(0.01)
|
||||
|
||||
time.sleep(1)
|
||||
managed_processes["loggerd"].stop()
|
||||
|
||||
qlog_path = os.path.join(self._get_latest_log_dir(), "qlog")
|
||||
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
|
||||
self.assertEqual(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
|
||||
self.assertEqual(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))
|
||||
pm = messaging.PubMaster(services)
|
||||
|
||||
# sleep enough for the first poll to time out
|
||||
# TODO: fix loggerd bug dropping the msgs from the first poll
|
||||
managed_processes["loggerd"].start()
|
||||
for s in services:
|
||||
while not pm.all_readers_updated(s):
|
||||
time.sleep(0.1)
|
||||
|
||||
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)
|
||||
|
||||
time.sleep(2)
|
||||
managed_processes["loggerd"].stop()
|
||||
|
||||
lr = list(LogReader(os.path.join(self._get_latest_log_dir(), "rlog")))
|
||||
|
||||
# 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()
|
||||
self.assertEqual(sent.to_bytes(), m.as_builder().to_bytes())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,2 @@
|
||||
#define CATCH_CONFIG_MAIN
|
||||
#include "catch2/catch.hpp"
|
||||
Executable
+158
@@ -0,0 +1,158 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
import time
|
||||
import threading
|
||||
import unittest
|
||||
import logging
|
||||
import json
|
||||
|
||||
from system.swaglog import cloudlog
|
||||
import system.loggerd.uploader as uploader
|
||||
|
||||
from system.loggerd.tests.loggerd_tests_common import UploaderTestCase
|
||||
|
||||
|
||||
class TestLogHandler(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 = TestLogHandler()
|
||||
cloudlog.addHandler(log_handler)
|
||||
|
||||
|
||||
class TestUploader(UploaderTestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
log_handler.reset()
|
||||
|
||||
def start_thread(self):
|
||||
self.end_event = threading.Event()
|
||||
self.up_thread = threading.Thread(target=uploader.uploader_fn, 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, boot=True):
|
||||
f_paths = list()
|
||||
for t in ["qlog", "rlog", "dcamera.hevc", "fcamera.hevc"]:
|
||||
f_paths.append(self.make_file_with_data(self.seg_dir, t, 1, lock=lock))
|
||||
|
||||
if boot:
|
||||
f_paths.append(self.make_file_with_data("boot", f"{self.seg_dir}", 1, lock=lock))
|
||||
return f_paths
|
||||
|
||||
def gen_order(self, seg1, seg2, boot=True):
|
||||
keys = []
|
||||
if boot:
|
||||
keys += [f"boot/{self.seg_format.format(i)}.bz2" for i in seg1]
|
||||
keys += [f"boot/{self.seg_format2.format(i)}.bz2" for i in seg2]
|
||||
keys += [f"{self.seg_format.format(i)}/qlog.bz2" for i in seg1]
|
||||
keys += [f"{self.seg_format2.format(i)}/qlog.bz2" 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(5)
|
||||
self.join_thread()
|
||||
|
||||
exp_order = self.gen_order([self.seg_num], [])
|
||||
|
||||
self.assertTrue(len(log_handler.upload_ignored) == 0, "Some files were ignored")
|
||||
self.assertFalse(len(log_handler.upload_order) < len(exp_order), "Some files failed to upload")
|
||||
self.assertFalse(len(log_handler.upload_order) > len(exp_order), "Some files were uploaded twice")
|
||||
for f_path in exp_order:
|
||||
self.assertTrue(os.getxattr(os.path.join(self.root, f_path.replace('.bz2', '')), uploader.UPLOAD_ATTR_NAME), "All files not uploaded")
|
||||
|
||||
self.assertTrue(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(5)
|
||||
self.join_thread()
|
||||
|
||||
exp_order = self.gen_order([self.seg_num], [])
|
||||
|
||||
self.assertTrue(len(log_handler.upload_order) == 0, "Some files were not ignored")
|
||||
self.assertFalse(len(log_handler.upload_ignored) < len(exp_order), "Some files failed to ignore")
|
||||
self.assertFalse(len(log_handler.upload_ignored) > len(exp_order), "Some files were ignored twice")
|
||||
for f_path in exp_order:
|
||||
self.assertTrue(os.getxattr(os.path.join(self.root, f_path.replace('.bz2', '')), uploader.UPLOAD_ATTR_NAME), "All files not ignored")
|
||||
|
||||
self.assertTrue(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(5)
|
||||
self.join_thread()
|
||||
|
||||
self.assertTrue(len(log_handler.upload_ignored) == 0, "Some files were ignored")
|
||||
self.assertFalse(len(log_handler.upload_order) < len(exp_order), "Some files failed to upload")
|
||||
self.assertFalse(len(log_handler.upload_order) > len(exp_order), "Some files were uploaded twice")
|
||||
for f_path in exp_order:
|
||||
self.assertTrue(os.getxattr(os.path.join(self.root, f_path.replace('.bz2', '')), uploader.UPLOAD_ATTR_NAME), "All files not uploaded")
|
||||
|
||||
self.assertTrue(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(5)
|
||||
self.join_thread()
|
||||
|
||||
for f_path in f_paths:
|
||||
uploaded = uploader.UPLOAD_ATTR_NAME in os.listxattr(f_path.replace('.bz2', ''))
|
||||
self.assertFalse(uploaded, "File upload when locked")
|
||||
|
||||
def test_clear_locks_on_startup(self):
|
||||
f_paths = self.gen_files(lock=True, boot=False)
|
||||
self.start_thread()
|
||||
time.sleep(1)
|
||||
self.join_thread()
|
||||
|
||||
for f_path in f_paths:
|
||||
self.assertFalse(os.path.isfile(f_path + ".lock"), "File lock not cleared on startup")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,8 @@
|
||||
import os
|
||||
from system.loggerd.uploader import UPLOAD_ATTR_NAME, UPLOAD_ATTR_VALUE
|
||||
|
||||
from system.loggerd.config import ROOT
|
||||
for folder in os.walk(ROOT):
|
||||
for file1 in folder[2]:
|
||||
full_path = os.path.join(folder[0], file1)
|
||||
os.setxattr(full_path, UPLOAD_ATTR_NAME, UPLOAD_ATTR_VALUE)
|
||||
Executable
+8
@@ -0,0 +1,8 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
import sys
|
||||
from system.loggerd.uploader import UPLOAD_ATTR_NAME
|
||||
|
||||
for fn in sys.argv[1:]:
|
||||
print(f"unmarking {fn}")
|
||||
os.removexattr(fn, UPLOAD_ATTR_NAME)
|
||||
@@ -0,0 +1,287 @@
|
||||
#!/usr/bin/env python3
|
||||
import bz2
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import requests
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
|
||||
from cereal import log
|
||||
import cereal.messaging as messaging
|
||||
from common.api import Api
|
||||
from common.params import Params
|
||||
from common.realtime import set_core_affinity
|
||||
from system.hardware import TICI
|
||||
from system.loggerd.xattr_cache import getxattr, setxattr
|
||||
from system.loggerd.config import ROOT
|
||||
from system.swaglog import cloudlog
|
||||
|
||||
NetworkType = log.DeviceState.NetworkType
|
||||
UPLOAD_ATTR_NAME = 'user.upload'
|
||||
UPLOAD_ATTR_VALUE = b'1'
|
||||
|
||||
UPLOAD_QLOG_QCAM_MAX_SIZE = 100 * 1e6 # MB
|
||||
|
||||
allow_sleep = bool(os.getenv("UPLOADER_SLEEP", "1"))
|
||||
force_wifi = os.getenv("FORCEWIFI") is not None
|
||||
fake_upload = os.getenv("FAKEUPLOAD") is not None
|
||||
|
||||
|
||||
def get_directory_sort(d):
|
||||
return list(map(lambda s: s.rjust(10, '0'), d.rsplit('--', 1)))
|
||||
|
||||
def listdir_by_creation(d):
|
||||
try:
|
||||
paths = os.listdir(d)
|
||||
paths = sorted(paths, key=get_directory_sort)
|
||||
return paths
|
||||
except OSError:
|
||||
cloudlog.exception("listdir_by_creation failed")
|
||||
return list()
|
||||
|
||||
def clear_locks(root):
|
||||
for logname in os.listdir(root):
|
||||
path = os.path.join(root, logname)
|
||||
try:
|
||||
for fname in os.listdir(path):
|
||||
if fname.endswith(".lock"):
|
||||
os.unlink(os.path.join(path, fname))
|
||||
except OSError:
|
||||
cloudlog.exception("clear_locks failed")
|
||||
|
||||
|
||||
class Uploader():
|
||||
def __init__(self, dongle_id, root):
|
||||
self.dongle_id = dongle_id
|
||||
self.api = Api(dongle_id)
|
||||
self.root = root
|
||||
|
||||
self.upload_thread = None
|
||||
|
||||
self.last_resp = None
|
||||
self.last_exc = None
|
||||
|
||||
self.immediate_size = 0
|
||||
self.immediate_count = 0
|
||||
|
||||
# stats for last successfully uploaded file
|
||||
self.last_time = 0.0
|
||||
self.last_speed = 0.0
|
||||
self.last_filename = ""
|
||||
|
||||
self.immediate_folders = ["crash/", "boot/"]
|
||||
self.immediate_priority = {"qlog": 0, "qlog.bz2": 0, "qcamera.ts": 1}
|
||||
|
||||
def get_upload_sort(self, name):
|
||||
if name in self.immediate_priority:
|
||||
return self.immediate_priority[name]
|
||||
return 1000
|
||||
|
||||
def list_upload_files(self):
|
||||
if not os.path.isdir(self.root):
|
||||
return
|
||||
|
||||
self.immediate_size = 0
|
||||
self.immediate_count = 0
|
||||
|
||||
for logname in listdir_by_creation(self.root):
|
||||
path = os.path.join(self.root, logname)
|
||||
try:
|
||||
names = os.listdir(path)
|
||||
except OSError:
|
||||
continue
|
||||
|
||||
if any(name.endswith(".lock") for name in names):
|
||||
continue
|
||||
|
||||
for name in sorted(names, key=self.get_upload_sort):
|
||||
key = os.path.join(logname, name)
|
||||
fn = os.path.join(path, name)
|
||||
# skip files already uploaded
|
||||
try:
|
||||
is_uploaded = getxattr(fn, UPLOAD_ATTR_NAME)
|
||||
except OSError:
|
||||
cloudlog.event("uploader_getxattr_failed", exc=self.last_exc, key=key, fn=fn)
|
||||
is_uploaded = True # deleter could have deleted
|
||||
if is_uploaded:
|
||||
continue
|
||||
|
||||
try:
|
||||
if name in self.immediate_priority:
|
||||
self.immediate_count += 1
|
||||
self.immediate_size += os.path.getsize(fn)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
yield (name, key, fn)
|
||||
|
||||
def next_file_to_upload(self):
|
||||
upload_files = list(self.list_upload_files())
|
||||
|
||||
for name, key, fn in upload_files:
|
||||
if any(f in fn for f in self.immediate_folders):
|
||||
return (name, key, fn)
|
||||
|
||||
for name, key, fn in upload_files:
|
||||
if name in self.immediate_priority:
|
||||
return (name, key, fn)
|
||||
|
||||
return None
|
||||
|
||||
def do_upload(self, key, fn):
|
||||
try:
|
||||
url_resp = self.api.get("v1.4/" + self.dongle_id + "/upload_url/", timeout=10, path=key, access_token=self.api.get_token())
|
||||
if url_resp.status_code == 412:
|
||||
self.last_resp = url_resp
|
||||
return
|
||||
|
||||
url_resp_json = json.loads(url_resp.text)
|
||||
url = url_resp_json['url']
|
||||
headers = url_resp_json['headers']
|
||||
cloudlog.debug("upload_url v1.4 %s %s", url, str(headers))
|
||||
|
||||
if fake_upload:
|
||||
cloudlog.debug(f"*** WARNING, THIS IS A FAKE UPLOAD TO {url} ***")
|
||||
|
||||
class FakeResponse():
|
||||
def __init__(self):
|
||||
self.status_code = 200
|
||||
|
||||
self.last_resp = FakeResponse()
|
||||
else:
|
||||
with open(fn, "rb") as f:
|
||||
if key.endswith('.bz2') and not fn.endswith('.bz2'):
|
||||
data = bz2.compress(f.read())
|
||||
data = io.BytesIO(data)
|
||||
else:
|
||||
data = f
|
||||
|
||||
self.last_resp = requests.put(url, data=data, headers=headers, timeout=10)
|
||||
except Exception as e:
|
||||
self.last_exc = (e, traceback.format_exc())
|
||||
raise
|
||||
|
||||
def normal_upload(self, key, fn):
|
||||
self.last_resp = None
|
||||
self.last_exc = None
|
||||
|
||||
try:
|
||||
self.do_upload(key, fn)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return self.last_resp
|
||||
|
||||
def upload(self, name, key, fn, network_type, metered):
|
||||
try:
|
||||
sz = os.path.getsize(fn)
|
||||
except OSError:
|
||||
cloudlog.exception("upload: getsize failed")
|
||||
return False
|
||||
|
||||
cloudlog.event("upload_start", key=key, fn=fn, sz=sz, network_type=network_type, metered=metered)
|
||||
|
||||
if sz == 0:
|
||||
# tag files of 0 size as uploaded
|
||||
success = True
|
||||
elif name in self.immediate_priority and sz > UPLOAD_QLOG_QCAM_MAX_SIZE:
|
||||
cloudlog.event("uploader_too_large", key=key, fn=fn, sz=sz)
|
||||
success = True
|
||||
else:
|
||||
start_time = time.monotonic()
|
||||
stat = self.normal_upload(key, fn)
|
||||
if stat is not None and stat.status_code in (200, 201, 401, 403, 412):
|
||||
self.last_filename = fn
|
||||
self.last_time = time.monotonic() - start_time
|
||||
self.last_speed = (sz / 1e6) / self.last_time
|
||||
success = True
|
||||
cloudlog.event("upload_success" if stat.status_code != 412 else "upload_ignored", key=key, fn=fn, sz=sz, network_type=network_type, metered=metered)
|
||||
else:
|
||||
success = False
|
||||
cloudlog.event("upload_failed", stat=stat, exc=self.last_exc, key=key, fn=fn, sz=sz, network_type=network_type, metered=metered)
|
||||
|
||||
if success:
|
||||
# tag file as uploaded
|
||||
try:
|
||||
setxattr(fn, UPLOAD_ATTR_NAME, UPLOAD_ATTR_VALUE)
|
||||
except OSError:
|
||||
cloudlog.event("uploader_setxattr_failed", exc=self.last_exc, key=key, fn=fn, sz=sz)
|
||||
|
||||
return success
|
||||
|
||||
def get_msg(self):
|
||||
msg = messaging.new_message("uploaderState")
|
||||
us = msg.uploaderState
|
||||
us.immediateQueueSize = int(self.immediate_size / 1e6)
|
||||
us.immediateQueueCount = self.immediate_count
|
||||
us.lastTime = self.last_time
|
||||
us.lastSpeed = self.last_speed
|
||||
us.lastFilename = self.last_filename
|
||||
return msg
|
||||
|
||||
|
||||
def uploader_fn(exit_event):
|
||||
try:
|
||||
set_core_affinity([0, 1, 2, 3])
|
||||
except Exception:
|
||||
cloudlog.exception("failed to set core affinity")
|
||||
|
||||
clear_locks(ROOT)
|
||||
|
||||
params = Params()
|
||||
dongle_id = params.get("DongleId", encoding='utf8')
|
||||
|
||||
if dongle_id is None:
|
||||
cloudlog.info("uploader missing dongle_id")
|
||||
raise Exception("uploader can't start without dongle id")
|
||||
|
||||
if TICI and not Path("/data/media").is_mount():
|
||||
cloudlog.warning("NVME not mounted")
|
||||
|
||||
sm = messaging.SubMaster(['deviceState'])
|
||||
pm = messaging.PubMaster(['uploaderState'])
|
||||
uploader = Uploader(dongle_id, ROOT)
|
||||
|
||||
backoff = 0.1
|
||||
while not exit_event.is_set():
|
||||
sm.update(0)
|
||||
offroad = params.get_bool("IsOffroad")
|
||||
network_type = sm['deviceState'].networkType if not force_wifi else NetworkType.wifi
|
||||
if network_type == NetworkType.none:
|
||||
if allow_sleep:
|
||||
time.sleep(60 if offroad else 5)
|
||||
continue
|
||||
|
||||
d = uploader.next_file_to_upload()
|
||||
if d is None: # Nothing to upload
|
||||
if allow_sleep:
|
||||
time.sleep(60 if offroad else 5)
|
||||
continue
|
||||
|
||||
name, key, fn = d
|
||||
|
||||
# qlogs and bootlogs need to be compressed before uploading
|
||||
if key.endswith(('qlog', 'rlog')) or (key.startswith('boot/') and not key.endswith('.bz2')):
|
||||
key += ".bz2"
|
||||
|
||||
success = uploader.upload(name, key, fn, sm['deviceState'].networkType.raw, sm['deviceState'].networkMetered)
|
||||
if success:
|
||||
backoff = 0.1
|
||||
elif allow_sleep:
|
||||
cloudlog.info("upload backoff %r", backoff)
|
||||
time.sleep(backoff + random.uniform(0, backoff))
|
||||
backoff = min(backoff*2, 120)
|
||||
|
||||
pm.send("uploaderState", uploader.get_msg())
|
||||
|
||||
|
||||
def main():
|
||||
uploader_fn(threading.Event())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,114 @@
|
||||
#pragma clang diagnostic ignored "-Wdeprecated-declarations"
|
||||
#include <cassert>
|
||||
#include <cstdlib>
|
||||
|
||||
#include "system/loggerd/video_writer.h"
|
||||
#include "common/swaglog.h"
|
||||
#include "common/util.h"
|
||||
|
||||
VideoWriter::VideoWriter(const char *path, const char *filename, bool remuxing, int width, int height, int fps, cereal::EncodeIndex::Type codec)
|
||||
: remuxing(remuxing) {
|
||||
raw = codec == cereal::EncodeIndex::Type::BIG_BOX_LOSSLESS;
|
||||
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) {
|
||||
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);
|
||||
|
||||
int err = avio_open(&this->ofmt_ctx->pb, this->vid_path.c_str(), AVIO_FLAG_WRITE);
|
||||
assert(err >= 0);
|
||||
|
||||
} else {
|
||||
this->of = util::safe_fopen(this->vid_path.c_str(), "wb");
|
||||
assert(this->of);
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
if (remuxing) {
|
||||
if (codecconfig) {
|
||||
if (len > 0) {
|
||||
codec_ctx->extradata = (uint8_t*)av_mallocz(len + AV_INPUT_BUFFER_PADDING_SIZE);
|
||||
codec_ctx->extradata_size = len;
|
||||
memcpy(codec_ctx->extradata, data, len);
|
||||
}
|
||||
int err = avcodec_parameters_from_context(out_stream->codecpar, codec_ctx);
|
||||
assert(err >= 0);
|
||||
err = avformat_write_header(ofmt_ctx, NULL);
|
||||
assert(err >= 0);
|
||||
} else {
|
||||
// input timestamps are in microseconds
|
||||
AVRational in_timebase = {1, 1000000};
|
||||
|
||||
AVPacket pkt;
|
||||
av_init_packet(&pkt);
|
||||
pkt.data = data;
|
||||
pkt.size = len;
|
||||
|
||||
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, ofmt_ctx->streams[0]->time_base, rnd);
|
||||
pkt.duration = av_rescale_q(50*1000, in_timebase, ofmt_ctx->streams[0]->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: %lu", len, timestamp); }
|
||||
|
||||
av_packet_unref(&pkt);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
VideoWriter::~VideoWriter() {
|
||||
if (this->remuxing) {
|
||||
if (this->raw) { avcodec_close(this->codec_ctx); }
|
||||
int err = av_write_trailer(this->ofmt_ctx);
|
||||
if (err != 0) LOGE("av_write_trailer failed %d", err);
|
||||
avcodec_free_context(&this->codec_ctx);
|
||||
err = avio_closep(&this->ofmt_ctx->pb);
|
||||
if (err != 0) LOGE("avio_closep failed %d", err);
|
||||
avformat_free_context(this->ofmt_ctx);
|
||||
} else {
|
||||
util::safe_fflush(this->of);
|
||||
fclose(this->of);
|
||||
this->of = nullptr;
|
||||
}
|
||||
unlink(this->lock_path.c_str());
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
#pragma once
|
||||
|
||||
#include <string>
|
||||
|
||||
extern "C" {
|
||||
#include <libavformat/avformat.h>
|
||||
#include <libavcodec/avcodec.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);
|
||||
~VideoWriter();
|
||||
private:
|
||||
std::string vid_path, lock_path;
|
||||
|
||||
FILE *of = nullptr;
|
||||
|
||||
AVCodecContext *codec_ctx;
|
||||
AVFormatContext *ofmt_ctx;
|
||||
AVStream *out_stream;
|
||||
bool remuxing, raw;
|
||||
};
|
||||
@@ -0,0 +1,23 @@
|
||||
import os
|
||||
import errno
|
||||
from typing import Dict, Tuple, Optional
|
||||
|
||||
_cached_attributes: Dict[Tuple, Optional[bytes]] = {}
|
||||
|
||||
def getxattr(path: str, attr_name: str) -> Optional[bytes]:
|
||||
key = (path, attr_name)
|
||||
if key not in _cached_attributes:
|
||||
try:
|
||||
response = os.getxattr(path, attr_name)
|
||||
except OSError as e:
|
||||
# ENODATA means attribute hasn't been set
|
||||
if e.errno == errno.ENODATA:
|
||||
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)
|
||||
return os.setxattr(path, attr_name, attr_value)
|
||||
Reference in New Issue
Block a user