refactor FrameReader (#21141)

* refactor FrameReader

* continue

* move that

* small cleanup

* little more

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>
pull/21184/head
Dean Lee 4 years ago committed by GitHub
parent 5dbf1c1e28
commit 67fe3feb09
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 200
      selfdrive/ui/replay/framereader.cc
  2. 48
      selfdrive/ui/replay/framereader.h
  3. 19
      selfdrive/ui/replay/replay.cc

@ -3,6 +3,9 @@
#include <assert.h> #include <assert.h>
#include <unistd.h> #include <unistd.h>
#include <QDebug>
static int ffmpeg_lockmgr_cb(void **arg, enum AVLockOp op) { static int ffmpeg_lockmgr_cb(void **arg, enum AVLockOp op) {
std::mutex *mutex = (std::mutex *)*arg; std::mutex *mutex = (std::mutex *)*arg;
switch (op) { switch (op) {
@ -21,122 +24,173 @@ static int ffmpeg_lockmgr_cb(void **arg, enum AVLockOp op) {
return 0; return 0;
} }
FrameReader::FrameReader(const std::string &fn) : url(fn) { class AVInitializer {
public:
AVInitializer() {
int ret = av_lockmgr_register(ffmpeg_lockmgr_cb); int ret = av_lockmgr_register(ffmpeg_lockmgr_cb);
assert(ret >= 0); assert(ret >= 0);
avformat_network_init();
av_register_all(); av_register_all();
avformat_network_init();
}
~AVInitializer() { avformat_network_deinit(); }
};
static AVInitializer av_initializer;
FrameReader::FrameReader(const std::string &url, QObject *parent) : url_(url), QObject(parent) {
process_thread_ = QThread::create(&FrameReader::process, this);
connect(process_thread_, &QThread::finished, process_thread_, &QThread::deleteLater);
process_thread_->start();
} }
FrameReader::~FrameReader() { FrameReader::~FrameReader() {
// wait until thread is finished.
exit_ = true; exit_ = true;
thread.join(); process_thread_->wait();
for (auto &f : frames) { cv_decode_.notify_all();
delete f->pkt; cv_frame_.notify_all();
if (f->picture) { if (decode_thread_.joinable()) {
av_frame_free(&f->picture); decode_thread_.join();
}
// free all.
for (auto &f : frames_) {
av_free_packet(&f.pkt);
if (f.data) {
delete[] f.data;
}
} }
delete f; while (!buffer_pool.empty()) {
delete[] buffer_pool.front();
buffer_pool.pop();
} }
avcodec_free_context(&pCodecCtx); av_frame_free(&frmRgb_);
avformat_free_context(pFormatCtx); avcodec_close(pCodecCtx_);
sws_freeContext(sws_ctx); avcodec_free_context(&pCodecCtx_);
avformat_network_deinit(); avformat_close_input(&pFormatCtx_);
sws_freeContext(sws_ctx_);
} }
void FrameReader::process() { void FrameReader::process() {
if (avformat_open_input(&pFormatCtx, url.c_str(), NULL, NULL) != 0) { if (processFrames()) {
fprintf(stderr, "error loading %s\n", url.c_str()); decode_thread_ = std::thread(&FrameReader::decodeThread, this);
valid = false; }
return; if (!exit_) {
emit finished();
}
}
bool FrameReader::processFrames() {
if (avformat_open_input(&pFormatCtx_, url_.c_str(), NULL, NULL) != 0) {
qDebug() << "error loading " << url_.c_str();
return false;
} }
avformat_find_stream_info(pFormatCtx, NULL); avformat_find_stream_info(pFormatCtx_, NULL);
av_dump_format(pFormatCtx, 0, url.c_str(), 0); av_dump_format(pFormatCtx_, 0, url_.c_str(), 0);
auto pCodecCtxOrig = pFormatCtx->streams[0]->codec; auto pCodecCtxOrig = pFormatCtx_->streams[0]->codec;
auto pCodec = avcodec_find_decoder(pCodecCtxOrig->codec_id); auto pCodec = avcodec_find_decoder(pCodecCtxOrig->codec_id);
assert(pCodec != NULL); assert(pCodec);
pCodecCtx = avcodec_alloc_context3(pCodec); pCodecCtx_ = avcodec_alloc_context3(pCodec);
int ret = avcodec_copy_context(pCodecCtx, pCodecCtxOrig); int ret = avcodec_copy_context(pCodecCtx_, pCodecCtxOrig);
assert(ret == 0); assert(ret == 0);
ret = avcodec_open2(pCodecCtx, pCodec, NULL); ret = avcodec_open2(pCodecCtx_, pCodec, NULL);
assert(ret >= 0); assert(ret >= 0);
width = pCodecCtxOrig->width; width = pCodecCtxOrig->width;
height = pCodecCtxOrig->height; height = pCodecCtxOrig->height;
sws_ctx = sws_getContext(width, height, AV_PIX_FMT_YUV420P, sws_ctx_ = sws_getContext(width, height, AV_PIX_FMT_YUV420P,
width, height, AV_PIX_FMT_BGR24, width, height, AV_PIX_FMT_BGR24,
SWS_BILINEAR, NULL, NULL, NULL); SWS_BILINEAR, NULL, NULL, NULL);
assert(sws_ctx != NULL); assert(sws_ctx_);
frmRgb_ = av_frame_alloc();
assert(frmRgb_);
frames_.reserve(60 * 20); // 20fps, one minute
do { do {
AVPacket *pkt = new AVPacket; Frame &frame = frames_.emplace_back();
if (av_read_frame(pFormatCtx, pkt) < 0) { if (av_read_frame(pFormatCtx_, &frame.pkt) < 0) {
delete pkt; frames_.pop_back();
break; break;
} }
Frame *frame = new Frame; } while (!exit_);
frame->pkt = pkt;
frames.push_back(frame);
} while (true);
printf("framereader download done\n"); valid_ = !exit_;
return valid_;
}
thread = std::thread(&FrameReader::decodeThread, this); uint8_t *FrameReader::get(int idx) {
if (!valid_ || idx < 0 || idx >= frames_.size()) {
return nullptr;
} }
void FrameReader::decodeThread() {
while (!exit_) {
int gop = 0;
{ {
std::unique_lock lk(mutex); std::unique_lock lk(mutex_);
cv_decode.wait(lk, [=] { return exit_ || decode_idx != -1; }); decode_idx_ = idx;
if (exit_) break; cv_decode_.notify_one();
cv_frame_.wait(lk, [=] { return exit_ || frames_[idx].data || frames_[idx].failed; });
gop = std::max(decode_idx - decode_idx % 15, 0);
decode_idx = -1;
} }
for (int i = gop; i < std::min(gop + 15, (int)frames.size()); ++i) { return frames_[idx].data;
if (frames[i]->picture != nullptr) continue; }
int frameFinished; void FrameReader::decodeThread() {
AVFrame *pFrame = av_frame_alloc(); int idx = 0;
avcodec_decode_video2(pCodecCtx, pFrame, &frameFinished, frames[i]->pkt); while (!exit_) {
AVFrame *picture = toRGB(pFrame); const int from = std::max(idx, 0);
av_frame_free(&pFrame); const int to = std::min(idx + 20, (int)frames_.size());
for (int i = 0; i < frames_.size() && !exit_; ++i) {
Frame &frame = frames_[i];
if (i >= from && i < to) {
if (frame.data || frame.failed) continue;
std::unique_lock lk(mutex); uint8_t *dat = decodeFrame(&frame.pkt);
frames[i]->picture = picture; std::unique_lock lk(mutex_);
cv_frame.notify_all(); frame.data = dat;
} frame.failed = !dat;
cv_frame_.notify_all();
} else if (frame.data) {
buffer_pool.push(frame.data);
frame.data = nullptr;
frame.failed = false;
} }
} }
AVFrame *FrameReader::toRGB(AVFrame *pFrame) { // sleep & wait
AVFrame *pFrameRGB = av_frame_alloc(); std::unique_lock lk(mutex_);
int numBytes = avpicture_get_size(AV_PIX_FMT_BGR24, pFrame->width, pFrame->height); cv_decode_.wait(lk, [=] { return exit_ || decode_idx_ != -1; });
uint8_t *buffer = (uint8_t *)av_malloc(numBytes * sizeof(uint8_t)); idx = decode_idx_;
avpicture_fill((AVPicture *)pFrameRGB, buffer, AV_PIX_FMT_BGR24, pFrame->width, pFrame->height); decode_idx_ = -1;
sws_scale(sws_ctx, (uint8_t const *const *)pFrame->data, }
pFrame->linesize, 0, pFrame->height,
pFrameRGB->data, pFrameRGB->linesize);
return pFrameRGB;
} }
uint8_t *FrameReader::get(int idx) { uint8_t *FrameReader::decodeFrame(AVPacket *pkt) {
if (!valid || idx < 0 || idx >= frames.size()) return nullptr; int gotFrame;
AVFrame *f = av_frame_alloc();
avcodec_decode_video2(pCodecCtx_, f, &gotFrame, pkt);
std::unique_lock lk(mutex); uint8_t *dat = nullptr;
decode_idx = idx; if (gotFrame) {
cv_decode.notify_one(); if (!buffer_pool.empty()) {
Frame *frame = frames[idx]; dat = buffer_pool.front();
if (!frame->picture) { buffer_pool.pop();
cv_frame.wait(lk, [=] { return exit_ || frame->picture != nullptr; }); } else {
dat = new uint8_t[getRGBSize()];
}
int ret = avpicture_fill((AVPicture *)frmRgb_, dat, AV_PIX_FMT_BGR24, f->width, f->height);
assert(ret > 0);
if (sws_scale(sws_ctx_, (const uint8_t **)f->data, f->linesize, 0,
f->height, frmRgb_->data, frmRgb_->linesize) <= 0) {
delete[] dat;
dat = nullptr;
}
} }
return frame->picture ? frame->picture->data[0] : nullptr; av_frame_free(&f);
return dat;
} }

@ -5,10 +5,13 @@
#include <atomic> #include <atomic>
#include <condition_variable> #include <condition_variable>
#include <mutex> #include <mutex>
#include <queue>
#include <string> #include <string>
#include <thread> #include <thread>
#include <vector> #include <vector>
#include <QThread>
// independent of QT, needs ffmpeg // independent of QT, needs ffmpeg
extern "C" { extern "C" {
#include <libavcodec/avcodec.h> #include <libavcodec/avcodec.h>
@ -16,38 +19,47 @@ extern "C" {
#include <libswscale/swscale.h> #include <libswscale/swscale.h>
} }
class FrameReader : public QObject {
Q_OBJECT
class FrameReader {
public: public:
FrameReader(const std::string &fn); FrameReader(const std::string &url, QObject *parent = nullptr);
~FrameReader(); ~FrameReader();
uint8_t *get(int idx); uint8_t *get(int idx);
AVFrame *toRGB(AVFrame *);
int getRGBSize() { return width * height * 3; } int getRGBSize() { return width * height * 3; }
void process(); bool valid() const { return valid_; }
int width = 0, height = 0; int width = 0, height = 0;
signals:
void finished();
private: private:
void process();
bool processFrames();
void decodeThread(); void decodeThread();
uint8_t *decodeFrame(AVPacket *pkt);
struct Frame { struct Frame {
AVPacket *pkt; AVPacket pkt = {};
AVFrame *picture; uint8_t *data = nullptr;
bool failed = false;
}; };
std::vector<Frame*> frames; std::vector<Frame> frames_;
AVFormatContext *pFormatCtx = NULL; AVFormatContext *pFormatCtx_ = NULL;
AVCodecContext *pCodecCtx = NULL; AVCodecContext *pCodecCtx_ = NULL;
struct SwsContext *sws_ctx = NULL; AVFrame *frmRgb_ = nullptr;
std::queue<uint8_t *> buffer_pool;
struct SwsContext *sws_ctx_ = NULL;
std::mutex mutex; std::mutex mutex_;
std::condition_variable cv_decode; std::condition_variable cv_decode_;
std::condition_variable cv_frame; std::condition_variable cv_frame_;
int decode_idx = -1; int decode_idx_ = 0;
std::atomic<bool> exit_ = false; std::atomic<bool> exit_ = false;
std::thread thread; bool valid_ = false;
std::string url_;
bool valid = true; QThread *process_thread_;
std::string url; std::thread decode_thread_;
}; };

@ -76,19 +76,11 @@ void Replay::addSegment(int n) {
QObject::connect(t, &QThread::started, lrs[n], &LogReader::process); QObject::connect(t, &QThread::started, lrs[n], &LogReader::process);
t->start(); t->start();
QThread *frame_thread = QThread::create([=]{ frs[n] = new FrameReader(qPrintable(camera_paths.at(n).toString()), this);
FrameReader *frame_reader = new FrameReader(qPrintable(camera_paths.at(n).toString()));
frame_reader->process();
frs.insert(n, frame_reader);
});
QObject::connect(frame_thread, &QThread::finished, frame_thread, &QThread::deleteLater);
frame_thread->start();
} }
void Replay::removeSegment(int n) { void Replay::removeSegment(int n) {
// TODO: fix FrameReader and LogReader destructors // TODO: fix LogReader destructors
/* /*
if (lrs.contains(n)) { if (lrs.contains(n)) {
auto lr = lrs.take(n); auto lr = lrs.take(n);
@ -255,8 +247,6 @@ void Replay::stream() {
auto pp = *it_; auto pp = *it_;
if (frs.find(pp.first) != frs.end()) { if (frs.find(pp.first) != frs.end()) {
auto frm = frs[pp.first]; auto frm = frs[pp.first];
auto data = frm->get(pp.second);
if (vipc_server == nullptr) { if (vipc_server == nullptr) {
cl_device_id device_id = cl_get_device_id(CL_DEVICE_TYPE_DEFAULT); cl_device_id device_id = cl_get_device_id(CL_DEVICE_TYPE_DEFAULT);
cl_context context = CL_CHECK_ERR(clCreateContext(NULL, 1, &device_id, NULL, NULL, &err)); cl_context context = CL_CHECK_ERR(clCreateContext(NULL, 1, &device_id, NULL, NULL, &err));
@ -267,13 +257,16 @@ void Replay::stream() {
vipc_server->start_listener(); vipc_server->start_listener();
} }
uint8_t *dat = frm->get(pp.second);
if (dat) {
VisionIpcBufExtra extra = {}; VisionIpcBufExtra extra = {};
VisionBuf *buf = vipc_server->get_buffer(VisionStreamType::VISION_STREAM_RGB_BACK); VisionBuf *buf = vipc_server->get_buffer(VisionStreamType::VISION_STREAM_RGB_BACK);
memcpy(buf->addr, data, frm->getRGBSize()); memcpy(buf->addr, dat, frm->getRGBSize());
vipc_server->send(buf, &extra, false); vipc_server->send(buf, &extra, false);
} }
} }
} }
}
// publish msg // publish msg
if (sm == nullptr) { if (sm == nullptr) {

Loading…
Cancel
Save