修复解码器异常退出时，接收器还在正常跑的问题

Hu Chunming
1 parent 826e3002
Showing 4 changed files with 36 additions and 29 deletions
src/decoder/dvpp/DvppRtpDecoder.cpp
src/decoder/dvpp/DvppRtpDecoder.h
src/decoder/gb28181/rtp2/RTPReceiver2.cpp
src/decoder/gb28181/rtp2/RTPReceiver2.h
@@ -101,8 +101,6 @@ bool DvppRtpDecoder::Init(FFDecConfig cfg) {
 	m_bResize = m_cfg.resize;
-    decode_finished_cbk = cfg.decode_finished_cbk;
-
 	bool bRet = init_dvpp(cfg);
 	if(!bRet){
 		return false;
@@ -264,6 +262,12 @@ void DvppRtpDecoder::setFinishedDecArg(const void* finishedDecArg){
     m_finishedDecArg = finishedDecArg;
 }
+void DvppRtpDecoder::SetFinishedCallback(CallBack_DecodeFinished cb, void* param)
+{
+    m_finish_cbk = cb;
+	m_finishParam = param;
+}
+
 void DvppRtpDecoder::pause(){
     m_bPause = true;
 }
@@ -370,27 +374,27 @@ void DvppRtpDecoder::CacheBuffer(uint8_t* recvBuf, int recvBufSize) {
 }
 int DvppRtpDecoder::ReadBuffer(uint8_t* buf, int buffsize) {
-	int ret = 0;
+
     int count = 0;
-    while(m_bRunning && (m_bufferSize < 4096 || m_bufferSize < buffsize)){
+    while(m_bufferSize < buffsize){
+        if(!m_bRunning){
+            return AVERROR_EXIT;
+        }
         std::this_thread::sleep_for(std::chrono::milliseconds(10));
         count++;
-        if (count >= 1000) {
-            // 只等待10s
-            return -1;
+        if (count >= 3000) {
+            // 只等待30s
+            return AVERROR(EIO);
         }
     }
-	if (m_bufferSize >= buffsize) {
-		memcpy(buf, m_buffer, buffsize);
-		m_bufferSize = m_bufferSize - buffsize;
-		memmove(m_buffer, m_buffer + buffsize, m_bufferSize);
-		ret = buffsize;
-	}
+    memcpy(buf, m_buffer, buffsize);
+    m_bufferSize = m_bufferSize - buffsize;
+    memmove(m_buffer, m_buffer + buffsize, m_bufferSize);
-    printf("m_bufferSize=%d  buffsize=%d\n", m_bufferSize.load(), buffsize);
+    // printf("m_bufferSize=%d  buffsize=%d\n", m_bufferSize.load(), buffsize);
-	return ret;
+	return buffsize;
 }
 bool DvppRtpDecoder::probe() {
@@ -426,7 +430,7 @@ bool DvppRtpDecoder::probe() {
         av_dict_set( &net_options, "max_delay", "500000", 0); //设置最大时延
 		//打开流
-        ret = avformat_open_input(&fmt_ctx, 0, 0, &net_options);
+        ret = avformat_open_input(&fmt_ctx, 0, inputFmt, &net_options);
 		if (ret != 0) {
 			LOG_ERROR("avformat_open_input error: {}", ret);
 			break;
@@ -577,6 +581,8 @@ void DvppRtpDecoder::read_thread() {
             if (m_DvppCacheCounter.load() > m_cache_gop){
                 // 解码器解码不过来。实时流在此处的处理会导致花屏，这是由于解码器性能问题导致，无法避免
                 // 实时流在这里处理是为了避免长时间不读取数据导致数据中断
+                av_packet_free(&pkt);
+                pkt = nullptr;
                 std::this_thread::sleep_for(std::chrono::milliseconds(10));
                 continue;
             }
@@ -609,11 +615,16 @@ void DvppRtpDecoder::read_thread() {
                 if(nSended < 0) {
                     // 执行出错，强行结束整个任务
                     m_bRunning=false;
+                    av_packet_free(&pkt);
+                    pkt = nullptr;
                     break;
                 }
     #ifdef USE_VILLAGE
                 m_recoderManager.cache_pkt(pkt, frame_nb, m_dec_name);
+    #else 
+                av_packet_free(&pkt);
+                pkt = nullptr;
     #endif
             } else {
                 av_packet_free(&pkt);
@@ -646,8 +657,8 @@ void DvppRtpDecoder::read_thread() {
     LOG_INFO("[{}]- read thread exit.", m_dec_name);
-    if(decode_finished_cbk) {
-        decode_finished_cbk(m_finishedDecArg);
+    if(m_finish_cbk) {
+        m_finish_cbk(m_finishParam);
     }
 }
@@ -21,6 +21,8 @@ using namespace std;
 #define MAX_RTP_BUFFER_SIZE 4194304 // 4M = 4 * 1024 * 1024 = 4194304 字节
+typedef void(*CallBack_DecodeFinished)(void* userdata);
+
 class DvppRtpDecoder
 {
 public:
@@ -73,6 +75,8 @@ public:
 	int ReadBuffer(uint8_t* buf, int buffsize);
+    void SetFinishedCallback(CallBack_DecodeFinished cb, void* param);
+
 public:
     void doVdppVdecCallBack(acldvppStreamDesc *input, acldvppPicDesc *output, void *pUserData);
     void doProcessReport();
@@ -96,7 +100,6 @@ private:
     string m_dec_name;
     const void * m_finishedDecArg {nullptr};
-    DECODE_FINISHED_CALLBACK decode_finished_cbk {nullptr};
     bool m_bFinished{false};
     bool m_bRunning{false};
@@ -152,5 +155,8 @@ private:
     VpcUtils m_vpcUtils;
+    void*           m_finishParam;
+	CallBack_DecodeFinished m_finish_cbk;				// 录像流结束回调
+
 };
 #endif //__DVPP_RTP_DECODER_H__
 \ No newline at end of file
 #include "RTPReceiver2.h"
-#include "rtppacket.h"
 #include <thread>
 #include "../common_header.h"
@@ -68,19 +67,12 @@ bool RTPReceiver2::Open(string channel_id, bool isUdp) {
 		Close();
 		return false;
 	}
-    
-    m_bOpened = true;
     LOG_INFO("[{}] started.", m_SipChannelId);
     return true;
 }
-bool RTPReceiver2::IsOpened(){
-    LOG_INFO("[{}] isopen:{} ", m_SipChannelId, m_bOpened);
-    return m_bOpened;
-}
-
 void RTPReceiver2::Close(){
     m_bRtpExit = true;
@@ -19,7 +19,6 @@ public:
     virtual ~RTPReceiver2();
 	bool Open(string channel_id, bool isUdp);
-	bool IsOpened();
 	void Close();
 	void SetVodEndCallback(CallBack_VodFileEnd cb, void* param);
@@ -50,7 +49,6 @@ public:
 	string m_SipChannelId;
 	int m_rtp_port{-1};
-	std::atomic_bool    m_bOpened;
     std::atomic_bool    m_bAccepted;
     std::atomic_bool    m_bClosing;