aclStream流处理多路并发Pipeline框架中 视频解码 代码调用流程整理、类的层次关系整理、回调函数赋值和调用流程整理

目录1 类的整体层次关系2 各个类的初始化函数调用层次关系3 各个类的process函数调用层次关系4 详细代码阅读流程笔记这个就不画详细的流程图了只把自己代码阅读过程中的笔记复制到这里备忘。1 类的整体层次关系class FileHandler类里面有个FileHandlerImpl *impl_ nullptr成员 --class FileHandlerImpl --class FFParser 解封装的类 --class DeviceDecoder --class DecodeService --class IDecoder 只是个虚基类被用来继承的 --IDecoder *decoder_ CreateDecoder();这里面就是new DecoderAcl()了。 DecoderAcl继承IDecoder --再往下就是各种的硬件相关的类了类的从上到下关系是先是DecodeService类然后DecoderAcl类然后是AclLiteVideoProc类然后是VideoDecoder类然后再往下是VdecHelper类VdecHelper中做具体的处理。2 各个类的初始化函数调用层次关系FileHandlerImpl::Open() std::thread(FileHandlerImpl::Loop PrepareResources() parser_.Open impl_-Open(url, result, only_key_frame); result_-OnParserInfo(info); decoder_-Create(info, extra)这个decoder_就是DeviceDecoder类 VdecCreate(void **vdec, VdecCreateParams *params) 这是个单纯的函数不在任何类里面函数内容在下一行 infer_server::DecodeService::Instance().Create(vdec, params); DecodeService::Create里面有下面两行 IDecoder *decoder_ CreateDecoder()先创建DecoderAcl decoder_-Create(params)然后调用DecoderAcl 的create函数 再往下就是硬件相关的各种类的init和open函数了这次英伟达的也是要创建个新的类然后在这里新类的 open或者init函数被调用3 各个类的process函数调用层次关系process分两个方向一个是从上到下送解码数据的另一个是解码完之后获取解码数据然后从下往上回传的。先看从上到下送frame的流程bool FileHandlerImpl::Process() { parser_.Parse(); impl_-Parse(); result_-OnParserFrame(frame); DeviceDecoder::Process(VideoEsPacket *pkt) { int VdecSendStream(void *vdec, const VdecStream *stream, int timeout_ms) { return infer_server::DecodeService::Instance().SendStream(vdec, stream, timeout_ms); } decoder_-SendStream(stream, timeout_ms);这就已经是DecoderAcl的sendstream了已经到硬件了。 vdec_-Decode(data_ptr, data_size, frame_id, this);vdec_是AclLiteVideoProc 再往下就不看了下面几层类都是硬件解码的。然后看一下从下往上回传的就是解码之后的数据一层层传到上层的类。VideoDecoder::DvppVdecCallbackV2(hi_video_frame_info *frame, void *userdata) { CallBackVdec(const std::shared_ptracllite::ImageData decoded_image, uint32_t channel_id, uint32_t frame_id, v decoder-OnFrame(decoded_image, channel_id, frame_id);这个decoder就是DecoderAcl类 create_params_.OnFrame(surf, create_params_.userdata); OnFrame_(BufSurface *surf, void *userdata)class DeviceDecoder类 result_-OnDecodeFrame(wrapper); FileHandlerImpl::OnDecodeFrame4 详细代码阅读流程笔记VdecHelper这个类只有你发送了任务之后才会创建整个调用关系是这样的。void VideoTaskAdd(std::string const param, std::string reply)try {int ret session-start(g_source_name);if (ret ! 0) {return reply_error(reply, ERROR_UNKOWN, get_error_msg(ERROR_UNKOWN));}}intTaskSession::start(std::string data_source) {ret AddSourceForFile(source, stream_id, filename, FLAGS_src_frame_rate);intAddSourceForFile(cnstream::DataSource *source, const std::string stream_id, const std::string filename,const int frame_rate 25) {cnstream::FileSourceParam param;param.filename filename;param.framerate frame_rate;param.loop FLAGS_loop;auto handler cnstream::CreateSource(source, stream_id, param);--std::make_sharedFileHandler(module, stream_id, param);--FileHandler::FileHandler(DataSource *module, const std::string stream_id, const FileSourceParam param): SourceHandler(module, stream_id) {impl_ new (std::nothrow)FileHandlerImpl(module, param, this);explicitFileHandlerImpl(DataSource *module, const FileSourceParam param, FileHandler *handler): SourceRender(handler),module_(module),handle_param_(param),stream_id_(handler-GetStreamId()),parser_(stream_id_) {}returnsource-AddSource(handler);return 0;}SourceModule::AddSource(std::shared_ptrSourceHandler handler) {if (handler-Open() ! true) {LOGE(CORE) [ stream_id ]: stream Open failed;return -1;}FileHandler::Open()return impl_-Open();boolFileHandlerImpl::Open(){thread_ std::thread(FileHandlerImpl::Loop,this);void FileHandlerImpl::Loop() {if (!SetCurrentDevice(param_.device_id)) return;if (!PrepareResources()) {//这个Loop是个线程函数然后这里的prepare是准备往下调用了一系列的openClearResources();if (nullptr ! module_) {Event e;e.type EventType::EVENT_STREAM_ERROR;e.module_name module_-GetName();e.message Prepare codec resources failed.;e.stream_id stream_id_;e.thread_id std::this_thread::get_id();module_-PostEvent(e);}LOGE(SOURCE) [FileHandlerImpl] Loop(): [ stream_id_ ]: PrepareResources failed.;return;}set_thread_name(demux_decode);FrController controller(handle_param_.framerate);if (handle_param_.framerate 0) controller.Start();VLOG1(SOURCE) [FileHandlerImpl] Loop(): [ stream_id_ ]: DecoderLoop;while (running_.load()) {if (!Process()) {//这里的process里面是处理函数。break;}if (handle_param_.framerate 0) controller.Control();}VLOG1(SOURCE) [FileHandlerImpl] Loop(): [ stream_id_ ]: DecoderLoop Exit.;ClearResources();}boolFileHandlerImpl::PrepareResources(bool demux_only) {VLOG1(SOURCE) [FileHandlerImpl] PrepareResources(): [ stream_id_ ]: Begin preprare resources;int ret parser_.Open(handle_param_.filename, this, handle_param_.only_key_frame);VLOG1(SOURCE) [FileHandlerImpl] PrepareResources(): [ stream_id_ ]: Finish preprare resources;if (ret 0 || dec_create_failed_) {return false;}return true;}int FFParser::Open(const std::string url, IParserResult* result, bool only_key_frame) {if (impl_) {return impl_-Open(url, result, only_key_frame);}return -1;}intOpen(const std::string url, IParserResult* result, bool only_key_frame false) {std::unique_lockstd::mutex guard(mutex_);if (result_) {result_-OnParserInfo(info);}classFileHandlerImpl: publicIParserResult, publicIDecodeResult, publicSourceRender, publicIUserPool //FileHandlerImpl 继承的IParserResult。void FileHandlerImpl::OnParserInfo(VideoInfo *info) {bool ret decoder_-Create(info, extra);boolDeviceDecoder::Create(VideoInfo *info, ExtraDecoderInfo *extra) {int ret VdecCreate(vdec_, create_params);intVdecCreate(void **vdec, VdecCreateParams *params) {return infer_server::DecodeService::Instance().Create(vdec, params);}intCreate(void **vdec, VdecCreateParams *params) {if (decoder_-Create(params) 0) {LOG(ERROR) [InferServer] [DecodeService] Create(): Create decoder failed;delete decoder_;return -1;}intDecoderAcl::Create(VdecCreateParams *params) {if (vdec_ nullptr) {vdec_ std::unique_ptracllite::AclLiteVideoProc(new acllite::AclLiteVideoProc(vdec_config_, create_params_.device_id, create_params_.channel_id));}AclLiteVideoProc::AclLiteVideoProc(VideoDecodeConfig vdecConfig, int32_t deviceId, uint32_t channelId) {cap_ new VideoDecoder(vdecConfig, deviceId, channelId);Open();}AclLiteErrorAclLiteVideoProc::Open() {return cap_-Open();VideoDecoder::Open()AclLiteError VideoDecoder::Open()//这个函数里面先是设置device,context哪一些然后创建了类然后又调用了VdecHelper的init函数。dvppVdec_ newVdecHelper(deviceId_, channelId_, config_.width, config_.height, config_.resizedWidth,aclRet aclrtGetCurrentContext(context_);然后创建完类接着调用了init函数dvppVdec_-Init();AclLiteError VdecHelper::Init() {ACLLITE_LOG_INFO(Vdec process init start...);aclError aclRet aclrtCreateStream(stream_);if (aclRet ! ACL_SUCCESS) {ACLLITE_LOG_ERROR(Vdec create stream failed, errorno:%d, aclRet);return ACLLITE_ERROR_CREATE_STREAM;}ACLLITE_LOG_INFO(Vdec create stream ok);int ret pthread_create(subscribeThreadId_, nullptr,SubscribeReportThreadFunc, (void *)this);if (ret) {ACLLITE_LOG_ERROR(Start vdec subscribe thread failed, return:%d, ret);return ACLLITE_ERROR_CREATE_THREAD;}aclRet aclrtSubscribeReport(subscribeThreadId_, stream_);if (aclRet ! ACL_SUCCESS) {ACLLITE_LOG_ERROR(Vdec subscrible report failed, error %d, aclRet);return ACLLITE_ERROR_SUBSCRIBE_REPORT;}ret CreateVdecChannelDesc();if (ret ! ACLLITE_OK) {ACLLITE_LOG_ERROR(Create vdec channel failed);return ret;}return ACLLITE_OK;}上面基本上是void FileHandlerImpl::Loop()里面的if (!PrepareResources()) {//这个Loop是个线程函数然后这里的prepare是准备往下调用了一系列的open分支这个Loop函数里面其实还有if (!Process()) {分支。这个Process往下调用的流程是bool FileHandlerImpl::Process() {parser_.Parse();int FFParser::Parse() {if (impl_) {impl_-Parse();}return -1;}intParse() {result_-OnParserFrame(frame);void FileHandlerImpl::OnParserFrame(VideoEsFrame *frame) {if (decoder_ decoder_-Process(pkt) true) {bool DeviceDecoder::Process(VideoEsPacket *pkt) {int ret VdecSendStream(vdec_,stream, 1000);intVdecSendStream(void *vdec, const VdecStream *stream, int timeout_ms) {return infer_server::DecodeService::Instance().SendStream(vdec, stream, timeout_ms);}intSendStream(void *vdec, const VdecStream *stream, int timeout_ms) {if (!vdec || !stream) {LOG(ERROR) [InferServer] [DecodeService] SendStream(): Decoder or stream pointer is invalid;return -1;}IDecoder *decoder_ static_castIDecoder *(vdec);returndecoder_-SendStream(stream, timeout_ms);}intDecoderAcl::SendStream(const VdecStream *stream, int timeout_ms) {acllite::AclLiteError codec_ret vdec_-Decode(data_ptr, data_size, frame_id, this);AclLiteErrorAclLiteVideoProc::Decode(std::shared_ptruint8_t data, uint32_t dataSize, uint64_t frameId, void* userData) {if (cap_ ! nullptr) {return cap_-Decode(data, dataSize, frameId, userData);}else {return ACLLITE_ERROR_UNSURPPORT_VIDEO_CAPTURE;}}AclLiteErrorVideoDecoder::Decode(const std::shared_ptruint8_t data, uint32_t dataSize, uint64_t frameId, void* userData) {void* buffer CopyDataToDevice(data.get(), dataSize,UserData* tempUserData new UserData;tempUserData-decoderSelf (void*)this;tempUserData-userData userData;ret dvppVdec_-Process(videoFrame,tempUserData);AclLiteErrorVdecHelper::Process(shared_ptrFrameData frameData, void* userData) {// create input descAclLiteError atlRet CreateInputStreamDesc(frameData);if (atlRet ! ACLLITE_OK) {ACLLITE_LOG_ERROR(Create stream desc failed);return atlRet;}if (!frameData-isFinished) {// create out descatlRet CreateOutputPicDesc(outputPicSize_.load());if (atlRet ! ACLLITE_OK) {ACLLITE_LOG_ERROR(Create pic desc failed);return atlRet;}}else {outputPicDesc_ acldvppCreatePicDesc();if (outputPicDesc_ nullptr) {ACLLITE_LOG_ERROR(Create vdec output pic desc failed);return ACLLITE_ERROR_CREATE_PIC_DESC;}}// send data to dvpp vdec to decoderet aclvdecSendFrame(vdecChannelDesc_, inputStreamDesc_,outputPicDesc_, nullptr, userData);//最终就到了这里的aclvdecSendFrame了。然后这不就到了那个回调函数那里去了。if (ret ! ACL_SUCCESS) {ACLLITE_LOG_ERROR(Send frame to vdec failed, errorno:%d, ret);return ACLLITE_ERROR_VDEC_SEND_FRAME;}return ACLLITE_OK;}所以最终的其实都是在VdecHelper再往下这里sendframe之后又去哪里了首先去了回调函数void VideoDecoder::DvppVdecCallbackV2(hi_video_frame_info *frame, void *userdata) {然后进一步去了int VideoDecoder::DecodeCallback(const std::shared_ptrImageData decodedImage, uint32_t frameId, void* userData) {return config_.callbackFunc(decodedImage, channelId_, frameId, userData);}然后又去了config_.callbackFunc(decodedImage, channelId_, frameId, userData);这里其实就是static int CallBackVdec(const std::shared_ptracllite::ImageData decoded_image, uint32_t channel_id, uint32_t frame_id, void* user_data) {auto decoder reinterpret_castDecoderAcl *(user_data);decoder-OnFrame(decoded_image, channel_id, frame_id);return 0;}然后就是// IVDecResultvoidDecoderAcl::OnFrame(const std::shared_ptracllite::ImageData codec_image, uint32_t channel_id, uint32_t frame_id) {BufSurface *surf nullptr;if (create_params_.GetBufSurf(surf, codec_image-width, codec_image-height, CastColorFmt(codec_image-format),create_params_.surf_timeout_ms, create_params_.userdata) 0) {LOG(ERROR) [InferServer] [DecoderAcl] OnFrame(): Get BufSurface failed;OnError(-1);return;}if (surf-mem_type ! BUF_MEMORY_DVPP) {LOG(ERROR) [InferServer] [DecoderAcl] OnFrame(): BufSurface memory type must be BUF_MEMORY_DVPP;return;}switch (codec_image-format) {case acllite::ImageFormat::YUV_SP_420:case acllite::ImageFormat::YVU_SP_420:if (surf-surface_list[0].width ! codec_image-width || surf-surface_list[0].height ! codec_image-height) {BufSurface transform_src;BufSurfaceParams src_param;memset(transform_src, 0, sizeof(BufSurface));memset(src_param, 0, sizeof(BufSurfaceParams));src_param.color_format CastColorFmt(codec_image-format);src_param.data_size codec_image-size;src_param.data_ptr reinterpret_castvoid *(codec_image-data.get());VLOG(5) [InferServer] [DecoderAcl] OnFrame(): codec_frame: width codec_image-width , height codec_image-height , width stride codec_image-alignWidth , height stride codec_image-alignHeight;VLOG(5) [InferServer] [DecoderAcl] OnFrame(): surf-surface_list[0]: width surf-surface_list[0].width , height surf-surface_list[0].height , width stride surf-surface_list[0].width_stride , height stride surf-surface_list[0].height_stride;src_param.width codec_image-width;src_param.height codec_image-height;src_param.width_stride codec_image-alignWidth;src_param.height_stride codec_image-alignHeight;transform_src.batch_size 1;transform_src.num_filled 1;transform_src.device_id create_params_.device_id;transform_src.mem_type BUF_MEMORY_DVPP;transform_src.surface_list src_param;TransformParams trans_params;memset(trans_params, 0, sizeof(trans_params));trans_params.transform_flag TRANSFORM_RESIZE_SRC;if (Transform(transformer_, transform_src, surf, trans_params) 0) {LOG(ERROR) [InferServer] [DecoderAcl] OnFrame(): Transfrom failed;break;}}else {CALL_ACL_FUNC(acllite::CopyDataToHostEx(surf-surface_list[0].data_ptr, codec_image-size, codec_image-data.get(), codec_image-size, codec_image-deviceId), [DecoderAcl] OnFrame(): copy codec buffer data to surf failed);}break;default:break;}surf-pts frame_id;create_params_.OnFrame(surf, create_params_.userdata);}然后到了class DeviceDecoder类的static int OnFrame_(BufSurface *surf, void *userdata) {DeviceDecoder *thiz reinterpret_castDeviceDecoder *(userdata);return thiz-OnFrame(surf);}然后到了intDeviceDecoder::OnFrame(BufSurface *surf){surf-surface_list[0].width - surf-surface_list[0].width 1;surf-surface_list[0].height - surf-surface_list[0].height 1;BufSurfWrapperPtr wrapper std::make_sharedBufSurfaceWrapper(surf);if (result_) {result_-OnDecodeFrame(wrapper);return 0;}return -1;}然后到了voidFileHandlerImpl::OnDecodeFrame(BufSurfWrapperPtr wrapper){if (frame_count_ % param_.interval ! 0) {// LOGI(SOURCE) frames are discarded frame_count_;return; // discard frames}std::shared_ptrCNFrameInfo data this-CreateFrameInfo();if (!data) {LOGW(SOURCE) [FileHandlerImpl] OnDecodeFrame(): failed to create FrameInfo.;return;}data-timestamp wrapper-GetPts();if (!wrapper-GetBufSurface()) {data-flags static_castsize_t(CNFrameFlag::CN_FRAME_FLAG_INVALID);this-SendFrameInfo(data);return;}#if 0BufSurface* src_buf wrapper-GetBufSurface();std::vectoruint8_t indata(src_buf-surface_list[0].data_size);MemcpyHD(indata.data(), BUF_MEMORY_NORMAL, src_buf-surface_list[0].data_ptr, BUF_MEMORY_NORMAL, src_buf-surface_list[0].data_size);cv::Mat yuv_mat(src_buf-surface_list[0].height * 3 / 2, src_buf-surface_list[0].width, CV_8UC1, indata.data());cv::Mat bgr_mat;cv::cvtColor(yuv_mat, bgr_mat, cv::COLOR_YUV2BGR_NV12);cv::imwrite(decoded.jpg, bgr_mat);#endifint ret SourceRender::Process(data, std::move(wrapper), frame_id_, param_);if (ret 0) {LOGE(SOURCE) [FileHandlerImpl] OnDecodeFrame(): [ stream_id_ ]: Render frame failed;return;}this-SendFrameInfo(data);}然后是bool SendFrameInfo(std::shared_ptrCNFrameInfo data) { return handler_-SendData(data); }bool SendData(std::shared_ptrCNFrameInfo data) {if (this-module_) {returnthis-module_-SendData(data);}return false;}bool Module::TransmitData(std::shared_ptrCNFrameInfo data) {if (!HasTransmit()) {return true;}if (!DoTransmitData(data)) {return true;}return false;}