改用FFRecoder2

Hu Chunming
1 parent 77e1339c
Showing 5 changed files with 112 additions and 45 deletions
src/decoder/dvpp/DvppDecoder.cpp
src/decoder/dvpp/FFRecoder2.cpp
src/decoder/dvpp/FFRecoder2.h
src/decoder/dvpp/FFRecoderTaskManager.cpp
src/decoder/dvpp/FFRecoderTaskManager.h
@@ -197,7 +197,7 @@ AVCodecContext* DvppDecoder::init_FFmpeg(FFDecConfig config){
         m_vdec_out_size = frame_width * frame_height * 3 / 2;
  
 #ifdef USE_VILLAGE
-        bool bRet = m_recoderManager.init(stream, avctx);
+        bool bRet = m_recoderManager.init2(frame_width, frame_height, m_fps, avctx->bit_rate);
         if (!bRet){
             LOG_ERROR("[{}]- m_recoderManager 初始化失败!", m_dec_name);
         }  
@@ -72,7 +72,8 @@ bool FFRecoder2::init(int w, int h, int fps, int bit_rate, const char* outfile_n
 	av_opt_set(codec_ctx_->priv_data, "tune", "zerolatency", 0);
  
 	// 打开解码器
-	if (avcodec_open2(codec_ctx_, encoder, nullptr) < 0) {
+	int ret = avcodec_open2(codec_ctx_, encoder, nullptr);
+	if (ret < 0) {
 		fprintf(stderr, "Open encoder failed!\n");
 		return false;
 	}
@@ -206,7 +207,14 @@ bool FFRecoder2::write_frame(const AVFrame* frame)
 	return true;
 }
  
+static double a2d(AVRational a) {
+	return a.den  / a.num;
+}
+
 bool FFRecoder2::write_pkt(AVPacket* pkt) {
+	frame_nb++;
+	pkt->duration = int(a2d(codec_ctx_->time_base));
+	pkt->pts = frame_nb;
 	// 将pts缩放到输出流的time_base上
 	av_packet_rescale_ts(pkt, codec_ctx_->time_base, out_stream_->time_base);
 	pkt->stream_index = out_stream_->index;
@@ -35,4 +35,6 @@ private:
 	AVFormatContext* fmt_ctx_;
 	AVStream* out_stream_;
 	AVFrame* yuv_frame_;
+
+	int frame_nb{0};
 };
 \ No newline at end of file
@@ -31,9 +31,16 @@ static bool is_key_frame(AVPacket *pkt) {
  
 FFRecoderTaskManager::FFRecoderTaskManager(){
     m_recoder_thread = nullptr;
+    m_cache_thread = nullptr;
+
+    m_bExit = false;
+    m_bExitRecoderThread = false;
 }
  
 FFRecoderTaskManager::~FFRecoderTaskManager(){
+
+    close();
+
     LOG_DEBUG("~FFRecoderTaskManager()");
 }
  
@@ -71,11 +78,11 @@ bool FFRecoderTaskManager::init2(int w, int h, int fps, int bit_rate) {
         m_fps = 25;
     }
  
-    m_recoder_thread = new std::thread(
+    m_cache_thread = new std::thread(
         [](void* arg) {
             FFRecoderTaskManager* _this=(FFRecoderTaskManager*)arg;
             if(_this != nullptr) {
-                _this->recode_thread2();
+                _this->pkt_cache_thread();
             }else{
                 LOG_ERROR("recode 线程启动失败 !");
             }
@@ -89,16 +96,15 @@ static AVPacket* packet_clone(AVPacket* pkt) {
     AVPacket *new_pkt = av_packet_alloc();
 	av_init_packet( new_pkt );
     av_new_packet(new_pkt, pkt->size);
-    // new_pkt->data = (uint8_t *)av_malloc(pkt->size) ;
     memcpy(new_pkt->data, pkt->data, pkt->size);
     new_pkt->size = pkt->size;
-    new_pkt->pts = pkt->pts;
-    new_pkt->dts = pkt->dts;
-    new_pkt->stream_index = pkt->stream_index;
-    new_pkt->duration = pkt->duration;
-    new_pkt->pos = pkt->pos;
-    new_pkt->flags = pkt->flags;
-    av_copy_packet_side_data(new_pkt, pkt);
+    // new_pkt->pts = pkt->pts;
+    // new_pkt->dts = pkt->dts;
+    // new_pkt->stream_index = pkt->stream_index;
+    // new_pkt->duration = pkt->duration;
+    // new_pkt->pos = pkt->pos;
+    // new_pkt->flags = pkt->flags;
+    // av_copy_packet_side_data(new_pkt, pkt);
     return new_pkt;
 }
  
@@ -124,19 +130,9 @@ void FFRecoderTaskManager::cache_pkt(AVPacket* pkt, long long frame_nb){
     } else {
         newDataPkt->isKeyFrame = false;
     }
-
-    m_pkt_list.emplace_back(newDataPkt);
-
-    std::lock_guard<std::mutex> l_info(m_recoderinfo_list_mtx);
-    if(m_recoderinfo_list.size() <= 0){
-        // 没有任务的时候，维持500的长度
-        while(m_pkt_list.size() > 1000) {
-            DataPacket* dataPkt = m_pkt_list.front();
-            delete dataPkt;
-            dataPkt = nullptr;
-            m_pkt_list.pop_front();
-        }
-    }
+    
+    std::lock_guard<std::mutex> l_info(m_pkt_list_short_mtx);
+    m_pkt_list_short.push_back(newDataPkt);
 }
  
 void FFRecoderTaskManager::save_intask_recoderinfo(RecoderInfo info) {
@@ -284,11 +280,76 @@ void FFRecoderTaskManager::recode_thread() {
     LOG_INFO("recode_thread end.");
 }
  
+void FFRecoderTaskManager::pkt_cache_thread() {
+    LOG_INFO("pkt_cache_thread start...");
+
+     m_recoder_thread = new std::thread(
+        [](void* arg) {
+            FFRecoderTaskManager* _this=(FFRecoderTaskManager*)arg;
+            if(_this != nullptr) {
+                _this->recode_thread2();
+            }else{
+                LOG_ERROR("recode 线程启动失败 !");
+            }
+            return (void*)0;
+        }, this);
+
+    // 开始缓存
+    while(true) {
+        if(m_bExit) {
+            break;
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+        m_pkt_list_short_mtx.lock();
+        m_pkt_list_mtx.lock();
+        for (size_t i = 0; i < m_pkt_list_short.size(); i++) {
+            auto item = m_pkt_list_short.front();
+            m_pkt_list.push_back(item);
+            m_pkt_list_short.pop_front();
+        }
+        m_pkt_list_mtx.unlock();
+        m_pkt_list_short_mtx.unlock();
+        
+
+        std::lock_guard<std::mutex> l_info(m_recoderinfo_list_mtx);
+        if(m_recoderinfo_list.size() <= 0){
+            // 没有任务的时候，维持500的长度
+            m_pkt_list_mtx.lock();
+            while(m_pkt_list.size() > 1000) {
+                DataPacket* dataPkt = m_pkt_list.front();
+                delete dataPkt;
+                dataPkt = nullptr;
+                m_pkt_list.pop_front();
+            }
+            m_pkt_list_mtx.unlock();
+        }
+    }
+
+    m_bExitRecoderThread = true;
+
+    if (m_recoder_thread) {
+        m_recoder_thread->join();
+        delete m_recoder_thread;
+        m_recoder_thread = nullptr;
+    }
+
+    // 清空数据
+    while(!m_pkt_list.empty()) {
+        DataPacket* dataPkt = m_pkt_list.front();
+        delete dataPkt;
+        dataPkt = nullptr;
+        m_pkt_list.pop_front();
+    }
+
+    LOG_INFO("pkt_cache_thread end.");
+}
  
 void FFRecoderTaskManager::recode_thread2() {
     LOG_INFO("recode_thread2 start...");
     while(true) {
-        if(m_bExit) {
+        if(m_bExitRecoderThread) {
             break;
         }
  
@@ -313,16 +374,7 @@ void FFRecoderTaskManager::recode_thread2() {
  
             LOG_INFO("start frame_nb: {}", (*it_data)->frame_nb);
  
-            m_pkt_list_mtx.lock();
-            auto it = m_pkt_list.begin();
-            while (it != it_data) {
-                DataPacket* dataPkt = m_pkt_list.front();
-                delete dataPkt;
-                dataPkt = nullptr;
-                m_pkt_list.pop_front();
-                it = m_pkt_list.begin();
-            }
-            m_pkt_list_mtx.unlock();
+            std::lock_guard<std::mutex> l_long(m_pkt_list_mtx);
  
             string file_name = recoderinfo.recoderPath;
  
@@ -377,24 +429,24 @@ void FFRecoderTaskManager::recode_thread2() {
         m_recoderinfo_list_mtx.unlock();  
     }
  
-    LOG_INFO("recode_thread end.");
+    LOG_INFO("recode_thread2 end.");
 }
  
 void FFRecoderTaskManager::close() {
     m_bExit = true;
  
-    if (m_recoder_thread) {
-        m_recoder_thread->join();
-        delete m_recoder_thread;
-        m_recoder_thread = nullptr;
+    if (m_cache_thread) {
+        m_cache_thread->join();
+        delete m_cache_thread;
+        m_cache_thread = nullptr;
     }
  
     // 清空数据
-    while(!m_pkt_list.empty()) {
-        DataPacket* dataPkt = m_pkt_list.front();
+    while(!m_pkt_list_short.empty()) {
+        DataPacket* dataPkt = m_pkt_list_short.front();
         delete dataPkt;
         dataPkt = nullptr;
-        m_pkt_list.pop_front();
+        m_pkt_list_short.pop_front();
     }
 }
  
@@ -39,6 +39,7 @@ public:
 public:
     void recode_thread();
     void recode_thread2();
+    void pkt_cache_thread();
  
 private:
     void save_intask_recoderinfo(RecoderInfo info);
@@ -47,19 +48,23 @@ private:
 private:
     std::queue<int> m_key_frame_interval;
     std::list<unsigned long long> m_keyframe_nb_list;
-    std::list<DataPacket*> m_pkt_list;
+    std::list<DataPacket*> m_pkt_list_short;    // 临时缓存
+    mutex m_pkt_list_short_mtx; 
+    std::list<DataPacket*> m_pkt_list;      //主要缓存
     mutex m_pkt_list_mtx; 
  
     std::list<RecoderInfo> m_recoderinfo_list;
     mutex m_recoderinfo_list_mtx;
  
     bool m_bExit{false};
+    bool m_bExitRecoderThread{false};
  
     AVRational m_time_base;
     AVCodecContext* m_avctx;
     AVStream* m_inStream;
  
     thread* m_recoder_thread{nullptr};
+    thread* m_cache_thread{nullptr};
  
     mq_callback_t mq_publish_func;