From 274a2d36d09e3ca5ec79c3b929dcf8b476d17b3e Mon Sep 17 00:00:00 2001 From: sivanov-work Date: Tue, 17 Aug 2021 17:00:47 +0300 Subject: [PATCH 1/8] Add perf test --- .../gapi_streaming_source_perf_tests.cpp | 79 +++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp diff --git a/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp new file mode 100644 index 000000000000..f8d3c421b31d --- /dev/null +++ b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp @@ -0,0 +1,79 @@ +// This file is part of OpenCV project. +// It is subject to the license terms in the LICENSE file found in the top-level directory +// of this distribution and at http://opencv.org/license.html. +// +// Copyright (C) 2021 Intel Corporation + + +#include "../perf_precomp.hpp" +#include "../../test/common/gapi_tests_common.hpp" +#include +#include + +namespace opencv_test +{ +using namespace perf; + +const std::string files[] = { + "highgui/video/big_buck_bunny.avi", + "highgui/video/big_buck_bunny.mov", + "highgui/video/big_buck_bunny.mp4", + "highgui/video/big_buck_bunny.wmv" +}; + +const std::string codec[] = { + "MFX_CODEC_HEVC", + "MFX_CODEC_AVC" +}; + +using source_t = std::string; +using codec_t = std::string; +using source_description_t = std::tuple; + +class OneVPLSourcePerfTest : public TestPerfParams {}; +class VideoCapSourcePerfTest : public TestPerfParams {}; + +PERF_TEST_P_(OneVPLSourcePerfTest, TestPerformance) +{ + using namespace cv::gapi::wip; + + const auto params = GetParam(); + const source_t& src = get<0>(params); + codec_t type = get<1>(params); + + std::vector cfg_params { + oneVPL_cfg_param::create("mfxImplDescription.Impl", "MFX_IMPL_TYPE_HARDWARE"), + oneVPL_cfg_param::create("mfxImplDescription.mfxDecoderDescription.decoder.CodecID", type), + }; + + auto source_ptr = make_vpl_src(src, cfg_params); + Data out; + TEST_CYCLE() + { + source_ptr->pull(out); + } + + SANITY_CHECK_NOTHING(); +} + +PERF_TEST_P_(VideoCapSourcePerfTest, TestPerformance) +{ + using namespace cv::gapi::wip; + + const source_t& src = GetParam(); + auto source_ptr = make_src(src); + Data out; + TEST_CYCLE() + { + source_ptr->pull(out); + } + + SANITY_CHECK_NOTHING(); +} + +INSTANTIATE_TEST_CASE_P(OneVPLSourcePerfTest, OneVPLSourcePerfTest, + Values(source_description_t(files[0], codec[0]))); + +INSTANTIATE_TEST_CASE_P(VideoCapSourcePerfTest, VideoCapSourcePerfTest, + Values(files[0])); +} // namespace opencv_test From c0119af4a660d93ce09f4a83417f53b593f9bfa0 Mon Sep 17 00:00:00 2001 From: sivanov-work Date: Wed, 18 Aug 2021 14:39:10 +0300 Subject: [PATCH 2/8] Correct path to video --- .../gapi_streaming_source_perf_tests.cpp | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp index f8d3c421b31d..b4b141523306 100644 --- a/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp +++ b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp @@ -15,10 +15,8 @@ namespace opencv_test using namespace perf; const std::string files[] = { - "highgui/video/big_buck_bunny.avi", - "highgui/video/big_buck_bunny.mov", - "highgui/video/big_buck_bunny.mp4", - "highgui/video/big_buck_bunny.wmv" + "highgui/video/big_buck_bunny.h265", + "highgui/video/big_buck_bunny.h264", }; const std::string codec[] = { @@ -38,7 +36,7 @@ PERF_TEST_P_(OneVPLSourcePerfTest, TestPerformance) using namespace cv::gapi::wip; const auto params = GetParam(); - const source_t& src = get<0>(params); + source_t src = findDataFile(get<0>(params)); codec_t type = get<1>(params); std::vector cfg_params { @@ -60,7 +58,7 @@ PERF_TEST_P_(VideoCapSourcePerfTest, TestPerformance) { using namespace cv::gapi::wip; - const source_t& src = GetParam(); + source_t src = findDataFile(GetParam()); auto source_ptr = make_src(src); Data out; TEST_CYCLE() @@ -71,9 +69,11 @@ PERF_TEST_P_(VideoCapSourcePerfTest, TestPerformance) SANITY_CHECK_NOTHING(); } -INSTANTIATE_TEST_CASE_P(OneVPLSourcePerfTest, OneVPLSourcePerfTest, - Values(source_description_t(files[0], codec[0]))); +INSTANTIATE_TEST_CASE_P(Streaming, OneVPLSourcePerfTest, + Values(source_description_t(files[0], codec[0]), + source_description_t(files[1], codec[1]))); -INSTANTIATE_TEST_CASE_P(VideoCapSourcePerfTest, VideoCapSourcePerfTest, - Values(files[0])); +INSTANTIATE_TEST_CASE_P(Streaming, VideoCapSourcePerfTest, + Values(files[0], + files[1])); } // namespace opencv_test From 00f7e68487e8163735dfc7196cc4906b9c9da723 Mon Sep 17 00:00:00 2001 From: sivanov-work Date: Wed, 18 Aug 2021 17:56:49 +0300 Subject: [PATCH 3/8] Improve performance --- .../engine/decode/decode_engine_legacy.cpp | 16 +++++++++++++++- .../engine/decode/decode_engine_legacy.hpp | 2 ++ .../gapi/src/streaming/onevpl/onevpl_utils.cpp | 2 ++ 3 files changed, 19 insertions(+), 1 deletion(-) diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp index 2b8e33cf76b7..e73a1749b902 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp @@ -138,6 +138,13 @@ VPLLegacyDecodeEngine::VPLLegacyDecodeEngine(std::unique_ptrget_handle(), &my_sess.output_surface_ptr, &my_sess.sync); + + if (!pending_frames.empty()) + { + auto data = pending_frames.front(); + pending_frames.pop(); + ready_frames.push(std::move(data)); + } return ExecutionStatus::Continue; }, // 3) Wait for ASYNC decode result @@ -227,7 +234,7 @@ void VPLLegacyDecodeEngine::on_frame_ready(LegacyDecodeSession& sess) // manage memory ownership rely on acceleration policy auto frame_adapter = acceleration_policy->create_frame_adapter(sess.decoder_pool_id, sess.output_surface_ptr); - ready_frames.emplace(cv::MediaFrame(std::move(frame_adapter)), sess.generate_frame_meta()); + pending_frames.emplace(cv::MediaFrame(std::move(frame_adapter)), sess.generate_frame_meta()); } ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngine::process_error(mfxStatus status, LegacyDecodeSession& sess) @@ -236,6 +243,13 @@ ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngine::process_error(mfxSt switch (status) { case MFX_ERR_NONE: + try { + sess.swap_surface(*this); + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << + "Abort"); + ExecutionStatus::Failed; + } return ExecutionStatus::Continue; case MFX_ERR_MORE_DATA: // The function requires more bitstream at input before decoding can proceed if (!sess.data_provider || sess.data_provider->empty()) { diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp index db4a2519ff41..66263be31cf4 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp @@ -39,6 +39,8 @@ class VPLLegacyDecodeEngine : public ProcessingEngineBase { ExecutionStatus process_error(mfxStatus status, LegacyDecodeSession& sess); void on_frame_ready(LegacyDecodeSession& sess); + + frames_container_t pending_frames; }; } // namespace wip } // namespace gapi diff --git a/modules/gapi/src/streaming/onevpl/onevpl_utils.cpp b/modules/gapi/src/streaming/onevpl/onevpl_utils.cpp index 9a7212a3a80d..7ecc3f5abb84 100644 --- a/modules/gapi/src/streaming/onevpl/onevpl_utils.cpp +++ b/modules/gapi/src/streaming/onevpl/onevpl_utils.cpp @@ -386,6 +386,8 @@ std::string mfxstatus_to_string(mfxStatus err) { return "MFX_WRN_DEVICE_BUSY"; case MFX_WRN_VIDEO_PARAM_CHANGED: return "MFX_WRN_VIDEO_PARAM_CHANGED"; + case MFX_WRN_IN_EXECUTION: + return "MFX_WRN_IN_EXECUTION"; default: From 84b8c4b3d717c6cea93150b9248ec1093b37df74 Mon Sep 17 00:00:00 2001 From: sivanov-work Date: Wed, 18 Aug 2021 18:19:48 +0300 Subject: [PATCH 4/8] Improve perf: add async pipeline --- modules/gapi/CMakeLists.txt | 2 + .../engine/decode/decode_engine_async.cpp | 348 ++++++++++++++++++ .../engine/decode/decode_engine_async.hpp | 47 +++ .../engine/decode/decode_session_async.cpp | 81 ++++ .../engine/decode/decode_session_async.hpp | 58 +++ .../streaming/onevpl/onevpl_source_priv.cpp | 4 +- 6 files changed, 539 insertions(+), 1 deletion(-) create mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp create mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp create mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp create mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp diff --git a/modules/gapi/CMakeLists.txt b/modules/gapi/CMakeLists.txt index a5492c42bb4a..0d69e78d9838 100644 --- a/modules/gapi/CMakeLists.txt +++ b/modules/gapi/CMakeLists.txt @@ -186,6 +186,8 @@ if(HAVE_GAPI_ONEVPL) src/streaming/onevpl/engine/processing_engine_base.cpp src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp src/streaming/onevpl/engine/decode/decode_session.cpp + src/streaming/onevpl/engine/decode/decode_engine_async.cpp + src/streaming/onevpl/engine/decode/decode_session_async.cpp src/streaming/onevpl/onevpl_utils.cpp src/streaming/onevpl/onevpl_cfg_params_parser.cpp ) diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp new file mode 100644 index 000000000000..9234cc77eddc --- /dev/null +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp @@ -0,0 +1,348 @@ +// This file is part of OpenCV project. +// It is subject to the license terms in the LICENSE file found in the top-level directory +// of this distribution and at http://opencv.org/license.html. +// +// Copyright (C) 2021 Intel Corporation + +#include +#include + +#include "streaming/onevpl/engine/decode/decode_engine_async.hpp" +#include "streaming/onevpl/engine/decode/decode_session_async.hpp" +#include "streaming/onevpl/accelerators/accel_policy_interface.hpp" +#include "streaming/onevpl/accelerators/surface/surface.hpp" +#include "streaming/onevpl/onevpl_utils.hpp" +//TODO #include "streaming/vpl/vpl_utils.hpp" +#include "logger.hpp" + + +namespace cv { +namespace gapi { +namespace wip { +/* UTILS */ +mfxU32 GetSurfaceSize_(mfxU32 FourCC, mfxU32 width, mfxU32 height) { + mfxU32 nbytes = 0; + + mfxU32 half_width = width / 2; + mfxU32 half_height = height / 2; + switch (FourCC) { + case MFX_FOURCC_I420: + case MFX_FOURCC_NV12: + nbytes = width * height + 2 * half_width * half_height; + break; + case MFX_FOURCC_I010: + case MFX_FOURCC_P010: + nbytes = width * height + 2 * half_width * half_height; + nbytes *= 2; + break; + case MFX_FOURCC_RGB4: + nbytes = width * height * 4; + break; + default: + break; + } + + return nbytes; +} + +surface_ptr_t create_surface_RGB4_(mfxFrameInfo frameInfo, + std::shared_ptr out_buf_ptr, + size_t out_buf_ptr_offset, + size_t out_buf_size) +{ + mfxU8* buf = reinterpret_cast(out_buf_ptr.get()); + mfxU16 surfW = frameInfo.Width * 4; + mfxU16 surfH = frameInfo.Height; + (void)surfH; + + // TODO more intelligent check + if (out_buf_size <= out_buf_ptr_offset) { + GAPI_LOG_WARNING(nullptr, "Not enough buffer, ptr: " << out_buf_ptr << + ", size: " << out_buf_size << + ", offset: " << out_buf_ptr_offset << + ", W: " << surfW << + ", H: " << surfH); + GAPI_Assert(false && "Invalid offset"); + } + + std::unique_ptr handle(new mfxFrameSurface1); + memset(handle.get(), 0, sizeof(mfxFrameSurface1)); + + handle->Info = frameInfo; + handle->Data.B = buf + out_buf_ptr_offset; + handle->Data.G = handle->Data.B + 1; + handle->Data.R = handle->Data.B + 2; + handle->Data.A = handle->Data.B + 3; + handle->Data.Pitch = surfW; + + return Surface::create_surface(std::move(handle), out_buf_ptr); +} + +surface_ptr_t create_surface_other_(mfxFrameInfo frameInfo, + std::shared_ptr out_buf_ptr, + size_t out_buf_ptr_offset, + size_t out_buf_size) +{ + mfxU8* buf = reinterpret_cast(out_buf_ptr.get()); + mfxU16 surfH = frameInfo.Height; + mfxU16 surfW = (frameInfo.FourCC == MFX_FOURCC_P010) ? frameInfo.Width * 2 : frameInfo.Width; + + // TODO more intelligent check + if (out_buf_size <= + out_buf_ptr_offset + (surfW * surfH) + ((surfW / 2) * (surfH / 2))) { + GAPI_LOG_WARNING(nullptr, "Not enough buffer, ptr: " << out_buf_ptr << + ", size: " << out_buf_size << + ", offset: " << out_buf_ptr_offset << + ", W: " << surfW << + ", H: " << surfH); + GAPI_Assert(false && "Invalid offset"); + } + + std::unique_ptr handle(new mfxFrameSurface1); + memset(handle.get(), 0, sizeof(mfxFrameSurface1)); + + handle->Info = frameInfo; + handle->Data.Y = buf + out_buf_ptr_offset; + handle->Data.U = buf + out_buf_ptr_offset + (surfW * surfH); + handle->Data.V = handle->Data.U + ((surfW / 2) * (surfH / 2)); + handle->Data.Pitch = surfW; + + return Surface::create_surface(std::move(handle), out_buf_ptr); +} + +VPLLegacyDecodeEngineAsync::VPLLegacyDecodeEngineAsync(std::unique_ptr&& accel) + : ProcessingEngineBase(std::move(accel)) { + + GAPI_LOG_INFO(nullptr, "Create Legacy Decode Engine"); + create_pipeline( + // 1) Reade File + [this] (EngineSession& sess) -> ExecutionStatus + { + LegacyDecodeSessionAsync &my_sess = static_cast(sess); + my_sess.last_status = ReadEncodedStream(my_sess.stream, my_sess.data_provider); + if (my_sess.last_status != MFX_ERR_NONE) { + my_sess.data_provider.reset(); //close source + } + return ExecutionStatus::Continue; + }, + // 2) enqueue ASYNC decode + [this] (EngineSession& sess) -> ExecutionStatus + { + LegacyDecodeSessionAsync &my_sess = static_cast(sess); + + my_sess.last_status = + MFXVideoDECODE_DecodeFrameAsync(my_sess.session, + my_sess.last_status == MFX_ERR_NONE + ? &my_sess.stream + : nullptr, /* No more data to read, start decode draining mode*/ + my_sess.procesing_surface_ptr.lock()->get_handle(), + + &my_sess.sync_queue.back().second, + &my_sess.sync_queue.back().first); + + while (my_sess.last_status == MFX_ERR_MORE_SURFACE) { + try { + my_sess.swap_surface(*this, false); + my_sess.last_status = + MFXVideoDECODE_DecodeFrameAsync(my_sess.session, + &my_sess.stream, + my_sess.procesing_surface_ptr.lock()->get_handle(), + + &my_sess.sync_queue.back().second, + &my_sess.sync_queue.back().first); + + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << my_sess.session << "] error: " << ex.what() << + "Abort"); + } + } + return ExecutionStatus::Continue; + }, + // 3) Wait for ASYNC decode result + [this] (EngineSession& sess) -> ExecutionStatus + { + LegacyDecodeSessionAsync& my_sess = static_cast(sess); + if (!my_sess.sync_queue.empty()/*sess.last_status == MFX_ERR_NONE*/) // Got 1 decoded frame + { + GAPI_LOG_DEBUG(nullptr, "my_sess.sync_queue size: " << my_sess.sync_queue.size() << + "my_sess.sync_queue.front().first: " << + my_sess.sync_queue.front().first); + sess.last_status = MFXVideoCORE_SyncOperation(sess.session, my_sess.sync_queue.front().first, 0); + /*GAPI_LOG_WARNING(nullptr, "my_sess.sync_queue size: " << my_sess.sync_queue.size() << + "my_sess.sync_queue.front().first: " << + my_sess.sync_queue.front().first << + "status: " << mfxstatus_to_string(sess.last_status));*/ + if (MFX_ERR_NONE == sess.last_status) { + on_frame_ready(my_sess, my_sess.sync_queue.front().second); + } + } + return ExecutionStatus::Continue; + }, + // 4) Falls back on generic status procesing + [this] (EngineSession& sess) -> ExecutionStatus + { + return this->process_error(sess.last_status, static_cast(sess)); + } + ); +} + +void VPLLegacyDecodeEngineAsync::initialize_session(mfxSession mfx_session, + DecoderParams&& decoder_param, + std::shared_ptr provider) +{ + mfxFrameAllocRequest decRequest = {}; + // Query number required surfaces for decoder + MFXVideoDECODE_QueryIOSurf(mfx_session, &decoder_param.param, &decRequest); + + // External (application) allocation of decode surfaces + GAPI_LOG_DEBUG(nullptr, "Query IOSurf for session: " << mfx_session << + ", mfxFrameAllocRequest.NumFrameSuggested: " << decRequest.NumFrameSuggested << + ", mfxFrameAllocRequest.Type: " << decRequest.Type); + + mfxU32 singleSurfaceSize = GetSurfaceSize_(decoder_param.param.mfx.FrameInfo.FourCC, + decoder_param.param.mfx.FrameInfo.Width, + decoder_param.param.mfx.FrameInfo.Height); + if (!singleSurfaceSize) { + throw std::runtime_error("Cannot determine surface size for: fourCC" + + std::to_string(decoder_param.param.mfx.FrameInfo.FourCC) + + ", width: " + std::to_string(decoder_param.param.mfx.FrameInfo.Width) + + ", height: " + std::to_string(decoder_param.param.mfx.FrameInfo.Height)); + } + + const auto &frameInfo = decoder_param.param.mfx.FrameInfo; + auto surface_creator = + [&frameInfo] (std::shared_ptr out_buf_ptr, size_t out_buf_ptr_offset, + size_t out_buf_size) -> surface_ptr_t { + return (frameInfo.FourCC == MFX_FOURCC_RGB4) ? + create_surface_RGB4_(frameInfo, out_buf_ptr, out_buf_ptr_offset, + out_buf_size) : + create_surface_other_(frameInfo, out_buf_ptr, out_buf_ptr_offset, + out_buf_size);}; + + //TODO Configure preallocation size (how many frames we can hold) + const size_t preallocated_frames_count = 30; + VPLAccelerationPolicy::pool_key_t decode_pool_key = + acceleration_policy->create_surface_pool(decRequest.NumFrameSuggested * preallocated_frames_count, + singleSurfaceSize, + surface_creator); + + // create session + std::shared_ptr sess_ptr = + register_session(mfx_session, + std::move(decoder_param), + provider); + + sess_ptr->init_surface_pool(decode_pool_key); + // prepare working decode surface + sess_ptr->swap_surface(*this); +} + +ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngineAsync::execute_op(operation_t& op, EngineSession& sess) { + return op(sess); +} + +void VPLLegacyDecodeEngineAsync::on_frame_ready(LegacyDecodeSessionAsync& sess, mfxFrameSurface1* ready_surface) +{ + GAPI_LOG_DEBUG(nullptr, "[" << sess.session << "], frame ready"); + + // manage memory ownership rely on acceleration policy + auto frame_adapter = acceleration_policy->create_frame_adapter(sess.decoder_pool_id, + ready_surface); + ready_frames.emplace(cv::MediaFrame(std::move(frame_adapter)), sess.generate_frame_meta()); + + // pop ready sync onject + sess.sync_queue.pop(); +} + +ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngineAsync::process_error(mfxStatus status, LegacyDecodeSessionAsync& sess) +{ + GAPI_LOG_DEBUG(nullptr, "status: " << mfxstatus_to_string(status)); + + switch (status) { + case MFX_ERR_NONE: + { + // prepare sync object for new surface + try { + sess.swap_surface(*this); + return ExecutionStatus::Continue; + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << + "Abort"); + } + //return ExecutionStatus::Continue; + } + case MFX_ERR_MORE_DATA: // The function requires more bitstream at input before decoding can proceed + if (!sess.data_provider || sess.data_provider->empty()) { + // No more data to drain from decoder, start encode draining mode + return ExecutionStatus::Processed; + } + else + return ExecutionStatus::Continue; // read more data + break; + case MFX_ERR_MORE_SURFACE: + { + // The function requires more frame surface at output before decoding can proceed. + // This applies to external memory allocations and should not be expected for + // a simple internal allocation case like this + try { + sess.swap_surface(*this); + return ExecutionStatus::Continue; + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << + "Abort"); + } + break; + } + case MFX_ERR_DEVICE_LOST: + // For non-CPU implementations, + // Cleanup if device is lost + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " + "MFX_ERR_DEVICE_LOST is not processed"); + break; + case MFX_WRN_DEVICE_BUSY: + // For non-CPU implementations, + // Wait a few milliseconds then try again + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " + "MFX_WRN_DEVICE_BUSY is not processed"); + break; + case MFX_WRN_VIDEO_PARAM_CHANGED: + // The decoder detected a new sequence header in the bitstream. + // Video parameters may have changed. + // In external memory allocation case, might need to reallocate the output surface + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " + "MFX_WRN_VIDEO_PARAM_CHANGED is not processed"); + break; + case MFX_ERR_INCOMPATIBLE_VIDEO_PARAM: + // The function detected that video parameters provided by the application + // are incompatible with initialization parameters. + // The application should close the component and then reinitialize it + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " + "MFX_ERR_INCOMPATIBLE_VIDEO_PARAM is not processed"); + break; + case MFX_ERR_REALLOC_SURFACE: + // Bigger surface_work required. May be returned only if + // mfxInfoMFX::EnableReallocRequest was set to ON during initialization. + // This applies to external memory allocations and should not be expected for + // a simple internal allocation case like this + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " + "MFX_ERR_REALLOC_SURFACE is not processed"); + break; + case MFX_WRN_IN_EXECUTION: + try { + sess.swap_surface(*this); + return ExecutionStatus::Continue; + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << + "Abort"); + } + default: + GAPI_LOG_WARNING(nullptr, "Unknown status code: " << mfxstatus_to_string(status)); + break; + } + + return ExecutionStatus::Failed; +} + +} // namespace wip +} // namespace gapi +} // namespace cv diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp new file mode 100644 index 000000000000..db6ac8e53636 --- /dev/null +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp @@ -0,0 +1,47 @@ +// This file is part of OpenCV project. +// It is subject to the license terms in the LICENSE file found in the top-level directory +// of this distribution and at http://opencv.org/license.html. +// +// Copyright (C) 2021 Intel Corporation + +#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP +#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP +#include +#include + +#include "streaming/onevpl/engine/processing_engine_base.hpp" + +#ifdef HAVE_ONEVPL +#if (MFX_VERSION >= 2000) + #include +#endif +#include + +namespace cv { +namespace gapi { +namespace wip { + + +class LegacyDecodeSessionAsync; +struct DecoderParams; +struct IDataProvider; +struct VPLAccelerationPolicy; + +class VPLLegacyDecodeEngineAsync : public ProcessingEngineBase { +public: + + VPLLegacyDecodeEngineAsync(std::unique_ptr&& accel); + void initialize_session(mfxSession mfx_session, DecoderParams&& decoder_param, + std::shared_ptr provider) override; + +private: + ExecutionStatus execute_op(operation_t& op, EngineSession& sess) override; + ExecutionStatus process_error(mfxStatus status, LegacyDecodeSessionAsync& sess); + + void on_frame_ready(LegacyDecodeSessionAsync& sess, mfxFrameSurface1* ready_surface); +}; +} // namespace wip +} // namespace gapi +} // namespace cv +#endif // HAVE_ONEVPL +#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp new file mode 100644 index 000000000000..8231d34de453 --- /dev/null +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp @@ -0,0 +1,81 @@ +// This file is part of OpenCV project. +// It is subject to the license terms in the LICENSE file found in the top-level directory +// of this distribution and at http://opencv.org/license.html. +// +// Copyright (C) 2021 Intel Corporation + +#include +#include + +#include "streaming/onevpl/engine/decode/decode_session_async.hpp" +#include "streaming/onevpl/engine/decode/decode_engine_async.hpp" +#include "streaming/onevpl/accelerators/accel_policy_interface.hpp" +#include "streaming/onevpl/accelerators/surface/surface.hpp" +#include "streaming/onevpl/onevpl_utils.hpp" + +#include "logger.hpp" +namespace cv { +namespace gapi { +namespace wip { +LegacyDecodeSessionAsync::LegacyDecodeSessionAsync(mfxSession sess, + DecoderParams&& decoder_param, + std::shared_ptr provider) : + EngineSession(sess, std::move(decoder_param.stream)), + mfx_decoder_param(std::move(decoder_param.param)), + data_provider(std::move(provider)), + procesing_surface_ptr(), + sync_queue(), + decoded_frames_count() +{ +} + +LegacyDecodeSessionAsync::~LegacyDecodeSessionAsync() +{ + GAPI_LOG_INFO(nullptr, "Close Decode for session: " << session); + MFXVideoDECODE_Close(session); +} + +void LegacyDecodeSessionAsync::swap_surface(VPLLegacyDecodeEngineAsync& engine, bool new_async) { + VPLAccelerationPolicy* acceleration_policy = engine.get_accel(); + GAPI_Assert(acceleration_policy && "Empty acceleration_policy"); + try { + auto cand = acceleration_policy->get_free_surface(decoder_pool_id).lock(); + + GAPI_LOG_DEBUG(nullptr, "[" << session << "] swap surface" + ", old: " << (!procesing_surface_ptr.expired() + ? procesing_surface_ptr.lock()->get_handle() + : nullptr) << + ", new: "<< cand->get_handle()); + + procesing_surface_ptr = cand; + + // prepare sync object for new surface + if (new_async) { + decode_handle_t sync_pair{}; + sync_queue.emplace(sync_pair); + } + + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << session << "] error: " << ex.what() << + "Abort"); + } +} + +void LegacyDecodeSessionAsync::init_surface_pool(VPLAccelerationPolicy::pool_key_t key) { + GAPI_Assert(key && "Init decode pull with empty key"); + decoder_pool_id = key; +} + +Data::Meta LegacyDecodeSessionAsync::generate_frame_meta() { + const auto now = std::chrono::system_clock::now(); + const auto dur = std::chrono::duration_cast + (now.time_since_epoch()); + Data::Meta meta { + {cv::gapi::streaming::meta_tag::timestamp, int64_t{dur.count()} }, + {cv::gapi::streaming::meta_tag::seq_id, int64_t{decoded_frames_count++}} + }; + return meta; +} +} // namespace wip +} // namespace gapi +} // namespace cv diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp new file mode 100644 index 000000000000..a9dc38430b87 --- /dev/null +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp @@ -0,0 +1,58 @@ +// This file is part of OpenCV project. +// It is subject to the license terms in the LICENSE file found in the top-level directory +// of this distribution and at http://opencv.org/license.html. +// +// Copyright (C) 2021 Intel Corporation + +#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP +#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP +#include +#include +#include + +#include "streaming/onevpl/engine/engine_session.hpp" +#include "streaming/onevpl/accelerators/accel_policy_interface.hpp" +#ifdef HAVE_ONEVPL +#if (MFX_VERSION >= 2000) + #include +#endif +#include + +namespace cv { +namespace gapi { +namespace wip { + +struct IDataProvider; +class Surface; +struct VPLAccelerationPolicy; + +class LegacyDecodeSessionAsync : public EngineSession { +public: + friend class VPLLegacyDecodeEngineAsync; + + LegacyDecodeSessionAsync(mfxSession sess, DecoderParams&& decoder_param, std::shared_ptr provider); + ~LegacyDecodeSessionAsync(); + using EngineSession::EngineSession; + + void swap_surface(VPLLegacyDecodeEngineAsync& engine, bool new_async = true); + void init_surface_pool(VPLAccelerationPolicy::pool_key_t key); + + mfxVideoParam mfx_decoder_param; + std::shared_ptr data_provider; + + Data::Meta generate_frame_meta(); +private: + VPLAccelerationPolicy::pool_key_t decoder_pool_id; + mfxFrameAllocRequest request; + + std::weak_ptr procesing_surface_ptr; + + using decode_handle_t = std::pair; + std::queue sync_queue; + int64_t decoded_frames_count; +}; +} // namespace wip +} // namespace gapi +} // namespace cv +#endif // HAVE_ONEVPL +#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP diff --git a/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp b/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp index 4163c6a8a402..819d71a21614 100644 --- a/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp +++ b/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp @@ -9,6 +9,7 @@ #include "streaming/onevpl/onevpl_source_priv.hpp" #include "streaming/onevpl/engine/decode/decode_engine_legacy.hpp" +#include "streaming/onevpl/engine/decode/decode_engine_async.hpp" #include "streaming/onevpl/accelerators/accel_policy_dx11.hpp" #include "streaming/onevpl/accelerators/accel_policy_cpu.hpp" #include "streaming/onevpl/onevpl_utils.hpp" @@ -205,7 +206,8 @@ OneVPLSource::Priv::Priv(std::shared_ptr provider, const std::vec } else { - engine.reset(new VPLLegacyDecodeEngine(std::move(acceleration))); + //engine.reset(new VPLLegacyDecodeEngine(std::move(acceleration))); + engine.reset(new VPLLegacyDecodeEngineAsync(std::move(acceleration))); } } From 79fe2246964dabf8ddeff413a0a3c4b8cc5afa85 Mon Sep 17 00:00:00 2001 From: sivanov-work Date: Thu, 19 Aug 2021 10:15:52 +0300 Subject: [PATCH 5/8] Improve perf: activate CachedPool & Fix decode error --- .../gapi_streaming_source_perf_tests.cpp | 10 ++++++++-- modules/gapi/samples/infer_single_roi.cpp | 8 ++++++++ .../onevpl/accelerators/accel_policy_cpu.hpp | 1 + .../engine/decode/decode_engine_async.cpp | 17 ++++++++++++++--- 4 files changed, 31 insertions(+), 5 deletions(-) diff --git a/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp index b4b141523306..82babaf55b81 100644 --- a/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp +++ b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp @@ -16,6 +16,7 @@ using namespace perf; const std::string files[] = { "highgui/video/big_buck_bunny.h265", + "highgui/video/Putin.raw", "highgui/video/big_buck_bunny.h264", }; @@ -45,10 +46,13 @@ PERF_TEST_P_(OneVPLSourcePerfTest, TestPerformance) }; auto source_ptr = make_vpl_src(src, cfg_params); + //auto def_source_ptr = make_src(src); Data out; + //Data def_out; TEST_CYCLE() { source_ptr->pull(out); + //def_source_ptr->pull(def_out); } SANITY_CHECK_NOTHING(); @@ -71,9 +75,11 @@ PERF_TEST_P_(VideoCapSourcePerfTest, TestPerformance) INSTANTIATE_TEST_CASE_P(Streaming, OneVPLSourcePerfTest, Values(source_description_t(files[0], codec[0]), - source_description_t(files[1], codec[1]))); + source_description_t(files[1], codec[0]), + source_description_t(files[2], codec[1]))); INSTANTIATE_TEST_CASE_P(Streaming, VideoCapSourcePerfTest, Values(files[0], - files[1])); + files[1], + files[2])); } // namespace opencv_test diff --git a/modules/gapi/samples/infer_single_roi.cpp b/modules/gapi/samples/infer_single_roi.cpp index 6054a3f4a629..fcef19db49f8 100644 --- a/modules/gapi/samples/infer_single_roi.cpp +++ b/modules/gapi/samples/infer_single_roi.cpp @@ -256,9 +256,17 @@ int main(int argc, char *argv[]) pipeline.start(); cv::Mat out; + int framesCount = 0; + cv::TickMeter t; + t.start(); while (pipeline.pull(cv::gout(out))) { cv::imshow("Out", out); cv::waitKey(1); + framesCount++; } + t.stop(); + std::cout << "Elapsed time: " << t.getTimeSec() << std::endl; + std::cout << "FPS: " << framesCount / t.getTimeSec() << std::endl; + std::cout << "framesCount: " << framesCount << std::endl; return 0; } diff --git a/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp b/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp index ea326d234e92..350c8036959c 100644 --- a/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp +++ b/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp @@ -15,6 +15,7 @@ #ifdef HAVE_ONEVPL #include #include "streaming/onevpl/accelerators/accel_policy_interface.hpp" +#define TEST_PERF #ifdef TEST_PERF #include "streaming/onevpl/accelerators/surface/surface_pool.hpp" #endif // TEST_PERF diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp index 9234cc77eddc..cfeb891402df 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp @@ -140,9 +140,12 @@ VPLLegacyDecodeEngineAsync::VPLLegacyDecodeEngineAsync(std::unique_ptr Date: Thu, 19 Aug 2021 11:29:55 +0300 Subject: [PATCH 6/8] Enable permanent CachedPool --- .../gapi_streaming_source_perf_tests.cpp | 10 +-- .../onevpl/accelerators/accel_policy_cpu.cpp | 62 ++----------------- .../onevpl/accelerators/accel_policy_cpu.hpp | 8 +-- .../accelerators/surface/surface_pool.cpp | 13 +++- .../accelerators/surface/surface_pool.hpp | 4 +- 5 files changed, 22 insertions(+), 75 deletions(-) diff --git a/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp index 82babaf55b81..b4b141523306 100644 --- a/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp +++ b/modules/gapi/perf/streaming/gapi_streaming_source_perf_tests.cpp @@ -16,7 +16,6 @@ using namespace perf; const std::string files[] = { "highgui/video/big_buck_bunny.h265", - "highgui/video/Putin.raw", "highgui/video/big_buck_bunny.h264", }; @@ -46,13 +45,10 @@ PERF_TEST_P_(OneVPLSourcePerfTest, TestPerformance) }; auto source_ptr = make_vpl_src(src, cfg_params); - //auto def_source_ptr = make_src(src); Data out; - //Data def_out; TEST_CYCLE() { source_ptr->pull(out); - //def_source_ptr->pull(def_out); } SANITY_CHECK_NOTHING(); @@ -75,11 +71,9 @@ PERF_TEST_P_(VideoCapSourcePerfTest, TestPerformance) INSTANTIATE_TEST_CASE_P(Streaming, OneVPLSourcePerfTest, Values(source_description_t(files[0], codec[0]), - source_description_t(files[1], codec[0]), - source_description_t(files[2], codec[1]))); + source_description_t(files[1], codec[1]))); INSTANTIATE_TEST_CASE_P(Streaming, VideoCapSourcePerfTest, Values(files[0], - files[1], - files[2])); + files[1])); } // namespace opencv_test diff --git a/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.cpp b/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.cpp index c352c509c2f4..6a9a16862db0 100644 --- a/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.cpp +++ b/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.cpp @@ -107,7 +107,7 @@ VPLCPUAccelerationPolicy::create_surface_pool(size_t pool_size, size_t surface_s // remember pool by key GAPI_LOG_INFO(nullptr, "New pool allocated, key: " << preallocated_pool_memory_ptr << - ", surface count: " << pool.size() << + ", surface count: " << pool.total_size() << ", surface size bytes: " << surface_size_bytes); try { if (!pool_table.emplace(preallocated_pool_memory_ptr, std::move(pool)).second) { @@ -132,27 +132,7 @@ VPLCPUAccelerationPolicy::surface_weak_ptr_t VPLCPUAccelerationPolicy::get_free_ } pool_t& requested_pool = pool_it->second; -#ifdef TEST_PERF return requested_pool.find_free(); -#else // TEST_PERF - auto it = - std::find_if(requested_pool.begin(), requested_pool.end(), - [](const surface_ptr_t& val) { - GAPI_DbgAssert(val && "Pool contains empty surface"); - return !val->get_locks_count(); - }); - - // Limitation realloc pool might be a future extension - if (it == requested_pool.end()) { - std::stringstream ss; - ss << "cannot get free surface from pool, key: " << key << ", size: " << requested_pool.size(); - const std::string& str = ss.str(); - GAPI_LOG_WARNING(nullptr, str); - throw std::runtime_error(std::string(__FUNCTION__) + " - " + str); - } - - return *it; -#endif // TEST_PERF } size_t VPLCPUAccelerationPolicy::get_free_surface_count(pool_key_t key) const { @@ -162,18 +142,8 @@ size_t VPLCPUAccelerationPolicy::get_free_surface_count(pool_key_t key) const { ", table size: " << pool_table.size()); return 0; } -#ifdef TEST_PERF - return 0; -#else // TEST_PERF const pool_t& requested_pool = pool_it->second; - size_t free_surf_count = - std::count_if(requested_pool.begin(), requested_pool.end(), - [](const surface_ptr_t& val) { - GAPI_Assert(val && "Pool contains empty surface"); - return !val->get_locks_count(); - }); - return free_surf_count; -#endif // TEST_PERF + return requested_pool.available_size(); } size_t VPLCPUAccelerationPolicy::get_surface_count(pool_key_t key) const { @@ -183,11 +153,8 @@ size_t VPLCPUAccelerationPolicy::get_surface_count(pool_key_t key) const { ", table size: " << pool_table.size()); return 0; } -#ifdef TEST_PERF - return 0; -#else // TEST_PERF - return pool_it->second.size(); -#endif // TEST_PERF + + return pool_it->second.total_size(); } cv::MediaFrame::AdapterPtr VPLCPUAccelerationPolicy::create_frame_adapter(pool_key_t key, @@ -202,28 +169,7 @@ cv::MediaFrame::AdapterPtr VPLCPUAccelerationPolicy::create_frame_adapter(pool_k } pool_t& requested_pool = pool_it->second; -#ifdef TEST_PERF return cv::MediaFrame::AdapterPtr{new VPLMediaFrameCPUAdapter(requested_pool.find_by_handle(surface))}; -#else // TEST_PERF - auto it = - std::find_if(requested_pool.begin(), requested_pool.end(), - [surface](const surface_ptr_t& val) { - GAPI_DbgAssert(val && "Pool contains empty surface"); - return val->get_handle() == surface; - }); - - // Limitation realloc pool might be a future extension - if (it == requested_pool.end()) { - std::stringstream ss; - ss << "cannot get requested surface from pool, key: " << key << ", surf: " - << surface << ", pool size: " << requested_pool.size(); - const std::string& str = ss.str(); - GAPI_LOG_WARNING(nullptr, str); - throw std::runtime_error(std::string(__FUNCTION__) + " - " + str); - } - - return cv::MediaFrame::AdapterPtr{new VPLMediaFrameCPUAdapter(*it)}; -#endif // TEST_PERF } } // namespace wip } // namespace gapi diff --git a/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp b/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp index 350c8036959c..2cfefafde7fd 100644 --- a/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp +++ b/modules/gapi/src/streaming/onevpl/accelerators/accel_policy_cpu.hpp @@ -15,10 +15,7 @@ #ifdef HAVE_ONEVPL #include #include "streaming/onevpl/accelerators/accel_policy_interface.hpp" -#define TEST_PERF -#ifdef TEST_PERF #include "streaming/onevpl/accelerators/surface/surface_pool.hpp" -#endif // TEST_PERF namespace cv { namespace gapi { @@ -29,11 +26,8 @@ struct VPLCPUAccelerationPolicy final : public VPLAccelerationPolicy // GAPI_EXPORTS for tests GAPI_EXPORTS VPLCPUAccelerationPolicy(); GAPI_EXPORTS ~VPLCPUAccelerationPolicy(); -#ifdef TEST_PERF + using pool_t = CachedPool; -#else // TEST_PERF - using pool_t = std::vector; -#endif // TEST_PERF GAPI_EXPORTS void init(session_t session) override; GAPI_EXPORTS void deinit(session_t session) override; diff --git a/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.cpp b/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.cpp index 10e145c9a9d9..1e6b06b5048d 100644 --- a/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.cpp +++ b/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.cpp @@ -10,7 +10,7 @@ void CachedPool::reserve(size_t size) { surfaces.reserve(size); } -size_t CachedPool::size() const { +size_t CachedPool::total_size() const { return surfaces.size(); } @@ -26,6 +26,17 @@ void CachedPool::push_back(surface_ptr_t &&surf) { next_free_it = surfaces.begin(); } +size_t CachedPool::available_size() const { + + size_t free_surf_count = + std::count_if(surfaces.begin(), surfaces.end(), + [](const surface_ptr_t& val) { + GAPI_DbgAssert(val && "Pool contains empty surface"); + return !val->get_locks_count(); + }); + return free_surf_count; +} + CachedPool::surface_ptr_t CachedPool::find_free() { auto it = diff --git a/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.hpp b/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.hpp index a1eba15616b6..09779a1dc1ac 100644 --- a/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.hpp +++ b/modules/gapi/src/streaming/onevpl/accelerators/surface/surface_pool.hpp @@ -29,8 +29,10 @@ class CachedPool { // GAPI_EXPORTS for tests GAPI_EXPORTS void push_back(surface_ptr_t &&surf); GAPI_EXPORTS void reserve(size_t size); - GAPI_EXPORTS size_t size() const; + GAPI_EXPORTS size_t total_size() const; + GAPI_EXPORTS size_t available_size() const; GAPI_EXPORTS void clear(); + GAPI_EXPORTS surface_ptr_t find_free(); GAPI_EXPORTS surface_ptr_t find_by_handle(mfxFrameSurface1* handle); private: From 3903b0022b94c8f572f36f44a64f94c9483df059 Mon Sep 17 00:00:00 2001 From: sivanov-work Date: Thu, 19 Aug 2021 12:29:47 +0300 Subject: [PATCH 7/8] Clear async engine code --- .../engine/decode/decode_engine_async.cpp | 51 ++++++++++--------- .../engine/decode/decode_session_async.cpp | 9 +--- .../engine/decode/decode_session_async.hpp | 6 +-- 3 files changed, 30 insertions(+), 36 deletions(-) diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp index cfeb891402df..347300b25f96 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp @@ -115,7 +115,7 @@ VPLLegacyDecodeEngineAsync::VPLLegacyDecodeEngineAsync(std::unique_ptr ExecutionStatus { LegacyDecodeSessionAsync &my_sess = static_cast(sess); @@ -125,34 +125,37 @@ VPLLegacyDecodeEngineAsync::VPLLegacyDecodeEngineAsync(std::unique_ptr ExecutionStatus { LegacyDecodeSessionAsync &my_sess = static_cast(sess); + // prepare sync object for new surface + LegacyDecodeSessionAsync::op_handle_t sync_pair{}; + + // queue qecode operation my_sess.last_status = MFXVideoDECODE_DecodeFrameAsync(my_sess.session, my_sess.last_status == MFX_ERR_NONE ? &my_sess.stream : nullptr, /* No more data to read, start decode draining mode*/ my_sess.procesing_surface_ptr.lock()->get_handle(), + &sync_pair.second, + &sync_pair.first); - &my_sess.sync_queue.back().second, - &my_sess.sync_queue.back().first); - + // process wait-like statuses in-place while (my_sess.last_status == MFX_ERR_MORE_SURFACE || my_sess.last_status == MFX_WRN_DEVICE_BUSY) { try { if (my_sess.last_status == MFX_ERR_MORE_SURFACE) { - my_sess.swap_surface(*this, false); + my_sess.swap_surface(*this); } my_sess.last_status = MFXVideoDECODE_DecodeFrameAsync(my_sess.session, &my_sess.stream, my_sess.procesing_surface_ptr.lock()->get_handle(), - - &my_sess.sync_queue.back().second, - &my_sess.sync_queue.back().first); + &sync_pair.second, + &sync_pair.first); } catch (const std::exception& ex) { GAPI_LOG_WARNING(nullptr, "[" << my_sess.session << "] error: " << ex.what() << @@ -160,11 +163,12 @@ VPLLegacyDecodeEngineAsync::VPLLegacyDecodeEngineAsync(std::unique_ptr ExecutionStatus { LegacyDecodeSessionAsync& my_sess = static_cast(sess); - if (!my_sess.sync_queue.empty()/*sess.last_status == MFX_ERR_NONE*/) // Got 1 decoded frame + if (!my_sess.sync_queue.empty()) // FIFO: check the oldest async operation complete { - GAPI_LOG_DEBUG(nullptr, "my_sess.sync_queue size: " << my_sess.sync_queue.size() << - "my_sess.sync_queue.front().first: " << - my_sess.sync_queue.front().first); - sess.last_status = MFXVideoCORE_SyncOperation(sess.session, my_sess.sync_queue.front().first, 0); - /*GAPI_LOG_WARNING(nullptr, "my_sess.sync_queue size: " << my_sess.sync_queue.size() << - "my_sess.sync_queue.front().first: " << - my_sess.sync_queue.front().first << - "status: " << mfxstatus_to_string(sess.last_status));*/ + LegacyDecodeSessionAsync::op_handle_t& pending_op = my_sess.sync_queue.front(); + sess.last_status = MFXVideoCORE_SyncOperation(sess.session, pending_op.first, 0); + + GAPI_LOG_DEBUG(nullptr, "pending ops count: " << my_sess.sync_queue.size() << + ", sync id: " << pending_op.first << + ", status: " << mfxstatus_to_string(my_sess.last_status)); if (MFX_ERR_NONE == sess.last_status) { - on_frame_ready(my_sess, my_sess.sync_queue.front().second); + on_frame_ready(my_sess, pending_op.second); } } return ExecutionStatus::Continue; @@ -279,7 +281,6 @@ ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngineAsync::process_error( GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << "Abort"); } - //return ExecutionStatus::Continue; } case MFX_ERR_MORE_DATA: // The function requires more bitstream at input before decoding can proceed if (!sess.data_provider || sess.data_provider->empty()) { diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp index 8231d34de453..1366d114b7f5 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp @@ -35,7 +35,7 @@ LegacyDecodeSessionAsync::~LegacyDecodeSessionAsync() MFXVideoDECODE_Close(session); } -void LegacyDecodeSessionAsync::swap_surface(VPLLegacyDecodeEngineAsync& engine, bool new_async) { +void LegacyDecodeSessionAsync::swap_surface(VPLLegacyDecodeEngineAsync& engine) { VPLAccelerationPolicy* acceleration_policy = engine.get_accel(); GAPI_Assert(acceleration_policy && "Empty acceleration_policy"); try { @@ -48,13 +48,6 @@ void LegacyDecodeSessionAsync::swap_surface(VPLLegacyDecodeEngineAsync& engine, ", new: "<< cand->get_handle()); procesing_surface_ptr = cand; - - // prepare sync object for new surface - if (new_async) { - decode_handle_t sync_pair{}; - sync_queue.emplace(sync_pair); - } - } catch (const std::exception& ex) { GAPI_LOG_WARNING(nullptr, "[" << session << "] error: " << ex.what() << "Abort"); diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp index a9dc38430b87..6dc7752a7199 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp @@ -34,7 +34,7 @@ class LegacyDecodeSessionAsync : public EngineSession { ~LegacyDecodeSessionAsync(); using EngineSession::EngineSession; - void swap_surface(VPLLegacyDecodeEngineAsync& engine, bool new_async = true); + void swap_surface(VPLLegacyDecodeEngineAsync& engine); void init_surface_pool(VPLAccelerationPolicy::pool_key_t key); mfxVideoParam mfx_decoder_param; @@ -47,8 +47,8 @@ class LegacyDecodeSessionAsync : public EngineSession { std::weak_ptr procesing_surface_ptr; - using decode_handle_t = std::pair; - std::queue sync_queue; + using op_handle_t = std::pair; + std::queue sync_queue; int64_t decoded_frames_count; }; } // namespace wip From 85da2d36688ac8c79451a268428407d94e3e05b3 Mon Sep 17 00:00:00 2001 From: sivanov-work Date: Thu, 19 Aug 2021 14:21:32 +0300 Subject: [PATCH 8/8] Move out Async engine to Legacy --- modules/gapi/CMakeLists.txt | 2 - .../engine/decode/decode_engine_async.cpp | 360 ------------------ .../engine/decode/decode_engine_async.hpp | 47 --- .../engine/decode/decode_engine_legacy.cpp | 130 ++++--- .../engine/decode/decode_engine_legacy.hpp | 18 +- .../onevpl/engine/decode/decode_session.cpp | 17 +- .../onevpl/engine/decode/decode_session.hpp | 20 +- .../engine/decode/decode_session_async.cpp | 74 ---- .../engine/decode/decode_session_async.hpp | 58 --- .../streaming/onevpl/onevpl_source_priv.cpp | 2 - 10 files changed, 113 insertions(+), 615 deletions(-) delete mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp delete mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp delete mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp delete mode 100644 modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp diff --git a/modules/gapi/CMakeLists.txt b/modules/gapi/CMakeLists.txt index 0d69e78d9838..a5492c42bb4a 100644 --- a/modules/gapi/CMakeLists.txt +++ b/modules/gapi/CMakeLists.txt @@ -186,8 +186,6 @@ if(HAVE_GAPI_ONEVPL) src/streaming/onevpl/engine/processing_engine_base.cpp src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp src/streaming/onevpl/engine/decode/decode_session.cpp - src/streaming/onevpl/engine/decode/decode_engine_async.cpp - src/streaming/onevpl/engine/decode/decode_session_async.cpp src/streaming/onevpl/onevpl_utils.cpp src/streaming/onevpl/onevpl_cfg_params_parser.cpp ) diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp deleted file mode 100644 index 347300b25f96..000000000000 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.cpp +++ /dev/null @@ -1,360 +0,0 @@ -// This file is part of OpenCV project. -// It is subject to the license terms in the LICENSE file found in the top-level directory -// of this distribution and at http://opencv.org/license.html. -// -// Copyright (C) 2021 Intel Corporation - -#include -#include - -#include "streaming/onevpl/engine/decode/decode_engine_async.hpp" -#include "streaming/onevpl/engine/decode/decode_session_async.hpp" -#include "streaming/onevpl/accelerators/accel_policy_interface.hpp" -#include "streaming/onevpl/accelerators/surface/surface.hpp" -#include "streaming/onevpl/onevpl_utils.hpp" -//TODO #include "streaming/vpl/vpl_utils.hpp" -#include "logger.hpp" - - -namespace cv { -namespace gapi { -namespace wip { -/* UTILS */ -mfxU32 GetSurfaceSize_(mfxU32 FourCC, mfxU32 width, mfxU32 height) { - mfxU32 nbytes = 0; - - mfxU32 half_width = width / 2; - mfxU32 half_height = height / 2; - switch (FourCC) { - case MFX_FOURCC_I420: - case MFX_FOURCC_NV12: - nbytes = width * height + 2 * half_width * half_height; - break; - case MFX_FOURCC_I010: - case MFX_FOURCC_P010: - nbytes = width * height + 2 * half_width * half_height; - nbytes *= 2; - break; - case MFX_FOURCC_RGB4: - nbytes = width * height * 4; - break; - default: - break; - } - - return nbytes; -} - -surface_ptr_t create_surface_RGB4_(mfxFrameInfo frameInfo, - std::shared_ptr out_buf_ptr, - size_t out_buf_ptr_offset, - size_t out_buf_size) -{ - mfxU8* buf = reinterpret_cast(out_buf_ptr.get()); - mfxU16 surfW = frameInfo.Width * 4; - mfxU16 surfH = frameInfo.Height; - (void)surfH; - - // TODO more intelligent check - if (out_buf_size <= out_buf_ptr_offset) { - GAPI_LOG_WARNING(nullptr, "Not enough buffer, ptr: " << out_buf_ptr << - ", size: " << out_buf_size << - ", offset: " << out_buf_ptr_offset << - ", W: " << surfW << - ", H: " << surfH); - GAPI_Assert(false && "Invalid offset"); - } - - std::unique_ptr handle(new mfxFrameSurface1); - memset(handle.get(), 0, sizeof(mfxFrameSurface1)); - - handle->Info = frameInfo; - handle->Data.B = buf + out_buf_ptr_offset; - handle->Data.G = handle->Data.B + 1; - handle->Data.R = handle->Data.B + 2; - handle->Data.A = handle->Data.B + 3; - handle->Data.Pitch = surfW; - - return Surface::create_surface(std::move(handle), out_buf_ptr); -} - -surface_ptr_t create_surface_other_(mfxFrameInfo frameInfo, - std::shared_ptr out_buf_ptr, - size_t out_buf_ptr_offset, - size_t out_buf_size) -{ - mfxU8* buf = reinterpret_cast(out_buf_ptr.get()); - mfxU16 surfH = frameInfo.Height; - mfxU16 surfW = (frameInfo.FourCC == MFX_FOURCC_P010) ? frameInfo.Width * 2 : frameInfo.Width; - - // TODO more intelligent check - if (out_buf_size <= - out_buf_ptr_offset + (surfW * surfH) + ((surfW / 2) * (surfH / 2))) { - GAPI_LOG_WARNING(nullptr, "Not enough buffer, ptr: " << out_buf_ptr << - ", size: " << out_buf_size << - ", offset: " << out_buf_ptr_offset << - ", W: " << surfW << - ", H: " << surfH); - GAPI_Assert(false && "Invalid offset"); - } - - std::unique_ptr handle(new mfxFrameSurface1); - memset(handle.get(), 0, sizeof(mfxFrameSurface1)); - - handle->Info = frameInfo; - handle->Data.Y = buf + out_buf_ptr_offset; - handle->Data.U = buf + out_buf_ptr_offset + (surfW * surfH); - handle->Data.V = handle->Data.U + ((surfW / 2) * (surfH / 2)); - handle->Data.Pitch = surfW; - - return Surface::create_surface(std::move(handle), out_buf_ptr); -} - -VPLLegacyDecodeEngineAsync::VPLLegacyDecodeEngineAsync(std::unique_ptr&& accel) - : ProcessingEngineBase(std::move(accel)) { - - GAPI_LOG_INFO(nullptr, "Create Legacy Decode Engine"); - create_pipeline( - // 1) Read File - [this] (EngineSession& sess) -> ExecutionStatus - { - LegacyDecodeSessionAsync &my_sess = static_cast(sess); - my_sess.last_status = ReadEncodedStream(my_sess.stream, my_sess.data_provider); - if (my_sess.last_status != MFX_ERR_NONE) { - my_sess.data_provider.reset(); //close source - } - return ExecutionStatus::Continue; - }, - // 2) enqueue ASYNC decode operation - [this] (EngineSession& sess) -> ExecutionStatus - { - LegacyDecodeSessionAsync &my_sess = static_cast(sess); - - // prepare sync object for new surface - LegacyDecodeSessionAsync::op_handle_t sync_pair{}; - - // queue qecode operation - my_sess.last_status = - MFXVideoDECODE_DecodeFrameAsync(my_sess.session, - my_sess.last_status == MFX_ERR_NONE - ? &my_sess.stream - : nullptr, /* No more data to read, start decode draining mode*/ - my_sess.procesing_surface_ptr.lock()->get_handle(), - &sync_pair.second, - &sync_pair.first); - - // process wait-like statuses in-place - while (my_sess.last_status == MFX_ERR_MORE_SURFACE || - my_sess.last_status == MFX_WRN_DEVICE_BUSY) { - try { - if (my_sess.last_status == MFX_ERR_MORE_SURFACE) { - my_sess.swap_surface(*this); - } - my_sess.last_status = - MFXVideoDECODE_DecodeFrameAsync(my_sess.session, - &my_sess.stream, - my_sess.procesing_surface_ptr.lock()->get_handle(), - &sync_pair.second, - &sync_pair.first); - - } catch (const std::exception& ex) { - GAPI_LOG_WARNING(nullptr, "[" << my_sess.session << "] error: " << ex.what() << - "Abort"); - } - } - - if (my_sess.last_status == MFX_ERR_NONE) { - my_sess.sync_queue.emplace(sync_pair); - } else if (MFX_ERR_MORE_DATA != my_sess.last_status) /* suppress MFX_ERR_MORE_DATA warning */ { - GAPI_LOG_WARNING(nullptr, "pending ops count: " << my_sess.sync_queue.size() << - ", sync id: " << sync_pair.first << - ", status: " << mfxstatus_to_string(my_sess.last_status)); - } - return ExecutionStatus::Continue; - }, - // 3) Wait for ASYNC decode result - [this] (EngineSession& sess) -> ExecutionStatus - { - LegacyDecodeSessionAsync& my_sess = static_cast(sess); - if (!my_sess.sync_queue.empty()) // FIFO: check the oldest async operation complete - { - LegacyDecodeSessionAsync::op_handle_t& pending_op = my_sess.sync_queue.front(); - sess.last_status = MFXVideoCORE_SyncOperation(sess.session, pending_op.first, 0); - - GAPI_LOG_DEBUG(nullptr, "pending ops count: " << my_sess.sync_queue.size() << - ", sync id: " << pending_op.first << - ", status: " << mfxstatus_to_string(my_sess.last_status)); - if (MFX_ERR_NONE == sess.last_status) { - on_frame_ready(my_sess, pending_op.second); - } - } - return ExecutionStatus::Continue; - }, - // 4) Falls back on generic status procesing - [this] (EngineSession& sess) -> ExecutionStatus - { - return this->process_error(sess.last_status, static_cast(sess)); - } - ); -} - -void VPLLegacyDecodeEngineAsync::initialize_session(mfxSession mfx_session, - DecoderParams&& decoder_param, - std::shared_ptr provider) -{ - mfxFrameAllocRequest decRequest = {}; - // Query number required surfaces for decoder - MFXVideoDECODE_QueryIOSurf(mfx_session, &decoder_param.param, &decRequest); - - // External (application) allocation of decode surfaces - GAPI_LOG_DEBUG(nullptr, "Query IOSurf for session: " << mfx_session << - ", mfxFrameAllocRequest.NumFrameSuggested: " << decRequest.NumFrameSuggested << - ", mfxFrameAllocRequest.Type: " << decRequest.Type); - - mfxU32 singleSurfaceSize = GetSurfaceSize_(decoder_param.param.mfx.FrameInfo.FourCC, - decoder_param.param.mfx.FrameInfo.Width, - decoder_param.param.mfx.FrameInfo.Height); - if (!singleSurfaceSize) { - throw std::runtime_error("Cannot determine surface size for: fourCC" + - std::to_string(decoder_param.param.mfx.FrameInfo.FourCC) + - ", width: " + std::to_string(decoder_param.param.mfx.FrameInfo.Width) + - ", height: " + std::to_string(decoder_param.param.mfx.FrameInfo.Height)); - } - - const auto &frameInfo = decoder_param.param.mfx.FrameInfo; - auto surface_creator = - [&frameInfo] (std::shared_ptr out_buf_ptr, size_t out_buf_ptr_offset, - size_t out_buf_size) -> surface_ptr_t { - return (frameInfo.FourCC == MFX_FOURCC_RGB4) ? - create_surface_RGB4_(frameInfo, out_buf_ptr, out_buf_ptr_offset, - out_buf_size) : - create_surface_other_(frameInfo, out_buf_ptr, out_buf_ptr_offset, - out_buf_size);}; - - //TODO Configure preallocation size (how many frames we can hold) - const size_t preallocated_frames_count = 30; - VPLAccelerationPolicy::pool_key_t decode_pool_key = - acceleration_policy->create_surface_pool(decRequest.NumFrameSuggested * preallocated_frames_count, - singleSurfaceSize, - surface_creator); - - // create session - std::shared_ptr sess_ptr = - register_session(mfx_session, - std::move(decoder_param), - provider); - - sess_ptr->init_surface_pool(decode_pool_key); - // prepare working decode surface - sess_ptr->swap_surface(*this); -} - -ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngineAsync::execute_op(operation_t& op, EngineSession& sess) { - return op(sess); -} - -void VPLLegacyDecodeEngineAsync::on_frame_ready(LegacyDecodeSessionAsync& sess, mfxFrameSurface1* ready_surface) -{ - GAPI_LOG_DEBUG(nullptr, "[" << sess.session << "], frame ready"); - - // manage memory ownership rely on acceleration policy - auto frame_adapter = acceleration_policy->create_frame_adapter(sess.decoder_pool_id, - ready_surface); - ready_frames.emplace(cv::MediaFrame(std::move(frame_adapter)), sess.generate_frame_meta()); - - // pop ready sync onject - sess.sync_queue.pop(); -} - -ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngineAsync::process_error(mfxStatus status, LegacyDecodeSessionAsync& sess) -{ - GAPI_LOG_DEBUG(nullptr, "status: " << mfxstatus_to_string(status)); - - switch (status) { - case MFX_ERR_NONE: - { - // prepare sync object for new surface - try { - sess.swap_surface(*this); - return ExecutionStatus::Continue; - } catch (const std::exception& ex) { - GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << - "Abort"); - } - } - case MFX_ERR_MORE_DATA: // The function requires more bitstream at input before decoding can proceed - if (!sess.data_provider || sess.data_provider->empty()) { - // No more data to drain from decoder, start encode draining mode - return ExecutionStatus::Processed; - } - else - return ExecutionStatus::Continue; // read more data - break; - case MFX_ERR_MORE_SURFACE: - { - // The function requires more frame surface at output before decoding can proceed. - // This applies to external memory allocations and should not be expected for - // a simple internal allocation case like this - try { - sess.swap_surface(*this); - return ExecutionStatus::Continue; - } catch (const std::exception& ex) { - GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << - "Abort"); - } - break; - } - case MFX_ERR_DEVICE_LOST: - // For non-CPU implementations, - // Cleanup if device is lost - GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " - "MFX_ERR_DEVICE_LOST is not processed"); - break; - case MFX_WRN_DEVICE_BUSY: - // For non-CPU implementations, - // Wait a few milliseconds then try again - GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " - "MFX_WRN_DEVICE_BUSY is not processed"); - break; - case MFX_WRN_VIDEO_PARAM_CHANGED: - // The decoder detected a new sequence header in the bitstream. - // Video parameters may have changed. - // In external memory allocation case, might need to reallocate the output surface - GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " - "MFX_WRN_VIDEO_PARAM_CHANGED is not processed"); - break; - case MFX_ERR_INCOMPATIBLE_VIDEO_PARAM: - // The function detected that video parameters provided by the application - // are incompatible with initialization parameters. - // The application should close the component and then reinitialize it - GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " - "MFX_ERR_INCOMPATIBLE_VIDEO_PARAM is not processed"); - break; - case MFX_ERR_REALLOC_SURFACE: - // Bigger surface_work required. May be returned only if - // mfxInfoMFX::EnableReallocRequest was set to ON during initialization. - // This applies to external memory allocations and should not be expected for - // a simple internal allocation case like this - GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " - "MFX_ERR_REALLOC_SURFACE is not processed"); - break; - case MFX_WRN_IN_EXECUTION: - try { - sess.swap_surface(*this); - return ExecutionStatus::Continue; - } catch (const std::exception& ex) { - GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << - "Abort"); - } - default: - GAPI_LOG_WARNING(nullptr, "Unknown status code: " << mfxstatus_to_string(status) << - ", decoded frames: " << sess.decoded_frames_count); - break; - } - - return ExecutionStatus::Failed; -} - -} // namespace wip -} // namespace gapi -} // namespace cv diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp deleted file mode 100644 index db6ac8e53636..000000000000 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_async.hpp +++ /dev/null @@ -1,47 +0,0 @@ -// This file is part of OpenCV project. -// It is subject to the license terms in the LICENSE file found in the top-level directory -// of this distribution and at http://opencv.org/license.html. -// -// Copyright (C) 2021 Intel Corporation - -#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP -#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP -#include -#include - -#include "streaming/onevpl/engine/processing_engine_base.hpp" - -#ifdef HAVE_ONEVPL -#if (MFX_VERSION >= 2000) - #include -#endif -#include - -namespace cv { -namespace gapi { -namespace wip { - - -class LegacyDecodeSessionAsync; -struct DecoderParams; -struct IDataProvider; -struct VPLAccelerationPolicy; - -class VPLLegacyDecodeEngineAsync : public ProcessingEngineBase { -public: - - VPLLegacyDecodeEngineAsync(std::unique_ptr&& accel); - void initialize_session(mfxSession mfx_session, DecoderParams&& decoder_param, - std::shared_ptr provider) override; - -private: - ExecutionStatus execute_op(operation_t& op, EngineSession& sess) override; - ExecutionStatus process_error(mfxStatus status, LegacyDecodeSessionAsync& sess); - - void on_frame_ready(LegacyDecodeSessionAsync& sess, mfxFrameSurface1* ready_surface); -}; -} // namespace wip -} // namespace gapi -} // namespace cv -#endif // HAVE_ONEVPL -#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp index e73a1749b902..e5fbe928bd87 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.cpp @@ -20,7 +20,7 @@ namespace cv { namespace gapi { namespace wip { /* UTILS */ -mfxU32 GetSurfaceSize(mfxU32 FourCC, mfxU32 width, mfxU32 height) { +mfxU32 GetSurfaceSize_(mfxU32 FourCC, mfxU32 width, mfxU32 height) { mfxU32 nbytes = 0; mfxU32 half_width = width / 2; @@ -45,7 +45,7 @@ mfxU32 GetSurfaceSize(mfxU32 FourCC, mfxU32 width, mfxU32 height) { return nbytes; } -surface_ptr_t create_surface_RGB4(mfxFrameInfo frameInfo, +surface_ptr_t create_surface_RGB4_(mfxFrameInfo frameInfo, std::shared_ptr out_buf_ptr, size_t out_buf_ptr_offset, size_t out_buf_size) @@ -78,7 +78,7 @@ surface_ptr_t create_surface_RGB4(mfxFrameInfo frameInfo, return Surface::create_surface(std::move(handle), out_buf_ptr); } -surface_ptr_t create_surface_other(mfxFrameInfo frameInfo, +surface_ptr_t create_surface_other_(mfxFrameInfo frameInfo, std::shared_ptr out_buf_ptr, size_t out_buf_ptr_offset, size_t out_buf_size) @@ -110,69 +110,95 @@ surface_ptr_t create_surface_other(mfxFrameInfo frameInfo, return Surface::create_surface(std::move(handle), out_buf_ptr); } -VPLLegacyDecodeEngine::VPLLegacyDecodeEngine(std::unique_ptr&& accel) +VPLLegacyDecodeEngineAsync::VPLLegacyDecodeEngineAsync(std::unique_ptr&& accel) : ProcessingEngineBase(std::move(accel)) { GAPI_LOG_INFO(nullptr, "Create Legacy Decode Engine"); create_pipeline( - // 1) Reade File + // 1) Read File [this] (EngineSession& sess) -> ExecutionStatus { - LegacyDecodeSession &my_sess = static_cast(sess); + LegacyDecodeSessionAsync &my_sess = static_cast(sess); my_sess.last_status = ReadEncodedStream(my_sess.stream, my_sess.data_provider); if (my_sess.last_status != MFX_ERR_NONE) { my_sess.data_provider.reset(); //close source } return ExecutionStatus::Continue; }, - // 2) enqueue ASYNC decode + // 2) enqueue ASYNC decode operation [this] (EngineSession& sess) -> ExecutionStatus { - LegacyDecodeSession &my_sess = static_cast(sess); + LegacyDecodeSessionAsync &my_sess = static_cast(sess); + // prepare sync object for new surface + LegacyDecodeSessionAsync::op_handle_t sync_pair{}; + + // queue qecode operation my_sess.last_status = MFXVideoDECODE_DecodeFrameAsync(my_sess.session, my_sess.last_status == MFX_ERR_NONE ? &my_sess.stream : nullptr, /* No more data to read, start decode draining mode*/ my_sess.procesing_surface_ptr.lock()->get_handle(), - &my_sess.output_surface_ptr, - &my_sess.sync); + &sync_pair.second, + &sync_pair.first); + + // process wait-like statuses in-place + while (my_sess.last_status == MFX_ERR_MORE_SURFACE || + my_sess.last_status == MFX_WRN_DEVICE_BUSY) { + try { + if (my_sess.last_status == MFX_ERR_MORE_SURFACE) { + my_sess.swap_surface(*this); + } + my_sess.last_status = + MFXVideoDECODE_DecodeFrameAsync(my_sess.session, + &my_sess.stream, + my_sess.procesing_surface_ptr.lock()->get_handle(), + &sync_pair.second, + &sync_pair.first); - if (!pending_frames.empty()) - { - auto data = pending_frames.front(); - pending_frames.pop(); - ready_frames.push(std::move(data)); + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << my_sess.session << "] error: " << ex.what() << + "Abort"); + } + } + + if (my_sess.last_status == MFX_ERR_NONE) { + my_sess.sync_queue.emplace(sync_pair); + } else if (MFX_ERR_MORE_DATA != my_sess.last_status) /* suppress MFX_ERR_MORE_DATA warning */ { + GAPI_LOG_WARNING(nullptr, "pending ops count: " << my_sess.sync_queue.size() << + ", sync id: " << sync_pair.first << + ", status: " << mfxstatus_to_string(my_sess.last_status)); } return ExecutionStatus::Continue; }, // 3) Wait for ASYNC decode result [this] (EngineSession& sess) -> ExecutionStatus { - if (sess.last_status == MFX_ERR_NONE) // Got 1 decoded frame + LegacyDecodeSessionAsync& my_sess = static_cast(sess); + if (!my_sess.sync_queue.empty()) // FIFO: check the oldest async operation complete { - do { - //TODO try to extract TIMESTAMP - sess.last_status = MFXVideoCORE_SyncOperation(sess.session, sess.sync, 100); - if (MFX_ERR_NONE == sess.last_status) { - - LegacyDecodeSession& my_sess = static_cast(sess); - on_frame_ready(my_sess); - } - } while (sess.last_status == MFX_WRN_IN_EXECUTION); + LegacyDecodeSessionAsync::op_handle_t& pending_op = my_sess.sync_queue.front(); + sess.last_status = MFXVideoCORE_SyncOperation(sess.session, pending_op.first, 0); + + GAPI_LOG_DEBUG(nullptr, "pending ops count: " << my_sess.sync_queue.size() << + ", sync id: " << pending_op.first << + ", status: " << mfxstatus_to_string(my_sess.last_status)); + if (MFX_ERR_NONE == sess.last_status) { + on_frame_ready(my_sess, pending_op.second); + } } return ExecutionStatus::Continue; }, // 4) Falls back on generic status procesing [this] (EngineSession& sess) -> ExecutionStatus { - return this->process_error(sess.last_status, static_cast(sess)); + return this->process_error(sess.last_status, static_cast(sess)); } ); } -void VPLLegacyDecodeEngine::initialize_session(mfxSession mfx_session, +void VPLLegacyDecodeEngineAsync::initialize_session(mfxSession mfx_session, DecoderParams&& decoder_param, std::shared_ptr provider) { @@ -185,7 +211,7 @@ void VPLLegacyDecodeEngine::initialize_session(mfxSession mfx_session, ", mfxFrameAllocRequest.NumFrameSuggested: " << decRequest.NumFrameSuggested << ", mfxFrameAllocRequest.Type: " << decRequest.Type); - mfxU32 singleSurfaceSize = GetSurfaceSize(decoder_param.param.mfx.FrameInfo.FourCC, + mfxU32 singleSurfaceSize = GetSurfaceSize_(decoder_param.param.mfx.FrameInfo.FourCC, decoder_param.param.mfx.FrameInfo.Width, decoder_param.param.mfx.FrameInfo.Height); if (!singleSurfaceSize) { @@ -200,9 +226,9 @@ void VPLLegacyDecodeEngine::initialize_session(mfxSession mfx_session, [&frameInfo] (std::shared_ptr out_buf_ptr, size_t out_buf_ptr_offset, size_t out_buf_size) -> surface_ptr_t { return (frameInfo.FourCC == MFX_FOURCC_RGB4) ? - create_surface_RGB4(frameInfo, out_buf_ptr, out_buf_ptr_offset, + create_surface_RGB4_(frameInfo, out_buf_ptr, out_buf_ptr_offset, out_buf_size) : - create_surface_other(frameInfo, out_buf_ptr, out_buf_ptr_offset, + create_surface_other_(frameInfo, out_buf_ptr, out_buf_ptr_offset, out_buf_size);}; //TODO Configure preallocation size (how many frames we can hold) @@ -213,8 +239,8 @@ void VPLLegacyDecodeEngine::initialize_session(mfxSession mfx_session, surface_creator); // create session - std::shared_ptr sess_ptr = - register_session(mfx_session, + std::shared_ptr sess_ptr = + register_session(mfx_session, std::move(decoder_param), provider); @@ -223,34 +249,39 @@ void VPLLegacyDecodeEngine::initialize_session(mfxSession mfx_session, sess_ptr->swap_surface(*this); } -ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngine::execute_op(operation_t& op, EngineSession& sess) { +ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngineAsync::execute_op(operation_t& op, EngineSession& sess) { return op(sess); } -void VPLLegacyDecodeEngine::on_frame_ready(LegacyDecodeSession& sess) +void VPLLegacyDecodeEngineAsync::on_frame_ready(LegacyDecodeSessionAsync& sess, mfxFrameSurface1* ready_surface) { GAPI_LOG_DEBUG(nullptr, "[" << sess.session << "], frame ready"); // manage memory ownership rely on acceleration policy auto frame_adapter = acceleration_policy->create_frame_adapter(sess.decoder_pool_id, - sess.output_surface_ptr); - pending_frames.emplace(cv::MediaFrame(std::move(frame_adapter)), sess.generate_frame_meta()); + ready_surface); + ready_frames.emplace(cv::MediaFrame(std::move(frame_adapter)), sess.generate_frame_meta()); + + // pop ready sync onject + sess.sync_queue.pop(); } -ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngine::process_error(mfxStatus status, LegacyDecodeSession& sess) +ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngineAsync::process_error(mfxStatus status, LegacyDecodeSessionAsync& sess) { GAPI_LOG_DEBUG(nullptr, "status: " << mfxstatus_to_string(status)); switch (status) { case MFX_ERR_NONE: + { + // prepare sync object for new surface try { sess.swap_surface(*this); + return ExecutionStatus::Continue; } catch (const std::exception& ex) { GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << "Abort"); - ExecutionStatus::Failed; } - return ExecutionStatus::Continue; + } case MFX_ERR_MORE_DATA: // The function requires more bitstream at input before decoding can proceed if (!sess.data_provider || sess.data_provider->empty()) { // No more data to drain from decoder, start encode draining mode @@ -276,27 +307,27 @@ ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngine::process_error(mfxSt case MFX_ERR_DEVICE_LOST: // For non-CPU implementations, // Cleanup if device is lost - GAPI_DbgAssert(false && "VPLLegacyDecodeEngine::process_error - " + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " "MFX_ERR_DEVICE_LOST is not processed"); break; case MFX_WRN_DEVICE_BUSY: // For non-CPU implementations, // Wait a few milliseconds then try again - GAPI_DbgAssert(false && "VPLLegacyDecodeEngine::process_error - " + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " "MFX_WRN_DEVICE_BUSY is not processed"); break; case MFX_WRN_VIDEO_PARAM_CHANGED: // The decoder detected a new sequence header in the bitstream. // Video parameters may have changed. // In external memory allocation case, might need to reallocate the output surface - GAPI_DbgAssert(false && "VPLLegacyDecodeEngine::process_error - " + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " "MFX_WRN_VIDEO_PARAM_CHANGED is not processed"); break; case MFX_ERR_INCOMPATIBLE_VIDEO_PARAM: // The function detected that video parameters provided by the application // are incompatible with initialization parameters. // The application should close the component and then reinitialize it - GAPI_DbgAssert(false && "VPLLegacyDecodeEngine::process_error - " + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " "MFX_ERR_INCOMPATIBLE_VIDEO_PARAM is not processed"); break; case MFX_ERR_REALLOC_SURFACE: @@ -304,11 +335,20 @@ ProcessingEngineBase::ExecutionStatus VPLLegacyDecodeEngine::process_error(mfxSt // mfxInfoMFX::EnableReallocRequest was set to ON during initialization. // This applies to external memory allocations and should not be expected for // a simple internal allocation case like this - GAPI_DbgAssert(false && "VPLLegacyDecodeEngine::process_error - " + GAPI_DbgAssert(false && "VPLLegacyDecodeEngineAsync::process_error - " "MFX_ERR_REALLOC_SURFACE is not processed"); break; + case MFX_WRN_IN_EXECUTION: + try { + sess.swap_surface(*this); + return ExecutionStatus::Continue; + } catch (const std::exception& ex) { + GAPI_LOG_WARNING(nullptr, "[" << sess.session << "] error: " << ex.what() << + "Abort"); + } default: - GAPI_LOG_WARNING(nullptr, "Unknown status code: " << mfxstatus_to_string(status)); + GAPI_LOG_WARNING(nullptr, "Unknown status code: " << mfxstatus_to_string(status) << + ", decoded frames: " << sess.decoded_frames_count); break; } diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp index 66263be31cf4..db6ac8e53636 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_engine_legacy.hpp @@ -4,8 +4,8 @@ // // Copyright (C) 2021 Intel Corporation -#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_LEGACY_HPP -#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_LEGACY_HPP +#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP +#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP #include #include @@ -22,28 +22,26 @@ namespace gapi { namespace wip { -class LegacyDecodeSession; +class LegacyDecodeSessionAsync; struct DecoderParams; struct IDataProvider; struct VPLAccelerationPolicy; -class VPLLegacyDecodeEngine : public ProcessingEngineBase { +class VPLLegacyDecodeEngineAsync : public ProcessingEngineBase { public: - VPLLegacyDecodeEngine(std::unique_ptr&& accel); + VPLLegacyDecodeEngineAsync(std::unique_ptr&& accel); void initialize_session(mfxSession mfx_session, DecoderParams&& decoder_param, std::shared_ptr provider) override; private: ExecutionStatus execute_op(operation_t& op, EngineSession& sess) override; - ExecutionStatus process_error(mfxStatus status, LegacyDecodeSession& sess); + ExecutionStatus process_error(mfxStatus status, LegacyDecodeSessionAsync& sess); - void on_frame_ready(LegacyDecodeSession& sess); - - frames_container_t pending_frames; + void on_frame_ready(LegacyDecodeSessionAsync& sess, mfxFrameSurface1* ready_surface); }; } // namespace wip } // namespace gapi } // namespace cv #endif // HAVE_ONEVPL -#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_LEGACY_HPP +#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_ENGINE_ASYNC_HPP diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.cpp index 5653bb499491..7cb145d83df9 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.cpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.cpp @@ -17,33 +17,34 @@ namespace cv { namespace gapi { namespace wip { -LegacyDecodeSession::LegacyDecodeSession(mfxSession sess, +LegacyDecodeSessionAsync::LegacyDecodeSessionAsync(mfxSession sess, DecoderParams&& decoder_param, std::shared_ptr provider) : EngineSession(sess, std::move(decoder_param.stream)), mfx_decoder_param(std::move(decoder_param.param)), data_provider(std::move(provider)), procesing_surface_ptr(), - output_surface_ptr(), + sync_queue(), decoded_frames_count() { } -LegacyDecodeSession::~LegacyDecodeSession() +LegacyDecodeSessionAsync::~LegacyDecodeSessionAsync() { GAPI_LOG_INFO(nullptr, "Close Decode for session: " << session); MFXVideoDECODE_Close(session); } -void LegacyDecodeSession::swap_surface(VPLLegacyDecodeEngine& engine) { +void LegacyDecodeSessionAsync::swap_surface(VPLLegacyDecodeEngineAsync& engine) { VPLAccelerationPolicy* acceleration_policy = engine.get_accel(); GAPI_Assert(acceleration_policy && "Empty acceleration_policy"); - auto old_locked = procesing_surface_ptr.lock(); try { auto cand = acceleration_policy->get_free_surface(decoder_pool_id).lock(); GAPI_LOG_DEBUG(nullptr, "[" << session << "] swap surface" - ", old: " << (old_locked ? old_locked->get_handle() : nullptr) << + ", old: " << (!procesing_surface_ptr.expired() + ? procesing_surface_ptr.lock()->get_handle() + : nullptr) << ", new: "<< cand->get_handle()); procesing_surface_ptr = cand; @@ -53,12 +54,12 @@ void LegacyDecodeSession::swap_surface(VPLLegacyDecodeEngine& engine) { } } -void LegacyDecodeSession::init_surface_pool(VPLAccelerationPolicy::pool_key_t key) { +void LegacyDecodeSessionAsync::init_surface_pool(VPLAccelerationPolicy::pool_key_t key) { GAPI_Assert(key && "Init decode pull with empty key"); decoder_pool_id = key; } -Data::Meta LegacyDecodeSession::generate_frame_meta() { +Data::Meta LegacyDecodeSessionAsync::generate_frame_meta() { const auto now = std::chrono::system_clock::now(); const auto dur = std::chrono::duration_cast (now.time_since_epoch()); diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.hpp index bc6d7a479026..6dc7752a7199 100644 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.hpp +++ b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session.hpp @@ -4,10 +4,11 @@ // // Copyright (C) 2021 Intel Corporation -#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_HPP -#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_HPP +#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP +#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP #include #include +#include #include "streaming/onevpl/engine/engine_session.hpp" #include "streaming/onevpl/accelerators/accel_policy_interface.hpp" @@ -25,15 +26,15 @@ struct IDataProvider; class Surface; struct VPLAccelerationPolicy; -class LegacyDecodeSession : public EngineSession { +class LegacyDecodeSessionAsync : public EngineSession { public: - friend class VPLLegacyDecodeEngine; + friend class VPLLegacyDecodeEngineAsync; - LegacyDecodeSession(mfxSession sess, DecoderParams&& decoder_param, std::shared_ptr provider); - ~LegacyDecodeSession(); + LegacyDecodeSessionAsync(mfxSession sess, DecoderParams&& decoder_param, std::shared_ptr provider); + ~LegacyDecodeSessionAsync(); using EngineSession::EngineSession; - void swap_surface(VPLLegacyDecodeEngine& engine); + void swap_surface(VPLLegacyDecodeEngineAsync& engine); void init_surface_pool(VPLAccelerationPolicy::pool_key_t key); mfxVideoParam mfx_decoder_param; @@ -45,12 +46,13 @@ class LegacyDecodeSession : public EngineSession { mfxFrameAllocRequest request; std::weak_ptr procesing_surface_ptr; - mfxFrameSurface1* output_surface_ptr; + using op_handle_t = std::pair; + std::queue sync_queue; int64_t decoded_frames_count; }; } // namespace wip } // namespace gapi } // namespace cv #endif // HAVE_ONEVPL -#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_HPP +#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp deleted file mode 100644 index 1366d114b7f5..000000000000 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.cpp +++ /dev/null @@ -1,74 +0,0 @@ -// This file is part of OpenCV project. -// It is subject to the license terms in the LICENSE file found in the top-level directory -// of this distribution and at http://opencv.org/license.html. -// -// Copyright (C) 2021 Intel Corporation - -#include -#include - -#include "streaming/onevpl/engine/decode/decode_session_async.hpp" -#include "streaming/onevpl/engine/decode/decode_engine_async.hpp" -#include "streaming/onevpl/accelerators/accel_policy_interface.hpp" -#include "streaming/onevpl/accelerators/surface/surface.hpp" -#include "streaming/onevpl/onevpl_utils.hpp" - -#include "logger.hpp" -namespace cv { -namespace gapi { -namespace wip { -LegacyDecodeSessionAsync::LegacyDecodeSessionAsync(mfxSession sess, - DecoderParams&& decoder_param, - std::shared_ptr provider) : - EngineSession(sess, std::move(decoder_param.stream)), - mfx_decoder_param(std::move(decoder_param.param)), - data_provider(std::move(provider)), - procesing_surface_ptr(), - sync_queue(), - decoded_frames_count() -{ -} - -LegacyDecodeSessionAsync::~LegacyDecodeSessionAsync() -{ - GAPI_LOG_INFO(nullptr, "Close Decode for session: " << session); - MFXVideoDECODE_Close(session); -} - -void LegacyDecodeSessionAsync::swap_surface(VPLLegacyDecodeEngineAsync& engine) { - VPLAccelerationPolicy* acceleration_policy = engine.get_accel(); - GAPI_Assert(acceleration_policy && "Empty acceleration_policy"); - try { - auto cand = acceleration_policy->get_free_surface(decoder_pool_id).lock(); - - GAPI_LOG_DEBUG(nullptr, "[" << session << "] swap surface" - ", old: " << (!procesing_surface_ptr.expired() - ? procesing_surface_ptr.lock()->get_handle() - : nullptr) << - ", new: "<< cand->get_handle()); - - procesing_surface_ptr = cand; - } catch (const std::exception& ex) { - GAPI_LOG_WARNING(nullptr, "[" << session << "] error: " << ex.what() << - "Abort"); - } -} - -void LegacyDecodeSessionAsync::init_surface_pool(VPLAccelerationPolicy::pool_key_t key) { - GAPI_Assert(key && "Init decode pull with empty key"); - decoder_pool_id = key; -} - -Data::Meta LegacyDecodeSessionAsync::generate_frame_meta() { - const auto now = std::chrono::system_clock::now(); - const auto dur = std::chrono::duration_cast - (now.time_since_epoch()); - Data::Meta meta { - {cv::gapi::streaming::meta_tag::timestamp, int64_t{dur.count()} }, - {cv::gapi::streaming::meta_tag::seq_id, int64_t{decoded_frames_count++}} - }; - return meta; -} -} // namespace wip -} // namespace gapi -} // namespace cv diff --git a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp b/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp deleted file mode 100644 index 6dc7752a7199..000000000000 --- a/modules/gapi/src/streaming/onevpl/engine/decode/decode_session_async.hpp +++ /dev/null @@ -1,58 +0,0 @@ -// This file is part of OpenCV project. -// It is subject to the license terms in the LICENSE file found in the top-level directory -// of this distribution and at http://opencv.org/license.html. -// -// Copyright (C) 2021 Intel Corporation - -#ifndef GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP -#define GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP -#include -#include -#include - -#include "streaming/onevpl/engine/engine_session.hpp" -#include "streaming/onevpl/accelerators/accel_policy_interface.hpp" -#ifdef HAVE_ONEVPL -#if (MFX_VERSION >= 2000) - #include -#endif -#include - -namespace cv { -namespace gapi { -namespace wip { - -struct IDataProvider; -class Surface; -struct VPLAccelerationPolicy; - -class LegacyDecodeSessionAsync : public EngineSession { -public: - friend class VPLLegacyDecodeEngineAsync; - - LegacyDecodeSessionAsync(mfxSession sess, DecoderParams&& decoder_param, std::shared_ptr provider); - ~LegacyDecodeSessionAsync(); - using EngineSession::EngineSession; - - void swap_surface(VPLLegacyDecodeEngineAsync& engine); - void init_surface_pool(VPLAccelerationPolicy::pool_key_t key); - - mfxVideoParam mfx_decoder_param; - std::shared_ptr data_provider; - - Data::Meta generate_frame_meta(); -private: - VPLAccelerationPolicy::pool_key_t decoder_pool_id; - mfxFrameAllocRequest request; - - std::weak_ptr procesing_surface_ptr; - - using op_handle_t = std::pair; - std::queue sync_queue; - int64_t decoded_frames_count; -}; -} // namespace wip -} // namespace gapi -} // namespace cv -#endif // HAVE_ONEVPL -#endif // GAPI_STREAMING_ONVPL_ENGINE_DECODE_DECODE_SESSION_ASYNC_HPP diff --git a/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp b/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp index 819d71a21614..85c3c8077941 100644 --- a/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp +++ b/modules/gapi/src/streaming/onevpl/onevpl_source_priv.cpp @@ -9,7 +9,6 @@ #include "streaming/onevpl/onevpl_source_priv.hpp" #include "streaming/onevpl/engine/decode/decode_engine_legacy.hpp" -#include "streaming/onevpl/engine/decode/decode_engine_async.hpp" #include "streaming/onevpl/accelerators/accel_policy_dx11.hpp" #include "streaming/onevpl/accelerators/accel_policy_cpu.hpp" #include "streaming/onevpl/onevpl_utils.hpp" @@ -206,7 +205,6 @@ OneVPLSource::Priv::Priv(std::shared_ptr provider, const std::vec } else { - //engine.reset(new VPLLegacyDecodeEngine(std::move(acceleration))); engine.reset(new VPLLegacyDecodeEngineAsync(std::move(acceleration))); } }