等待时间优化

Hu Chunming
1 parent 5f4935b7
Showing 3 changed files with 29 additions and 18 deletions
src/decoder/dvpp/DvppRtpDecoder.cpp
src/decoder/dvpp/DvppRtpDecoder.h
src/decoder/gb28181/rtp2/RTPReceiver2.cpp
@@ -382,8 +382,8 @@ int DvppRtpDecoder::ReadBuffer(uint8_t* buf, int buffsize) {
         }
         std::this_thread::sleep_for(std::chrono::milliseconds(10));
         count++;
-        if (count >= 3000) {
-            // 只等待30s
+        if (count >= m_buffer_waiting_time) {
+            // 等待
             return AVERROR(EIO);
         }
     }
@@ -398,27 +398,22 @@ int DvppRtpDecoder::ReadBuffer(uint8_t* buf, int buffsize) {
 }
 bool DvppRtpDecoder::probe() {
+
+    m_buffer_waiting_time = 3000;   //probe只等待30s，避免卡主任务添加
 	// todo: 此处可能有泄露
 	unsigned char* avioBuff = (unsigned char*)av_malloc(MAX_RTP_BUFFER_SIZE);
 	AVIOContext  *ioCtx = avio_alloc_context(avioBuff, sizeof(avioBuff), 0, this, avio_read_packet, NULL, NULL);
-	//探测流（获取码流格式）
-	AVInputFormat* inputFmt = nullptr;
-    int ret = av_probe_input_buffer(ioCtx, &inputFmt, "", NULL, 0, 0);
-	if (ret < 0){
-		LOG_ERROR("av_probe_input_buffer error: {}", ret);
-		return false;
-	}
 	do{
 		fmt_ctx = avformat_alloc_context();
-        fmt_ctx->probesize = 10000000;//5 000 000
-        fmt_ctx->flags |= AVFMT_FLAG_NOBUFFER;
+        // fmt_ctx->probesize = 10000000;//5 000 000
+        // fmt_ctx->flags |= AVFMT_FLAG_NOBUFFER;
         av_opt_set(fmt_ctx->priv_data,"preset","ultrafast",0);
         //AV_TIME_BASE = 1000 000
-        fmt_ctx->max_analyze_duration = 90 * AV_TIME_BASE;
+        fmt_ctx->max_analyze_duration = 100 * AV_TIME_BASE;
 		fmt_ctx->pb = ioCtx;
@@ -428,9 +423,11 @@ bool DvppRtpDecoder::probe() {
         av_dict_set( &net_options, "bufsize", "655360", 0 );
         av_dict_set( &net_options, "stimeout", "30000000", 0 ); // 单位为 百万分之一秒
         av_dict_set( &net_options, "max_delay", "500000", 0); //设置最大时延
+        av_dict_set( &net_options, "probesize", "50M", 0); //设置最大时延
+         
 		//打开流
-        ret = avformat_open_input(&fmt_ctx, 0, inputFmt, &net_options);
+        int ret = avformat_open_input(&fmt_ctx, 0, 0, &net_options);
 		if (ret != 0) {
 			LOG_ERROR("avformat_open_input error: {}", ret);
 			break;
@@ -565,6 +562,8 @@ void DvppRtpDecoder::read_thread() {
         CHECK_AND_BREAK(aclvdecSetChannelDescOutPicFormat(vdecChannelDesc, PIXEL_FORMAT_YUV_SEMIPLANAR_420), "aclvdecSetChannelDescOutPicFormat failed");
         CHECK_AND_BREAK(aclvdecCreateChannel(vdecChannelDesc), "aclvdecCreateChannel failed");
+        m_buffer_waiting_time = 30000;   //read最多等待5分钟没有数据
+
         unsigned long long frame_nb = 0;
         while (m_bRunning){
@@ -158,5 +158,7 @@ private:
     void*           m_finishParam;
 	CallBack_DecodeFinished m_finish_cbk;				// 录像流结束回调
+    int m_buffer_waiting_time{3000} ;
+
 };
 #endif //__DVPP_RTP_DECODER_H__
 \ No newline at end of file
@@ -61,6 +61,8 @@ bool RTPReceiver2::Open(string channel_id, bool isUdp) {
 		return false;
 	}
+	m_bRtpExit = false;
+
 	bool bReq = start_server(channel_id, m_rtp_port, isUdp);
 	if (!bReq) {
 		LOG_INFO("[{}] start_server failed !", m_SipChannelId);
@@ -91,7 +93,6 @@ void RTPReceiver2::Close(){
 bool RTPReceiver2::start_server(string channel_id, int port, bool isUdp) {
 	WebsocketClient* pClient = WebsocketClient::getInstance();
 	if (pClient){
-
 		if (isUdp) {
 			m_server_thread = new std::thread([](void* arg) {
 				RTPReceiver2* a=(RTPReceiver2*)arg;
@@ -125,7 +126,7 @@ int RTPReceiver2::udp_server() {
 	LOG_INFO("udp {}",port);
-    int server_fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); //AF_INET:IPV4;SOCK_DGRAM:UDP
+    int server_fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); //AF_INET:IPV4;SOCK_DGRAM:UDP
     if(server_fd < 0)
     {
         printf("create socket fail!\n");
@@ -146,19 +147,28 @@ int RTPReceiver2::udp_server() {
 	uint8_t recvBuf[10000];
 	int  recvBufSize;
-
     socklen_t len;
     struct sockaddr_in clent_addr;  //clent_addr用于记录发送方的地址信息
+	long long last_time = get_cur_time_second();
     while(!m_bRtpExit)
     {
         memset(recvBuf, 0, sizeof(recvBuf));
         len = sizeof(clent_addr);
         recvBufSize = recvfrom(server_fd, recvBuf, sizeof(recvBuf), 0, (struct sockaddr*)&clent_addr, &len);  //recvfrom是拥塞函数，没有数据就一直拥塞
         if(recvBufSize <= 0) {
-            LOG_ERROR("recieve data fail!");
-            break;
+			long long cur_time = get_cur_time_second();
+			if (cur_time - last_time > 300) {
+				// 5分钟未能获取数据，则退出
+				m_bRtpExit = true;
+				break;
+			}
+
+			std::this_thread::sleep_for(std::chrono::milliseconds(2));
+            continue;
         }
+		last_time = get_cur_time_second();
+
 		// buffer 抛出
 		m_buffer_cbk(m_bufferParam, recvBuf, recvBufSize, 0);
     }