Hu Chunming / vpt_ascend_arm

  #include "DvppRtpDecoder.h"
  
  #include "DvppSourceManager.h"

  
  
  #define CHECK_AND_RETURN(ret, message)    \
              if(ret != 0) {LOG_ERROR("[{}]- {}", m_dec_name, message); return ret;}
  #define CHECK_NOT_RETURN(ret, message)    \
              if(ret != 0) {LOG_ERROR("[{}]- {}", m_dec_name, message);}
  #define CHECK_AND_RETURN_NOVALUE(ret, message)    \
              if(ret != 0) {LOG_ERROR("[{}]- {}", m_dec_name, message); return;}
  #define CHECK_AND_BREAK(ret, message)    \
              if(ret != 0) {LOG_ERROR("[{}]- {}", m_dec_name, message); break;}
  
  
  
  
  struct Vdec_CallBack_UserData {
      uint64_t frameId;
      uint64_t frame_nb;
      long startTime;
      long sendTime;
  	DvppRtpDecoder* self;
  
      Vdec_CallBack_UserData() {
          frameId = 0;
          frame_nb = 0;
      }
  };
  
  
  static long get_cur_time_ms() {
      chrono::time_point<chrono::system_clock, chrono::milliseconds> tpMicro
          = chrono::time_point_cast<chrono::milliseconds>(chrono::system_clock::now());
      return tpMicro.time_since_epoch().count();
  }
  
  static void *ReportThd(void *arg)
  {
      DvppRtpDecoder *self = (DvppRtpDecoder *)arg;
  	if(nullptr != self){
  		self->doProcessReport();
  	}
      return (void *)0;
  }
  
  static void VdecCallback(acldvppStreamDesc *input, acldvppPicDesc *output, void *pUserData)
  {
  	Vdec_CallBack_UserData *userData = (Vdec_CallBack_UserData *) pUserData;
      if(nullptr != userData){
          DvppRtpDecoder* self = userData->self;
          if(self != nullptr){
              self->doVdppVdecCallBack(input, output, userData);
          }
          delete userData;
  	    userData = nullptr;
      }
  }
  
  static int avio_read_packet(void* opaque, uint8_t* buf, int buffsize){
  	DvppRtpDecoder* rtpDecoder = (DvppRtpDecoder*)opaque;
  	if(rtpDecoder) {
  		return rtpDecoder->ReadBuffer(buf, buffsize);
  	}
  
  	LOG_ERROR("rtpDecoder is null");
  
  	return 0;
  }
  
  DvppRtpDecoder::DvppRtpDecoder(){
      m_read_thread = nullptr;
  
      fmt_ctx = nullptr;
  	m_bRunning = false;
  
      mVideoIndex = -1;
      pix_fmt = AV_PIX_FMT_NONE;
      m_dec_name = "";
  
  	m_bPause = false;
  
  	m_bFinished = false;
  	m_dec_keyframe = false;
  	m_fps = 0.0;
  }
  
  DvppRtpDecoder::~DvppRtpDecoder() {
      Close();
  
      LOG_DEBUG("[{}]- ~DvppRtpDecoder() in_count:{}  out_count:{}", m_dec_name, m_in_count, m_out_count);
  }
  
  bool DvppRtpDecoder::Init(FFDecConfig cfg) {
  
      m_dec_name = cfg.dec_name;
      m_frameSkip = cfg.skip_frame;
  
      m_cfg = cfg;
  
  	m_bResize = m_cfg.resize;
  

  	bool bRet = init_dvpp(cfg);
  	if(!bRet){
  		return false;
  	}
  
  	m_bFinished = false;
  
  	return true;
  }
  
  void DvppRtpDecoder::calcOutResolution(int width, int height) {
      if(m_bResize) {
          float srcRatio = width / (float)height;
          float stdRatio = 1920.0 / 1080.0f ;
          int outWidth = 1920;
          int outHeight = 1080;
          if (srcRatio > stdRatio) {
              outHeight = static_cast<int>(outWidth * (float)height / width) ;
              if (outHeight % 2 == 1) {
                  outHeight += 1;
              }
          } else if (srcRatio < stdRatio) {
              outWidth = static_cast<int>(outHeight * (float)width / height) ;
              if (outWidth % 2 == 1) {
                  outWidth += 1;
              }
          }
  
          out_frame_width = outWidth;
          out_frame_height = outHeight;
      } else {
          out_frame_width = width;
          out_frame_height = height;
      }
  }
  
  int DvppRtpDecoder::getVdecType(int videoType, int profile)
  {
      int streamFormat = H264_MAIN_LEVEL;
  
      // VDEC only support　H265 main level，264 baseline level，main level，high level
      if (videoType == AV_CODEC_ID_HEVC) {
          streamFormat = H265_MAIN_LEVEL;
      } else if (videoType == AV_CODEC_ID_H264) {
          switch (profile) {
              case FF_PROFILE_H264_BASELINE:
                  streamFormat = H264_BASELINE_LEVEL;
                  break;
              case FF_PROFILE_H264_MAIN:
                  streamFormat = H264_MAIN_LEVEL;
                  break;
              case FF_PROFILE_H264_HIGH:
              case FF_PROFILE_H264_HIGH_10:
              case FF_PROFILE_H264_HIGH_10_INTRA:
              case FF_PROFILE_H264_MULTIVIEW_HIGH:
              case FF_PROFILE_H264_HIGH_422:
              case FF_PROFILE_H264_HIGH_422_INTRA:
              case FF_PROFILE_H264_STEREO_HIGH:
              case FF_PROFILE_H264_HIGH_444:
              case FF_PROFILE_H264_HIGH_444_PREDICTIVE:
              case FF_PROFILE_H264_HIGH_444_INTRA:
                  streamFormat = H264_HIGH_LEVEL;
                  break;
              default:
                  LOG_INFO("Not support h264 profile {}, use as mp", profile);
                  streamFormat = H264_MAIN_LEVEL;
                  break;
          }
      } else {
          streamFormat = -1;
          LOG_ERROR("Not support stream, type {},  profile {}", videoType, profile);
      }
  
      return streamFormat;
  }
  
   bool DvppRtpDecoder::init_dvpp(FFDecConfig cfg) {
  
      LOG_INFO("[{}]- Init device start...", m_dec_name);
  
      m_dvpp_deviceId = atoi(cfg.gpuid.c_str());
  
      post_decoded_cbk = cfg.post_decoded_cbk;
  
      do{
          aclError ret = aclrtSetDevice(m_dvpp_deviceId);
          if(ret != ACL_ERROR_NONE){
              LOG_ERROR("[{}]-aclrtSetDevice failed !", m_dec_name);
              break;
          }
  
          ret = aclrtCreateContext(&m_context, m_dvpp_deviceId);
          if (ret != ACL_ERROR_NONE) {
              LOG_ERROR("[{}]-aclrtCreateContext failed !", m_dec_name);
              break;
          }
  
          // DvppSourceManager 创建时包含 aclInit，析构时包含 aclFinalize
          DvppSourceManager* pSrcMgr = DvppSourceManager::getInstance();
          m_dvpp_channel = pSrcMgr->getChannel(m_dvpp_deviceId);
          if(m_dvpp_channel < 0){
              LOG_ERROR("[{}]-该设备channel已经用完了!", m_dec_name);
              break;
          }
  
          m_vpcUtils.init(m_dvpp_deviceId);
  
          LOG_INFO("[{}]- init vdpp success! device:{} channel:{}", m_dec_name, m_dvpp_deviceId, m_dvpp_channel);
          return true;
      }while(0);
  
      release_dvpp();
  
      return false;
  }
  
  bool DvppRtpDecoder::isSurport(FFDecConfig& cfg){
      return true;
  }
  
  bool DvppRtpDecoder::start(){

  
      m_bRunning = true;

      
      if(!probe()) {
  		return false;
  	}
  

      m_read_thread = new std::thread([](void* arg)
          {
              DvppRtpDecoder* a=(DvppRtpDecoder*)arg;
              a->read_thread();
              return (void*)0;
          }, this);
  
  	return true;
  }
  
  void DvppRtpDecoder::Close(){
      m_bRunning=false;
  
  	if(m_read_thread != nullptr){
          m_read_thread->join();
          delete m_read_thread;
          m_read_thread = nullptr;
  	}
  
  	m_recoderManager.close();
      
      release_ffmpeg();
      release_dvpp();
  }
  
  void DvppRtpDecoder::setPostDecArg(const void* postDecArg){
      m_postDecArg = postDecArg;
  }
  
  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;
  }
  
  void DvppRtpDecoder::resume(){
      m_bPause = false;
  }
  
  void DvppRtpDecoder::setDecKeyframe(bool bKeyframe){
      m_dec_keyframe = bKeyframe;
  }
  
  bool DvppRtpDecoder::isRunning(){
      return m_bRunning;
  }
  
  bool DvppRtpDecoder::isFinished(){
      return m_bFinished;
  }
  
  bool DvppRtpDecoder::isPausing(){
      return m_bPause;
  }
  
  bool DvppRtpDecoder::getResolution(int &width, int &height){
      width = frame_width;
  	height = frame_height;
  	return true;
  }
  
  bool DvppRtpDecoder::getOutResolution( int &width, int &height ) {  
      width = out_frame_width;
      height = out_frame_height;
      return true;
  }
  
  float DvppRtpDecoder::fps(){
      return m_fps;
  }
  
  static int snap_count = 0;
  
  DeviceMemory* DvppRtpDecoder::snapshot(){
  
      int ret = aclrtSetCurrentContext(m_context);
      if(ret != ACL_ERROR_NONE){
          LOG_ERROR("[{}]- aclrtSetCurrentContext failed", m_dec_name);
          return nullptr;
      }
  
      // 注意有锁
      DeviceMemory* snapshot_mem = nullptr;
      int loop_times = 0;
      while(m_bRunning) {
          m_decoded_data_queue_mtx.lock();
          if(m_decoded_data_queue.size() <= 0) {
              m_decoded_data_queue_mtx.unlock();
              loop_times++;
              if(loop_times > 100) {
                  // 1s都没截取到图，退出
                  break;
              }
              std::this_thread::sleep_for(std::chrono::milliseconds(10));
              continue;
          }
  
          DvppDataMemory* mem = m_decoded_data_queue.front();
          snapshot_mem = new DvppDataMemory(mem);
          m_decoded_data_queue_mtx.unlock();
  
          // snap_count++;
          // LOG_INFO("[{}]- snap_count:{} ", m_dec_name, snap_count);
          break;
      }
  
      return snapshot_mem;
  }
  
  void DvppRtpDecoder::release_ffmpeg() {
  	m_dec_keyframe = false;
  	if(h264bsfc){
  		av_bsf_free(&h264bsfc);
  		h264bsfc = nullptr;
  	}
      if(avctx){
          avcodec_free_context(&avctx);
          avctx = nullptr;
      }
      if (fmt_ctx){
  		avformat_close_input(&fmt_ctx);
  		fmt_ctx = nullptr;
  	}
  
      LOG_DEBUG("[{}]- release_ffmpeg", m_dec_name);
  }
  
  void DvppRtpDecoder::CacheBuffer(uint8_t* recvBuf, int recvBufSize) {

  	if ((m_bufferSize + recvBufSize) < MAX_RTP_BUFFER_SIZE) {
          memcpy(m_buffer + m_bufferSize, recvBuf, recvBufSize);
          m_bufferSize += recvBufSize;

      } else {

  		LOG_WARN("recvBufSize = {} over MAX_RTP_BUFFER_SIZE ", recvBufSize);
  	}
  }
  
  int DvppRtpDecoder::ReadBuffer(uint8_t* buf, int buffsize) {

      int count = 0;

      while(m_bufferSize < buffsize){
          if(!m_bRunning){
              return AVERROR_EXIT;
          }

          std::this_thread::sleep_for(std::chrono::milliseconds(10));
          count++;

          if (count >= m_buffer_waiting_time) {
              // 等待

              return AVERROR(EIO);

          }
      }
  

      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);

  	return buffsize;

  }
  
  bool DvppRtpDecoder::probe() {

  
      m_buffer_waiting_time = 3000;   //probe只等待30s，避免卡主任务添加

  	// todo: 此处可能有泄露

  	unsigned char* avioBuff = (unsigned char*)av_malloc(2048);  // 32768 < MAX_RTP_BUFFER_SIZE - RTP_HEADER_SIZE
  	AVIOContext  *ioCtx = avio_alloc_context(avioBuff, 2048, 0, this, avio_read_packet, NULL, NULL);

  
  	do{
  		fmt_ctx = avformat_alloc_context();

          // 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 = 100 * AV_TIME_BASE;

  		fmt_ctx->pb = ioCtx;
  
  
  		AVDictionary* net_options{nullptr};//网络连接参数

  		// av_dict_set(&net_options, "fflags", "nobuffer", 0); //不缓存直接解码
          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); //设置最大时延
           

  
  		//打开流

          int ret = avformat_open_input(&fmt_ctx, 0, 0, &net_options);

  		if (ret != 0) {

  			LOG_ERROR("avformat_open_input error: {}", ret);
  			break;
  		}

  		//获取流信息

  		if (avformat_find_stream_info(fmt_ctx, NULL) < 0) {

  			LOG_ERROR("avformat_find_stream_info error");
  			break;
  		}

  		//获取视频流
  		mVideoIndex = av_find_best_stream(fmt_ctx, AVMEDIA_TYPE_VIDEO, -1, -1, NULL, 0);
  		if (mVideoIndex < 0)
  		{
  			LOG_ERROR("av_find_best_stream error");
  			break;
  		}
  		//获取解码信息
  		AVStream* stream = fmt_ctx->streams[mVideoIndex];
  		AVCodecParameters *codecpar = stream->codecpar;
  		const AVCodec* videoCodec = avcodec_find_decoder(codecpar->codec_id);
  		if (!videoCodec){
  			LOG_ERROR("avcodec_find_decoder error");
  			break;
  		}
  		avctx = avcodec_alloc_context3(videoCodec);
  
  		//codecpar为解码器上下文赋值
  		if (avcodec_parameters_to_context(avctx, codecpar) != 0)
  		{
  			LOG_ERROR("avcodec_parameters_to_context error");
  			break;
  		}
  
  		int enType = getVdecType(codecpar->codec_id, codecpar->profile);
  		if(-1 == enType) {
  			break;
  		}
  		m_enType = static_cast<acldvppStreamFormat>(enType);
  
  		const AVBitStreamFilter * filter = nullptr;
  		if(codecpar->codec_id == AV_CODEC_ID_H264){
  			filter = av_bsf_get_by_name("h264_mp4toannexb");
  		}else if(codecpar->codec_id == AV_CODEC_ID_HEVC){
  			filter = av_bsf_get_by_name("hevc_mp4toannexb");
  		}else {
  			LOG_ERROR("[{}]- codec_id is not supported!", m_dec_name);
  			break;
  		}
  
  		ret = av_bsf_alloc(filter, &h264bsfc);
  		if (ret < 0){
  			break;
  		}
  		
  		avcodec_parameters_copy(h264bsfc->par_in, codecpar);
  		av_bsf_init(h264bsfc);
  
  		frame_width = codecpar->width;
  		frame_height = codecpar->height;
  		pix_fmt = (AVPixelFormat)codecpar->format;
  
  		calcOutResolution(frame_width, frame_height);
  
  		if (stream->avg_frame_rate.den) {
  			m_fps = av_q2d(stream ->avg_frame_rate);
  		} else {
  			m_fps = 0.0;
  		}
  
  		m_vdec_out_size = frame_width * frame_height * 3 / 2;
  
  		if (avctx->gop_size > 0) {
  			m_cache_gop = avctx->gop_size + 1;
  		} else {
  			m_cache_gop = 20;
  		}
  
  	#ifdef USE_VILLAGE
  		bool bRet = m_recoderManager.init(frame_width, frame_height, m_fps, avctx->bit_rate);
  		if (!bRet){
  			LOG_ERROR("[{}]- m_recoderManager 初始化失败!", m_dec_name);
  		}  
  	#endif
  
  		LOG_INFO("[{}]- init ffmpeg success! src:({}, {}) out:({}, {}) fps:{} ", m_dec_name, frame_width, frame_height, out_frame_width, out_frame_height, m_fps);
  
  		return true;
  	} while(0);
  
  	release_ffmpeg();
  
  	return false;
  }
  
  void DvppRtpDecoder::read_thread() {
  
  	int ret = -1;
  
      m_bExitReportThd = false;
  	pthread_t report_thread;
  	ret = pthread_create(&report_thread, nullptr, ReportThd, (void *)this);
  	if(ret != 0){
          LOG_ERROR("[{}]- pthread_create failed", m_dec_name);
  		return;
  	}
  
      aclrtContext ctx = nullptr;
      aclvdecChannelDesc *vdecChannelDesc = nullptr;
  
      do {
          CHECK_AND_BREAK(aclrtSetDevice(m_dvpp_deviceId), "aclrtSetDevice failed");
          int ret = aclrtSetCurrentContext(m_context);
          if(ret != ACL_ERROR_NONE){
              LOG_ERROR("[{}]- aclrtSetCurrentContext failed", m_dec_name);
              break;
          }
  
          vdecChannelDesc = aclvdecCreateChannelDesc();
          if (vdecChannelDesc == nullptr) { 
              LOG_ERROR("[{}]- aclvdecCreateChannelDesc failed", m_dec_name);
              break;
          }
  
          // 创建 channel dec结构体
          // 通道ID在dvpp层面为0~31
          CHECK_AND_BREAK(aclvdecSetChannelDescChannelId(vdecChannelDesc, m_dvpp_channel), "aclvdecSetChannelDescChannelId failed");
          CHECK_AND_BREAK(aclvdecSetChannelDescThreadId(vdecChannelDesc, report_thread), "aclvdecSetChannelDescThreadId failed");
          CHECK_AND_BREAK(aclvdecSetChannelDescCallback(vdecChannelDesc, VdecCallback), "aclvdecSetChannelDescCallback failed");
          CHECK_AND_BREAK(aclvdecSetChannelDescEnType(vdecChannelDesc, m_enType), "aclvdecSetChannelDescEnType failed");
          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){
  
              AVPacket* pkt = av_packet_alloc();
              av_init_packet( pkt );
              int result = av_read_frame(fmt_ctx, pkt);
              if (result == AVERROR_EOF || result < 0){
                  av_packet_free(&pkt);
                  pkt = nullptr;

                  // LOG_WARN("[{}]- Failed to read frame!", m_dec_name);
                  continue;

              }
  

              // av_packet_free(&pkt);
              // pkt = nullptr;
              // continue;
  

              if (m_DvppCacheCounter.load() > m_cache_gop){
                  // 解码器解码不过来。实时流在此处的处理会导致花屏，这是由于解码器性能问题导致，无法避免
                  // 实时流在这里处理是为了避免长时间不读取数据导致数据中断

                  av_packet_free(&pkt);
                  pkt = nullptr;

                  std::this_thread::sleep_for(std::chrono::milliseconds(10));
                  continue;
              }
  
              if (m_dec_keyframe && !(pkt->flags & AV_PKT_FLAG_KEY)) {
                  av_packet_free(&pkt);
                  pkt = nullptr;
                  continue;
              }
  
              if (mVideoIndex == pkt->stream_index){
  
                  ret = av_bsf_send_packet(h264bsfc, pkt);
                  if(ret < 0) {
                      LOG_ERROR("[{}]- av_bsf_send_packet error!", m_dec_name);
                      av_packet_free(&pkt);
                      pkt = nullptr;
                      continue;
                  }
  
                  frame_nb++;
                  int nSended = -1;
                  while ((ret = av_bsf_receive_packet(h264bsfc, pkt)) == 0) {
                      if(!m_bRunning){
                          break;
                      }
                      nSended = sendPkt(vdecChannelDesc, pkt, frame_nb);
                  }
  
                  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);
                  pkt = nullptr;
              }
          }
  
          if (vdecChannelDesc) {
              sendVdecEos(vdecChannelDesc);
          }
  
          while(m_bRunning && m_decoded_data_queue.size() > 0) {
              std::this_thread::sleep_for(std::chrono::milliseconds(5));
          }
  
      } while (0);
  
      if (vdecChannelDesc) {
          CHECK_NOT_RETURN(aclvdecDestroyChannel(vdecChannelDesc), "aclvdecDestroyChannel failed");
          CHECK_NOT_RETURN(aclvdecDestroyChannelDesc(vdecChannelDesc), "aclvdecDestroyChannelDesc failed");
          vdecChannelDesc = nullptr;
      }
  
      m_bRunning=false;
  
      m_bExitReportThd = true;
  	CHECK_NOT_RETURN(pthread_join(report_thread, nullptr), "report_thread join failed");
  
      m_bFinished = true;
  
      LOG_INFO("[{}]- read thread exit.", m_dec_name);
  

      if(m_finish_cbk) {
          m_finish_cbk(m_finishParam);

      }
  }
  
  int DvppRtpDecoder::sendPkt(aclvdecChannelDesc *vdecChannelDesc, AVPacket* pkt, unsigned long long frame_nb){
  
      void *vdecInputbuf = nullptr;
      void *vdecOutputBuf = nullptr;
      acldvppStreamDesc *input_stream_desc = nullptr;
      acldvppPicDesc *output_pic_desc = nullptr;
      do{
          int ret = acldvppMalloc((void **)&vdecInputbuf, pkt->size);
          if(ACL_ERROR_NONE != ret){
              LOG_ERROR("[{}]- acldvppMalloc failed!, ret:{}", m_dec_name, ret);
              break;
          }
  
           ret = aclrtMemcpy(vdecInputbuf, pkt->size, pkt->data, pkt->size, ACL_MEMCPY_HOST_TO_DEVICE);
          if(ACL_ERROR_NONE != ret){
              LOG_ERROR("[{}]- aclrtMemcpy failed", m_dec_name);
              break;
          }
  
          ret = acldvppMalloc((void **)&vdecOutputBuf, m_vdec_out_size);
          if(ret != ACL_ERROR_NONE){
              LOG_ERROR("[{}]- acldvppMalloc failed", m_dec_name);
              break;
          }
  
          input_stream_desc = acldvppCreateStreamDesc();
          if (input_stream_desc == nullptr) { 
              LOG_ERROR("[{}]- acldvppCreateStreamDesc failed", m_dec_name);
              break;
          }
          output_pic_desc = acldvppCreatePicDesc();
          if (output_pic_desc == nullptr) { 
              LOG_ERROR("[{}]- acldvppCreatePicDesc failed", m_dec_name);
              break;
          }
          CHECK_AND_BREAK(acldvppSetStreamDescData(input_stream_desc, vdecInputbuf), "acldvppSetStreamDescData failed");
          CHECK_AND_BREAK(acldvppSetStreamDescSize(input_stream_desc, pkt->size), "acldvppSetStreamDescSize failed");
          CHECK_AND_BREAK(acldvppSetPicDescData(output_pic_desc, vdecOutputBuf), "acldvppSetPicDescData failed");
          CHECK_AND_BREAK(acldvppSetPicDescSize(output_pic_desc, m_vdec_out_size), "acldvppSetPicDescSize failed");
          
          Vdec_CallBack_UserData *user_data = NULL;
          user_data = new Vdec_CallBack_UserData;
          user_data->frameId = frame_nb;
          user_data->frame_nb = frame_nb;
          // user_data->startTime = startTime;
          user_data->sendTime = UtilTools::get_cur_time_ms();
          user_data->self = this;
  
          m_in_count++;
  
          // 内部缓存计数加1
          m_DvppCacheCounter++;
          ret = aclvdecSendFrame(vdecChannelDesc, input_stream_desc, output_pic_desc, nullptr, reinterpret_cast<void *>(user_data));
          if(ret != ACL_ERROR_NONE){
              LOG_ERROR("[{}]- aclvdecSendFrame failed", m_dec_name);
              delete user_data;
              user_data = nullptr;
              return -2;
          }
  
          return 0;
      }while (0);
  
      if (vdecInputbuf){
          acldvppFree(vdecInputbuf);
          vdecInputbuf = nullptr;
      }
  
      // 报错情形
      if(input_stream_desc){
          CHECK_NOT_RETURN(acldvppDestroyStreamDesc(input_stream_desc), "acldvppDestroyStreamDesc failed");
      }
  
      if (vdecOutputBuf){
          acldvppFree(vdecOutputBuf);
  	    vdecOutputBuf = nullptr;
      }
  
      if(output_pic_desc){
          CHECK_NOT_RETURN(acldvppDestroyPicDesc(output_pic_desc), "acldvppDestroyPicDesc failed");
      }
  
      return -1;
  }
  
  void DvppRtpDecoder::doProcessReport(){
  
      aclError ret = aclrtSetDevice(m_dvpp_deviceId);
      if(ret != ACL_ERROR_NONE){
          // cout << "aclrtSetDevice failed" << endl;
          LOG_ERROR("aclrtSetDevice failed !");
          return ;
      }
  
      aclrtContext ctx;
      ret = aclrtCreateContext(&ctx, m_dvpp_deviceId);
      if (ret != ACL_ERROR_NONE) {
          // cout << "aclrtCreateContext failed " << endl;
          LOG_ERROR("aclrtCreateContext failed !");
          return ;
      }
  
      while (!m_bExitReportThd) {
          aclrtProcessReport(1000);
      }
  
      ret = aclrtDestroyContext(ctx);
      if(ret != ACL_ERROR_NONE){
          LOG_ERROR("aclrtDestroyContext failed !");
      }
      LOG_INFO("doProcessReport exit.");
  }
  
  void DvppRtpDecoder::doVdppVdecCallBack(acldvppStreamDesc *input, acldvppPicDesc *output, void *pUserData){
  
      // 内部缓存计数减1
      m_DvppCacheCounter--;
  
      if(nullptr == pUserData){
          return;
      }
  
      Vdec_CallBack_UserData *userData = (Vdec_CallBack_UserData *) pUserData;
      uint64_t frame_nb = userData->frame_nb;
  
      m_out_count++;
  
      CHECK_AND_RETURN_NOVALUE(aclrtSetCurrentContext(m_context), "aclrtSetCurrentContext failed");
  
      void *inputDataDev = acldvppGetStreamDescData(input);
      acldvppFree(inputDataDev);
      inputDataDev = nullptr;
  
      void *outputDataDev = acldvppGetPicDescData(output);
      uint32_t outputSize = acldvppGetPicDescSize(output);
      uint32_t width = acldvppGetPicDescWidth(output);
      uint32_t width_stride = acldvppGetPicDescWidthStride(output);
      uint32_t height = acldvppGetPicDescHeight(output);
      uint32_t height_stride = acldvppGetPicDescHeightStride(output);
  
      do{
          int ret = acldvppGetPicDescRetCode(output);
          if(ret != ACL_ERROR_NONE){
              LOG_ERROR("[{}]- decode result error, retCode:{} ", m_dec_name, ret);
              acldvppFree(outputDataDev);
              outputDataDev = nullptr;
              break;
          }
  
          bool bCached = false;
          if(width > 0 && height > 0 && outputSize > 0){
              
              // cout << m_dec_name << " 解码时间间隔： " << get_cur_time_ms() - last_ts << endl;
              // last_ts = get_cur_time_ms();
  
              // 换成解码后数据， 这里这样做的是为了保证解码一直持续进行，避免后续操作阻碍文件读取和解码从而导致花屏
              DvppDataMemory* mem = nullptr;
              if (m_bResize && (width > 1920 || height > 1080)) {
                  
                  mem = m_vpcUtils.resize(output, out_frame_width, out_frame_height);
                  if (mem)  {
                      acldvppFree(outputDataDev);
                      outputDataDev = nullptr;
  
                      mem->setDeviceId(to_string(m_dvpp_deviceId));
                      mem->setId(m_dec_name);
                      mem->setFrameNb(frame_nb);
                  }
              } else {
                  mem = new DvppDataMemory(width, width_stride, height, height_stride, outputSize, m_dec_name, to_string(m_dvpp_deviceId), false, frame_nb, (unsigned char *)outputDataDev);
              }
              
              if(mem){
                  m_decoded_data_queue_mtx.lock();
                  m_decoded_data_queue.push(mem);
                  m_decoded_data_queue_mtx.unlock();
                  bCached = true;
              }
          } 
          
          if(!bCached) {
              LOG_WARN("[{}]- decode result warning, width:{} width_stride:{} height:{} height_stride:{} size:{}", m_dec_name, width, width_stride, height, height_stride, outputSize);
              acldvppFree(outputDataDev);
              outputDataDev = nullptr;
          }
      }while(0);
  
  	CHECK_AND_RETURN_NOVALUE(acldvppDestroyStreamDesc(input), "acldvppDestroyStreamDesc failed");
  	CHECK_AND_RETURN_NOVALUE(acldvppDestroyPicDesc(output), "acldvppDestroyPicDesc failed");
  }
  
  bool DvppRtpDecoder::sendVdecEos(aclvdecChannelDesc *vdecChannelDesc) {
      // create stream desc
      acldvppStreamDesc *streamInputDesc = acldvppCreateStreamDesc();
      if (streamInputDesc == nullptr) {
          LOG_ERROR("[{}]- fail to create input stream desc", m_dec_name);
          return false;
      }
      aclError ret = acldvppSetStreamDescEos(streamInputDesc, 1);
      if (ret != ACL_SUCCESS) {
          LOG_ERROR("[{}]- fail to set eos for stream desc, errorCode = {}", m_dec_name, static_cast<int32_t>(ret));
          (void)acldvppDestroyStreamDesc(streamInputDesc);
          return false;
      }
  
      // send vdec eos frame. when all vdec callback are completed, aclvdecSendFrame can be returned.
      LOG_INFO("[{}]- send eos", m_dec_name);
      ret = aclvdecSendFrame(vdecChannelDesc, streamInputDesc, nullptr, nullptr, nullptr);
      (void)acldvppDestroyStreamDesc(streamInputDesc);
      if (ret != ACL_SUCCESS) {
          LOG_ERROR("[{}]- fail to send eos frame, ret={}", m_dec_name, ret);
          return false;
      }
  
      return true;
  }
  
  DvppDataMemory* DvppRtpDecoder::GetFrame() {
      DvppDataMemory* mem = nullptr;
      m_decoded_data_queue_mtx.lock();
      if (m_decoded_data_queue.size() > 0) {
          mem = m_decoded_data_queue.front();
          m_decoded_data_queue.pop();
      }
      m_decoded_data_queue_mtx.unlock();
  
      return mem;
  }
  
  void DvppRtpDecoder::release_dvpp(){
      if(m_context){
          aclError ret = aclrtDestroyContext(m_context);
          if(ret != ACL_ERROR_NONE){
              LOG_ERROR("[{}]- aclrtDestroyContext failed !", m_dec_name);
          }
          m_context = nullptr;
      }
      
      if(m_dvpp_channel >= 0){
          DvppSourceManager* pSrcMgr = DvppSourceManager::getInstance();
  	    pSrcMgr->releaseChannel(m_dvpp_deviceId, m_dvpp_channel);
          m_dvpp_channel = -1;
      }
  }
  
  void DvppRtpDecoder::doRecode(RecoderInfo& recoderInfo) {
      m_recoderManager.create_recode_task(recoderInfo);
  }
  
  void DvppRtpDecoder::set_mq_callback(mq_callback_t cb) {
      m_recoderManager.set_mq_callback(cb);
  }