From: TolyaTalamanov Date: Thu, 29 Sep 2022 09:47:53 +0000 (+0000) Subject: Add sync/async executors for infer request X-Git-Tag: accepted/tizen/unified/20230127.161057~1^2~139^2~6 X-Git-Url: http://review.tizen.org/git/?a=commitdiff_plain;h=2af081363461c09da5ce015df8a154d37cbfd7e0;p=platform%2Fupstream%2Fopencv.git Add sync/async executors for infer request --- diff --git a/modules/gapi/src/backends/ie/giebackend.cpp b/modules/gapi/src/backends/ie/giebackend.cpp index be44d36..ccb5953 100644 --- a/modules/gapi/src/backends/ie/giebackend.cpp +++ b/modules/gapi/src/backends/ie/giebackend.cpp @@ -826,8 +826,10 @@ std::vector cv::gimpl::ie::IECompiled::createInfe return requests; } -class cv::gimpl::ie::RequestPool { +class IInferExecutor { public: + using Ptr = std::shared_ptr; + using NotifyCallbackF = std::function; using SetInputDataF = std::function; using ReadOutputDataF = std::function; @@ -835,41 +837,106 @@ public: // SetInputDataF - function which set input data. // ReadOutputDataF - function which read output data. struct Task { - SetInputDataF set_input_data; + SetInputDataF set_input_data; ReadOutputDataF read_output_data; }; - explicit RequestPool(std::vector&& requests); + IInferExecutor(IE::InferRequest request, NotifyCallbackF notify) + : m_request(std::move(request)), + m_notify(std::move(notify)) { + }; - void execute(Task&& t); - void waitAll(); + virtual void execute(const Task& task) = 0; + virtual ~IInferExecutor() = default; + +protected: + IE::InferRequest m_request; + NotifyCallbackF m_notify; +}; + +class SyncInferExecutor : public IInferExecutor { + using IInferExecutor::IInferExecutor; + virtual void execute(const IInferExecutor::Task& task) override; +}; + +void SyncInferExecutor::execute(const IInferExecutor::Task& task) { + try { + task.set_input_data(m_request); + m_request.Infer(); + task.read_output_data(m_request, IE::StatusCode::OK); + } catch (...) { + m_notify(); + throw; + } + // NB: Notify pool that executor has finished. + m_notify(); +} + +class AsyncInferExecutor : public IInferExecutor { +public: + using IInferExecutor::IInferExecutor; + virtual void execute(const IInferExecutor::Task& task) override; private: void callback(Task task, - size_t id, IE::InferRequest request, IE::StatusCode code) noexcept; +}; + +void AsyncInferExecutor::execute(const IInferExecutor::Task& task) { + using namespace std::placeholders; + using callback_t = std::function; + m_request.SetCompletionCallback( + static_cast( + std::bind(&AsyncInferExecutor::callback, this, task, _1, _2))); + try { + task.set_input_data(m_request); + m_request.StartAsync(); + } catch (...) { + m_request.SetCompletionCallback([](){}); + m_notify(); + throw; + } +} + +void AsyncInferExecutor::callback(IInferExecutor::Task task, + IE::InferRequest request, + IE::StatusCode code) noexcept { + task.read_output_data(request, code); + request.SetCompletionCallback([](){}); + // NB: Notify pool that executor has finished. + m_notify(); +} + +class cv::gimpl::ie::RequestPool { +public: + + explicit RequestPool(std::vector&& requests); + + IInferExecutor::Ptr getIdleRequest(); + void waitAll(); + +private: void setup(); - void releaseRequest(const int id); + void release(const int id); - QueueClass m_idle_ids; - std::vector m_requests; - bool m_use_sync_api = false; + QueueClass m_idle_ids; + std::vector m_requests; }; -void cv::gimpl::ie::RequestPool::releaseRequest(const int id) { - if (!m_use_sync_api) { - auto& request = m_requests[id]; - request.SetCompletionCallback([](){}); - } +void cv::gimpl::ie::RequestPool::release(const int id) { m_idle_ids.push(id); } // RequestPool implementation ////////////////////////////////////////////// -cv::gimpl::ie::RequestPool::RequestPool(std::vector&& requests) - : m_requests(std::move(requests)) { - setup(); +cv::gimpl::ie::RequestPool::RequestPool(std::vector&& requests) { + for (size_t i = 0; i < requests.size(); ++i) { + m_requests.emplace_back( + std::make_shared(std::move(requests[0]), + std::bind(&RequestPool::release, this, i))); } + setup(); +} void cv::gimpl::ie::RequestPool::setup() { for (size_t i = 0; i < m_requests.size(); ++i) { @@ -877,44 +944,10 @@ void cv::gimpl::ie::RequestPool::setup() { } } -void cv::gimpl::ie::RequestPool::execute(cv::gimpl::ie::RequestPool::Task&& task) { +IInferExecutor::Ptr cv::gimpl::ie::RequestPool::getIdleRequest() { size_t id = 0u; m_idle_ids.pop(id); - auto& request = m_requests[id]; - - try { - task.set_input_data(request); - if (m_use_sync_api) { - request.Infer(); - task.read_output_data(request, IE::StatusCode::OK); - releaseRequest(id); - } else { - using namespace std::placeholders; - using callback_t = std::function; - request.SetCompletionCallback( - static_cast( - std::bind(&cv::gimpl::ie::RequestPool::callback, this, - task, id, _1, _2))); - request.StartAsync(); - } - } catch (...) { - // NB: InferRequest is already marked as busy - // in case of exception need to return it back to the idle. - releaseRequest(id); - throw; - } -} - -void cv::gimpl::ie::RequestPool::callback(cv::gimpl::ie::RequestPool::Task task, - size_t id, - IE::InferRequest request, - IE::StatusCode code) noexcept { - // NB: Inference is over. - // 1. Run callback - // 2. Destroy callback to free resources. - // 3. Mark InferRequest as idle. - task.read_output_data(request, code); - releaseRequest(id); + return m_requests[id]; } // NB: Not thread-safe. @@ -1330,8 +1363,8 @@ struct Infer: public cv::detail::KernelTag { static void run(std::shared_ptr ctx, cv::gimpl::ie::RequestPool &reqPool) { using namespace std::placeholders; - reqPool.execute( - cv::gimpl::ie::RequestPool::Task { + reqPool.getIdleRequest()->execute( + IInferExecutor::Task { [ctx](InferenceEngine::InferRequest &req) { // non-generic version for now: // - assumes all inputs/outputs are always Mats @@ -1440,8 +1473,8 @@ struct InferROI: public cv::detail::KernelTag { static void run(std::shared_ptr ctx, cv::gimpl::ie::RequestPool &reqPool) { using namespace std::placeholders; - reqPool.execute( - cv::gimpl::ie::RequestPool::Task { + reqPool.getIdleRequest()->execute( + IInferExecutor::Task { [ctx](InferenceEngine::InferRequest &req) { GAPI_Assert(ctx->uu.params.num_in == 1); auto&& this_roi = ctx->inArg(0).rref(); @@ -1579,8 +1612,8 @@ struct InferList: public cv::detail::KernelTag { for (auto&& it : ade::util::indexed(in_roi_vec)) { auto pos = ade::util::index(it); const auto& rc = ade::util::value(it); - reqPool.execute( - cv::gimpl::ie::RequestPool::Task { + reqPool.getIdleRequest()->execute( + IInferExecutor::Task { [ctx, rc, this_blob](InferenceEngine::InferRequest &req) { setROIBlob(req, ctx->uu.params.input_names[0u], this_blob, rc, *ctx); }, @@ -1734,8 +1767,8 @@ struct InferList2: public cv::detail::KernelTag { PostOutputsList callback(list_size, ctx, std::move(cached_dims)); for (const auto &list_idx : ade::util::iota(list_size)) { - reqPool.execute( - cv::gimpl::ie::RequestPool::Task { + reqPool.getIdleRequest()->execute( + IInferExecutor::Task { [ctx, list_idx, list_size, blob_0](InferenceEngine::InferRequest &req) { for (auto in_idx : ade::util::iota(ctx->uu.params.num_in)) { const auto &this_vec = ctx->inArg(in_idx+1u);