Hu Chunming / vpt_ascend_arm

#include "DvppStreamDecoder.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;
	DvppStreamDecoder* self;

    Vdec_CallBack_UserData() {
        frameId = 0;
        frame_nb = 0;
    }
};

static void *ReportThd(void *arg)
{
    DvppStreamDecoder *self = (DvppStreamDecoder *)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){
        DvppStreamDecoder* self = userData->self;
        if(self != nullptr){
            self->doVdppVdecCallBack(input, output, userData);
        }
        delete userData;
	    userData = nullptr;
    }
}

DvppStreamDecoder::DvppStreamDecoder(/* args */)
{
}

DvppStreamDecoder::~DvppStreamDecoder()
{
    Close();

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

bool DvppStreamDecoder::Init(FFDecConfig cfg) {

    m_dec_name = cfg.dec_name;

    LOG_INFO("[{}]- Init device start...", m_dec_name);

    m_deviceId = atoi(cfg.gpuid.c_str());

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

        ret = aclrtCreateContext(&m_context, m_deviceId);
        if (ret != ACL_ERROR_NONE) {
            LOG_ERROR("[{}]-aclrtCreateContext failed !", m_dec_name);
            return false;
        }

        // DvppSourceManager 创建时包含 aclInit，析构时包含 aclFinalize
        DvppSourceManager* pSrcMgr = DvppSourceManager::getInstance();
        m_dvpp_channel = pSrcMgr->getChannel(m_deviceId);
        if(m_dvpp_channel < 0){
            LOG_ERROR("[{}]-该设备channel已经用完了!", m_dec_name);
            return false;
        }

        m_bResize = cfg.resize;

        m_vpcUtils.init(m_deviceId);

        decode_finished_cbk = cfg.decode_finished_cbk;

        LOG_INFO("[{}]- init vdpp success! device:{} channel:{}", m_dec_name, m_deviceId, m_dvpp_channel);
        return true;
    }while(0);

    release_dvpp();

    return false;
}

void DvppStreamDecoder::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_deviceId, m_dvpp_channel);
        m_dvpp_channel = -1;
    }
}

int DvppStreamDecoder::getVdecType(int videoType)
{
    int streamFormat = H264_MAIN_LEVEL;

    // VDEC only support　H265 main level，264 baseline level，main level，high level
    if (videoType == 1) {
        streamFormat = H265_MAIN_LEVEL;
    } else if (videoType == 0) {
        streamFormat = H264_MAIN_LEVEL;
    } else {
        streamFormat = -1;
        LOG_ERROR("Not support stream, type {}", videoType);
    }

    return streamFormat;
}

void DvppStreamDecoder::doProcessReport(){

    aclError ret = aclrtSetDevice(m_deviceId);
    if(ret != ACL_ERROR_NONE){
        // cout << "aclrtSetDevice failed" << endl;
        LOG_ERROR("aclrtSetDevice failed !");
        return ;
    }

    aclrtContext ctx;
    ret = aclrtCreateContext(&ctx, m_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 DvppStreamDecoder::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)) {
                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;
                
                mem = m_vpcUtils.resize(output, outWidth, outHeight);
                if (mem)  {
                    acldvppFree(outputDataDev);
                    outputDataDev = nullptr;

                    mem->setDeviceId(to_string(m_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_deviceId), false, frame_nb, (unsigned char *)outputDataDev);
                out_frame_width = width;
                out_frame_height = height;
            }
            
            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");
}

DvppDataMemory* DvppStreamDecoder::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;
}

bool DvppStreamDecoder::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;
}

int DvppStreamDecoder::sendPkt(aclvdecChannelDesc *vdecChannelDesc, AVPacket* pkt, unsigned long long frame_nb, int vdec_out_size) {

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

int DvppStreamDecoder::SendData(int videoType, char* data, int len, int isKey, uint64_t pts) {

    if (m_bExit) {
        return -1;
    }

    if (m_DvppCacheCounter.load() > 20) {
        // 解码器解码不过来。实时流在此处的处理会导致花屏，这是由于解码器性能问题导致，无法避免
        // 实时流在这里处理是为了避免长时间不读取数据导致数据中断
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        return -3;
    }
    
    int ret = aclrtSetCurrentContext(m_context);
    if(ret != ACL_ERROR_NONE){
        LOG_ERROR("[{}]- aclrtSetCurrentContext failed", m_dec_name);
        return -2;
    }

    if (vdecChannelDesc == nullptr) {
        vdecChannelDesc = aclvdecCreateChannelDesc();
        if (vdecChannelDesc == nullptr) { 
            LOG_ERROR("[{}]- aclvdecCreateChannelDesc failed", m_dec_name);
            return -2;
        }

        do
        {
            ret = -2;

             ret = pthread_create(&report_thread, nullptr, ReportThd, (void *)this);
            if(ret != 0){
                LOG_ERROR("[{}]- pthread_create failed", m_dec_name);
                break;
            }
            
            int  iType = getVdecType(videoType);
            if(-1 == iType) {
                break;
            }
            acldvppStreamFormat  enType = static_cast<acldvppStreamFormat>(iType);
            
            // 通道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, enType), "aclvdecSetChannelDescEnType failed");
            CHECK_AND_BREAK(aclvdecSetChannelDescOutPicFormat(vdecChannelDesc, PIXEL_FORMAT_YUV_SEMIPLANAR_420), "aclvdecSetChannelDescOutPicFormat failed");
            CHECK_AND_BREAK(aclvdecCreateChannel(vdecChannelDesc), "aclvdecCreateChannel failed");

            ret = 0;
        } while (0);

        if (ret < 0) {
            release_channeldesc();
            LOG_ERROR("[{}]- Set Channel Desc failed", m_dec_name);
            return ret;
        }
    }

    AVPacket* pkt = av_packet_alloc();
    av_init_packet(pkt);

    pkt->size = len;
    pkt->data = (uint8_t*)data;

    if (isKey) {
        pkt->flags = 1;
    } else {
        pkt->flags = 0;
    }

    do
    {
        int vdec_out_size = parse_stream_info(videoType, pkt);
        if (vdec_out_size <= 0) {
            ret = -4;
            break;
        }
        
        if (vdecChannelDesc)
        {
            ret = av_bsf_send_packet(h264bsfc, pkt);
            if(ret < 0) {
                LOG_ERROR("[{}]- av_bsf_send_packet error!", m_dec_name);
                ret = -3;
                break;
            }

            int nSended = -1;
            while ((ret = av_bsf_receive_packet(h264bsfc, pkt)) == 0) {
                if(m_bExit){
                    break;
                }

                m_frame_nb++;
                // dvpp 解码
                nSended = sendPkt(vdecChannelDesc, pkt, m_frame_nb, vdec_out_size);
            }

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

            if(nSended < 0) {
                // 执行出错，强行结束整个任务
                ret = -2;
                break;
            }
#ifdef USE_VILLAGE
            //直接返回，不释放pkt，pkt在recoderManager释放
            return 0;
#elif
            ret = 0;
#endif
        }
    } while (0);

    av_packet_free(&pkt);
    pkt = nullptr;
    
    return ret;
}


int DvppStreamDecoder::parse_stream_info(int videoType, AVPacket* pkt) {
    if (m_vdec_out_size > 0) {
        return m_vdec_out_size;
    }

    m_vdec_out_size = -1;
    
    AVCodecContext* avctx = nullptr;
    const AVCodec* pAVCodec = nullptr;

    try
    {
        if (0 == videoType) {
            pAVCodec = avcodec_find_decoder(AV_CODEC_ID_H264);
            LOG_INFO("m_avCodecName is H264");		
        } else if (1 == videoType) {
            pAVCodec = avcodec_find_decoder(AV_CODEC_ID_H265);
            LOG_INFO("m_avCodecName is H265");
        } else{
            LOG_INFO("m_avCodecName is unknown, videoType is {}", videoType);
        }

        if (!pAVCodec) {
            LOG_ERROR("frameCallback frame decode error, ERROR_DECODER_NOT_FOUND");
            throw -2;
        }

        avctx = avcodec_alloc_context3(pAVCodec);

        if (avcodec_open2(avctx, pAVCodec, nullptr) < 0) {
            LOG_ERROR("avcodec_open2 failed!");
            throw -2;
        }

        //开始解码
        int ret = avcodec_send_packet(avctx, pkt);
        if (ret < 0) {
            LOG_ERROR("Real stream视频解码失败，请检查视频设备{}: avcodec_send_packet failed. ret={}", m_dec_name, ret);
            throw -3;
        }
    
        if (frameW < 1) {
            frameW = avctx->width;
            frameH = avctx->height;
            if (frameW <= 0 || frameH <= 0) {
                LOG_ERROR("[{}] frame W or H is error! ({},{})", m_dec_name, frameW, frameH);
                throw -1;
            }

            const AVBitStreamFilter * filter = nullptr;
            if(0 == videoType){
                filter = av_bsf_get_by_name("h264_mp4toannexb");
            }else if(1 == videoType){
                filter = av_bsf_get_by_name("hevc_mp4toannexb");
            }else {
                LOG_ERROR("[{}]- codec_id is not supported!", m_dec_name);
                throw -4;
            }

            int ret = av_bsf_alloc(filter, &h264bsfc);
            if (ret < 0){
                LOG_ERROR("av_bsf_alloc failed!");
                throw -2;
            }
            
            avcodec_parameters_from_context(h264bsfc->par_in, avctx);
            av_bsf_init(h264bsfc);
        }

        if (avctx->framerate.den) {
            m_fps = av_q2d(avctx->framerate);
        } else {
            m_fps = 0.0;
        }

        AVFrame* frame = av_frame_alloc();
        do
        {
            ret = avcodec_receive_frame(avctx, frame);
            if ((ret == AVERROR(EAGAIN) || ret == AVERROR_EOF) || ret < 0){
                LOG_ERROR("{} - Failed to receive frame: {}", m_dec_name, ret);
                break;
            }
            
            if (frame->width != frameW || frame->height != frameH){
                LOG_ERROR("AVFrame is inconsistent: width is {}, height is {}; original frameW is {}, frameH is {}--{}", frame->width, frame->height, frameW, frameH , m_dec_name);
                break;
            }

            m_vdec_out_size = frame->width * frame->height * 3 / 2;
        } while (0);

#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

        av_frame_free(&frame); 
        frame = nullptr;

    }
    catch(const int& iError)  {
        m_vdec_out_size = iError;
    } catch(...) {
        m_vdec_out_size = -1;
    }

    if(avctx){
        avcodec_free_context(&avctx);
        avctx = nullptr;
    }
    
    return m_vdec_out_size;
}

void DvppStreamDecoder::release_channeldesc() {
    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");
    }
}

void DvppStreamDecoder::Close() {
    m_bExit = true;

    if (m_context) {
        int ret = aclrtSetCurrentContext(m_context);
        if(ret != ACL_ERROR_NONE){
            LOG_ERROR("[{}]- aclrtSetCurrentContext failed", m_dec_name);
            return;
        }

        release_channeldesc();
        release_dvpp();
    }

    if(h264bsfc){
		av_bsf_free(&h264bsfc);
		h264bsfc = nullptr;
	}

    m_recoderManager.close();

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

void DvppStreamDecoder::doRecode(RecoderInfo& recoderInfo) {
    m_recoderManager.create_recode_task(recoderInfo);
}

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