Hu Chunming / vpt_ascend_arm

#include "DvppDecoder.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;
	DvppDecoder* 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)
{
    DvppDecoder *self = (DvppDecoder *)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){
        DvppDecoder* self = userData->self;
        if(self != nullptr){
            self->doVdppVdecCallBack(input, output, userData);
        }
        delete userData;
	    userData = nullptr;
    }
}

static int AVInterruptCallBackFun(void *param)
{
    DvppDecoder *pDecoder = (DvppDecoder*)param;
    if (nullptr == pDecoder) return 0;

    long long cur_ts = get_cur_time_ms();
    long time_gap = cur_ts - pDecoder->get_last_read_ts();
    if (time_gap > 1000*30) {
        LOG_WARN("摄像头异常，释放阻塞");
        //通知FFMpeg可以从阻塞工作线程中释放操作
        return 1;
    } else {
        //通知FFMpeg继续阻塞工作
        return 0;
    }
}

DvppDecoder::DvppDecoder(){
    m_read_thread = nullptr;

    fmt_ctx = nullptr;
	m_bRunning = false;

    video_index = -1;
    pix_fmt = AV_PIX_FMT_NONE;
    m_dec_name = "";

	m_bPause = false;
	m_bReal = true;

	m_bFinished = false;
	m_dec_keyframe = false;
	m_fps = 0.0;
}

DvppDecoder::~DvppDecoder(){
    close();

    LOG_DEBUG("[{}]- ~DvppDecoder() in_count:{}  out_count:{}", m_dec_name, m_in_count, m_out_count);
}

bool DvppDecoder::init(FFDecConfig cfg){

    m_dec_name = cfg.dec_name;
    m_frameSkip = cfg.skip_frame;

    AVCodecContext* avctx = init_FFmpeg(cfg);
    if(avctx == nullptr){
        return false;
    }

    do
    {
        bool bRet = init_dvpp(cfg);
        if(!bRet){
            break;
        }

        m_cfg = cfg;

        decode_finished_cbk = cfg.decode_finished_cbk;

        m_bFinished = false;

        return true;
    } while (0);
    
    release_ffmpeg();
    
    return false;
}

AVCodecContext* DvppDecoder::init_FFmpeg(FFDecConfig config){

    const char* input_file = config.uri.c_str();

	// 打开输入视频文件
	AVDictionary *options = nullptr;
	av_dict_set( &options, "bufsize", "655360", 0 );
	av_dict_set( &options, "rtsp_transport", config.force_tcp ? "tcp" : "udp", 0 );
	av_dict_set( &options, "stimeout", "30000000", 0 ); // 单位为 百万分之一秒
    av_dict_set( &options, "max_delay", "500000", 0); //设置最大时延

    do{
        fmt_ctx = avformat_alloc_context();
        // fmt_ctx->flags |= AVFMT_FLAG_NONBLOCK;
        if (avformat_open_input(&fmt_ctx, input_file, nullptr, &options) != 0) {
            LOG_ERROR("[{}]- Cannot open input file: {}", m_dec_name, input_file);
            break;
        }
        av_dump_format(fmt_ctx, 0, input_file, 0);

        FILE* infile = fopen(input_file, "r");
        if(nullptr != infile) {
            // 可以作为本地文件打开，那必然是文件
            m_bReal = false;
            fclose(infile);
            infile = nullptr;
        } else {
            // http://xxxx/xxx.mp4   也与文件同
            if (fmt_ctx->iformat && fmt_ctx->iformat->extensions) {
                m_bReal = false;
            } else {
                m_bReal = true;

                fmt_ctx->interrupt_callback.callback = AVInterruptCallBackFun;
                fmt_ctx->interrupt_callback.opaque = this;
                m_last_read_ts = get_cur_time_ms();
            }
        }

        // 查找流信息
        if (avformat_find_stream_info(fmt_ctx, nullptr) < 0) {
            LOG_ERROR("[{}]- Cannot find input stream information!", m_dec_name);
            break;
        }

        // 查找视频流信息
        AVCodec *decoder = nullptr;
        video_index = av_find_best_stream(fmt_ctx, AVMEDIA_TYPE_VIDEO, -1, -1, &decoder, 0);
        if (video_index < 0) {
            LOG_ERROR("[{}]- Cannot find a video stream in the input file!", m_dec_name);
            break;
        }

        avctx = avcodec_alloc_context3(decoder);
        if(avctx == nullptr){
            LOG_ERROR("[{}]- alloc AVCodecContext failed!", m_dec_name);
            break;
        }
	
		// 得到视频流对象
		AVStream* stream = fmt_ctx->streams[video_index];
		AVCodecParameters *codecpar = stream->codecpar;
		if (avcodec_parameters_to_context(avctx, codecpar) < 0)
			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;
		}

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

        m_bResize = config.resize;
        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! input:{} src:({}, {}) out:({}, {}) fps:{} ", m_dec_name, input_file, frame_width, frame_height, out_frame_width, out_frame_height, m_fps);

		return avctx;
	}while(0);

    release_ffmpeg();

    LOG_ERROR("[{}]- init ffmpeg failed ! input:{} ", m_dec_name, input_file);

    return nullptr;
}

void DvppDecoder::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 DvppDecoder::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 DvppDecoder::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 = 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 DvppDecoder::isSurport(FFDecConfig& cfg){
    return true;
}

bool DvppDecoder::start(){
    m_bRunning = true;

    m_read_thread = new std::thread([](void* arg)
        {
            DvppDecoder* a=(DvppDecoder*)arg;
            a->read_thread();
            return (void*)0;
        }, this);

	return true;
}

void DvppDecoder::close(){
    m_bRunning=false;

	if(m_read_thread != nullptr){
        m_read_thread->join();
        delete m_read_thread;
        m_read_thread = nullptr;
	}
}

void DvppDecoder::setPostDecArg(const void* postDecArg){
    m_postDecArg = postDecArg;
}

void DvppDecoder::setFinishedDecArg(const void* finishedDecArg){
    m_finishedDecArg = finishedDecArg;
}

void DvppDecoder::pause(){
    m_bPause = true;
}

void DvppDecoder::resume(){
    m_bPause = false;
}

void DvppDecoder::setDecKeyframe(bool bKeyframe){
    m_dec_keyframe = bKeyframe;
}

bool DvppDecoder::isRunning(){
    return m_bRunning;
}

bool DvppDecoder::isFinished(){
    return m_bFinished;
}

bool DvppDecoder::isPausing(){
    return m_bPause;
}

bool DvppDecoder::getResolution(int &width, int &height){
    width = frame_width;
	height = frame_height;
	return true;
}

bool DvppDecoder::getOutResolution( int &width, int &height ) {  
    width = out_frame_width;
    height = out_frame_height;
    return true;
}

float DvppDecoder::fps(){
    return m_fps;
}

long long DvppDecoder::get_last_read_ts() {
    return m_last_read_ts;
}

static int snap_count = 0;

DeviceMemory* DvppDecoder::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;
}

int DvppDecoder::getCachedQueueLength(){
    return 0;
}

void DvppDecoder::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 DvppDecoder::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;
	}

    m_bExitDisplayThd = false;
    std::thread display_thread(
        [](void* arg)
        {
            DvppDecoder* a=(DvppDecoder*)arg;
            a->display_thread();
            return (void*)0;
        },
        this
    );

    aclvdecChannelDesc *vdecChannelDesc = nullptr;

    do {
        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");

        unsigned long long frame_nb = 0;
        AVPacket* pkt = av_packet_alloc();
        av_init_packet( pkt );
        while (m_bRunning){

            if (!m_bReal){
                if (m_bPause){
                    std::this_thread::sleep_for(std::chrono::milliseconds(10));
                    continue;
                }
                
                // 非实时流，即为文件情形，因为不存在花屏问题，为保证不丢帧，这里做个循环等待
                m_decoded_data_queue_mtx.lock();
                if(m_decoded_data_queue.size() > 5){
                    m_decoded_data_queue_mtx.unlock();
                    std::this_thread::sleep_for(std::chrono::milliseconds(5));
                    continue;
                }
                m_decoded_data_queue_mtx.unlock();

                if(m_DvppCacheCounter.load() > m_cache_gop) {
                    // 解码器解码不过来
                    std::this_thread::sleep_for(std::chrono::milliseconds(10));
                    continue;
                }
            } else {
                m_last_read_ts = get_cur_time_ms();
            }

            int result = av_read_frame(fmt_ctx, pkt);
            if (result == AVERROR_EOF || result < 0){
                av_packet_unref(pkt);
                LOG_WARN("[{}]- Failed to read frame!", m_dec_name);
                break;
            }

            if (m_bReal && m_DvppCacheCounter.load() > m_cache_gop){
                // 解码器解码不过来。实时流在此处的处理会导致花屏，这是由于解码器性能问题导致，无法避免
                // 实时流在这里处理是为了避免长时间不读取数据导致数据中断
                av_packet_unref(pkt);
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                continue;
            }

            if (m_dec_keyframe && !(pkt->flags & AV_PKT_FLAG_KEY)) {
                av_packet_unref(pkt);
                continue;
            }

            if (video_index == 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_unref(pkt);
                    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;
                    break;
                }

    #ifdef USE_VILLAGE
                m_recoderManager.cache_pkt(pkt, frame_nb, m_dec_name);
    #endif
            } 

            av_packet_unref(pkt);
        }

        av_packet_free(&pkt);
        pkt = nullptr;

    } while (0);

    if (vdecChannelDesc) {
        sendVdecEos(vdecChannelDesc);

        CHECK_NOT_RETURN(aclvdecDestroyChannel(vdecChannelDesc), "aclvdecDestroyChannel failed");
        CHECK_NOT_RETURN(aclvdecDestroyChannelDesc(vdecChannelDesc), "aclvdecDestroyChannelDesc failed");
        vdecChannelDesc = nullptr;
    }

    m_bExitReportThd = true;
	CHECK_NOT_RETURN(pthread_join(report_thread, nullptr), "report_thread join failed");

    while(m_bRunning && m_decoded_data_queue.size() > 0) {
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }

    m_bRunning=false;

    m_bExitDisplayThd = true;
    display_thread.join();

    m_recoderManager.close();
    
    release_ffmpeg();
    release_dvpp();

    m_bFinished = true;

    LOG_INFO("[{}]- read thread exit.", m_dec_name);

    if(decode_finished_cbk) {
        decode_finished_cbk(m_finishedDecArg);
    }
}

int DvppDecoder::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");
        CHECK_AND_BREAK(acldvppSetPicDescRetCode(output_pic_desc, 0), "acldvppSetPicDescRetCode 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 DvppDecoder::doProcessReport(){

    aclError ret = aclrtSetDevice(m_dvpp_deviceId);
    if(ret != ACL_ERROR_NONE){
        LOG_ERROR("[{}]-aclrtSetDevice failed !", m_dec_name);
        return;
    }

    while (!m_bExitReportThd) {
        aclrtProcessReport(1000);
    }

    ret = aclrtResetDevice(m_dvpp_deviceId);
    if(ret != ACL_ERROR_NONE){
        LOG_ERROR("aclrtDestroyContext failed !");
    }
    LOG_INFO("doProcessReport exit.");
}

void DvppDecoder::doVdppVdecCallBack(acldvppStreamDesc *input, acldvppPicDesc *output, void *pUserData){

    // 内部缓存计数减1
    m_DvppCacheCounter--;

    CHECK_AND_RETURN_NOVALUE(aclrtSetCurrentContext(m_context), "aclrtSetCurrentContext failed");

    void *inputDataDev = acldvppGetStreamDescData(input);
    acldvppFree(inputDataDev);
    inputDataDev = nullptr;

    if(nullptr == pUserData){
        return;
    }

    Vdec_CallBack_UserData *userData = (Vdec_CallBack_UserData *) pUserData;
    uint64_t frame_nb = userData->frame_nb;

    m_out_count++;

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

        if (m_bReal) {
            // 实时流缓存长度大于25时做丢帧处理
            m_decoded_data_queue_mtx.lock();
            if(m_decoded_data_queue.size() >= 25){
                m_decoded_data_queue_mtx.unlock();
                LOG_WARN("[{}]- m_decoded_data_queue >= 25 ", m_dec_name);
                acldvppFree(outputDataDev);
                outputDataDev = nullptr;
                break;
            }
            m_decoded_data_queue_mtx.unlock();
        }

        bool bCached = false;
        if(width > 0 && height > 0 && outputSize > 0){

            if (!m_bReal) {
                while(!m_bExitReportThd) {
                    // 非实时流，即为文件情形，因为不存在花屏问题，为保证不丢帧，这里做个循环等待
                    m_decoded_data_queue_mtx.lock();
                    if(m_decoded_data_queue.size() >= 5){
                        m_decoded_data_queue_mtx.unlock();
                        std::this_thread::sleep_for(std::chrono::milliseconds(5));
                        continue;
                    }
                    m_decoded_data_queue_mtx.unlock();
                    break;
                }
            }
            
            // 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);
                CHECK_AND_RETURN_NOVALUE(aclrtSetCurrentContext(m_context), "aclrtSetCurrentContext failed");
                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.push(mem);
                bCached = true;
            } else {
                // 逻辑完善
                delete mem;
                mem = nullptr;
            }
        } 
        
        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 DvppDecoder::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;
}

void DvppDecoder::display_thread() {
    LOG_INFO("[{}]- display_thread start...", m_dec_name);
    unsigned long index = 0;
    while(!m_bExitDisplayThd) {
        m_decoded_data_queue_mtx.lock();
        if(m_decoded_data_queue.size() <= 0) {
            m_decoded_data_queue_mtx.unlock();
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
            continue;
        }

        DvppDataMemory* mem = m_decoded_data_queue.front();
        m_decoded_data_queue.pop();
        m_decoded_data_queue_mtx.unlock();

        if (mem) {
            if ((m_frameSkip == 1 || index % m_frameSkip == 0) && post_decoded_cbk){
                post_decoded_cbk(m_postDecArg, mem);
            } else {
                delete mem;
                mem = nullptr;
            }
        }

        index++;
        if(index >= 100000){
            index = 0;
        }
    }

    m_decoded_data_queue_mtx.lock();
    while (m_decoded_data_queue.size() > 0){
        DvppDataMemory* mem = m_decoded_data_queue.front();
        m_decoded_data_queue.pop();
        delete mem;
        mem = nullptr;
    }
    m_decoded_data_queue_mtx.unlock();
    
    LOG_INFO("[{}]- display_thread exit.", m_dec_name);
}

void DvppDecoder::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 DvppDecoder::doRecode(RecoderInfo& recoderInfo) {
    m_recoderManager.create_recode_task(recoderInfo);
}

void DvppDecoder::set_mq_callback(mq_callback_t cb) {
    m_recoderManager.set_mq_callback(cb);
}