Administrator/opencv-MIRROR
1
0
Fork 0
mirror of https://github.com/opencv/opencv.git synced 2026-09-13 05:42:51 -05:00
Code Issues Packages Projects Releases Wiki Activity

Added ITT traces to GStreamingExecutor

Browse Source
This commit is contained in:
Anastasiya Pronina
2021-04-12 22:20:59 +03:00
parent 7de627c504
commit 3a49ff9e72
11 changed files with 272 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
modules/gapi/src/executor/gstreamingexecutor.cpp
View File

@@ -2,7 +2,7 @@
// 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) 2019-2020 Intel Corporation
// Copyright (C) 2019-2021 Intel Corporation
#include "precomp.hpp"
@@ -16,6 +16,8 @@
#include <opencv2/gapi/core.hpp> // GCopy -- FIXME - to be removed!
#endif // GAPI_STANDALONE
#include "utils/itt.hpp"
#include "api/gproto_priv.hpp" // ptr(GRunArgP)
#include "compiler/passes/passes.hpp"
#include "backends/common/gbackend.hpp" // createMat
@@ -492,9 +494,15 @@ void emitterActorThread(std::shared_ptr<cv::gimpl::GIslandEmitter> emitter,
return;
}
GAPI_ITT_STATIC_LOCAL_HANDLE(emitter_hndl, "emitter");
GAPI_ITT_STATIC_LOCAL_HANDLE(emitter_pull_hndl, "emitter_pull");
GAPI_ITT_STATIC_LOCAL_HANDLE(emitter_push_hndl, "emitter_push");
// Now start emitting the data from the source to the pipeline.
while (true)
{
GAPI_ITT_AUTO_TRACE_GUARD(emitter_hndl);
Cmd cancel;
if (in_queue.try_pop(cancel))
{
@@ -507,8 +515,15 @@ void emitterActorThread(std::shared_ptr<cv::gimpl::GIslandEmitter> emitter,
// Try to obtain next data chunk from the source
cv::GRunArg data;
if (emitter->pull(data))
const bool result = [&](){
GAPI_ITT_AUTO_TRACE_GUARD(emitter_pull_hndl);
return emitter->pull(data);
}();
if (result)
{
GAPI_ITT_AUTO_TRACE_GUARD(emitter_push_hndl);
// // On success, broadcast it to our readers
for (auto &&oq : out_queues)
{
@@ -539,7 +554,11 @@ void syncActorThread(std::vector<Q*> in_queues,
std::vector<bool> pop_nexts(in_queues.size());
std::vector<Cmd> cmds(in_queues.size());
GAPI_ITT_STATIC_LOCAL_HANDLE(sync_hndl, "sync_actor");
GAPI_ITT_STATIC_LOCAL_HANDLE(sync_pull_1_queue_hndl, "sync_actor_pull_from_1_queue");
GAPI_ITT_STATIC_LOCAL_HANDLE(sync_push_hndl, "sync_actor_push");
while (true) {
GAPI_ITT_AUTO_TRACE_GUARD(sync_hndl);
// pop_nexts indicates which queue still contains earlier timestamps and
// needs to be popped at least one more time.
// For each iteration (frame) we need to pull from each input queue at least once,
@@ -562,7 +581,10 @@ void syncActorThread(std::vector<Q*> in_queues,
auto& q = std::get<1>(val);
auto& cmd = std::get<2>(val);
q->pop(cmd);
{
GAPI_ITT_AUTO_TRACE_GUARD(sync_pull_1_queue_hndl);
q->pop(cmd);
}
if (cv::util::holds_alternative<Stop>(cmd)) {
// We got a stop command from one of the input queues.
// Rewind all input queues till Stop command,
@@ -603,9 +625,12 @@ void syncActorThread(std::vector<Q*> in_queues,
} while (ade::util::any_of(pop_nexts, [](bool v){ return v; }));
// Finally we got all our inputs synchronized, push them further down the graph
for (auto &&it : ade::util::zip(out_queues, cmds)) {
for (auto &&q : std::get<0>(it)) {
q->push(std::get<1>(it));
{
GAPI_ITT_AUTO_TRACE_GUARD(sync_push_hndl);
for (auto &&it : ade::util::zip(out_queues, cmds)) {
for (auto &&q : std::get<0>(it)) {
q->push(std::get<1>(it));
}
}
}
}
@@ -619,7 +644,11 @@ class StreamingInput final: public cv::gimpl::GIslandExecutable::IInput
virtual cv::gimpl::StreamMsg get() override
{
GAPI_ITT_STATIC_LOCAL_HANDLE(inputs_get_hndl, "StreamingInput::get");
GAPI_ITT_AUTO_TRACE_GUARD(inputs_get_hndl);
cv::GRunArgs isl_input_args;
if (!qr.getInputVector(in_queues, in_constants, isl_input_args))
{
// Stop case
@@ -680,6 +709,9 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
// Prepare this object for posting
virtual cv::GRunArgP get(int idx) override
{
GAPI_ITT_STATIC_LOCAL_HANDLE(outputs_get_hndl, "StreamingOutput::get (alloc)");
GAPI_ITT_AUTO_TRACE_GUARD(outputs_get_hndl);
std::lock_guard<std::mutex> lock{m_mutex};
using MatType = cv::Mat;
@@ -756,8 +788,12 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
m_postIdx[cv::gimpl::proto::ptr(ret_val)] = std::make_pair(idx, iter);
return ret_val;
}
virtual void post(cv::GRunArgP&& argp) override
{
GAPI_ITT_STATIC_LOCAL_HANDLE(outputs_post_hndl, "StreamingOutput::post");
GAPI_ITT_AUTO_TRACE_GUARD(outputs_post_hndl);
std::lock_guard<std::mutex> lock{m_mutex};
// Mark the output ready for posting. If it is the first in the line,
@@ -795,6 +831,7 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
post_iter = m_postings[out_idx].erase(post_iter);
}
}
virtual void post(cv::gimpl::EndOfStream&&) override
{
std::lock_guard<std::mutex> lock{m_mutex};
@@ -859,23 +896,27 @@ public:
// executable for processing.
// - Pushes processing results down to consumers - to the subsequent queues.
// Note: Every data object consumer has its own queue.
void islandActorThread(std::vector<cv::gimpl::RcDesc> in_rcs, // FIXME: this is...
std::vector<cv::gimpl::RcDesc> out_rcs, // FIXME: ...basically just...
cv::GMetaArgs out_metas, // ...
std::shared_ptr<cv::gimpl::GIslandExecutable> island, // FIXME: ...a copy of OpDesc{}.
void islandActorThread(std::vector<cv::gimpl::RcDesc> in_rcs, // FIXME: this is...
std::vector<cv::gimpl::RcDesc> out_rcs, // FIXME: ...basically just...
cv::GMetaArgs out_metas, // ...
std::shared_ptr<cv::gimpl::GIslandExecutable> island_exec, // FIXME: ...a copy of OpDesc{}.
std::vector<Q*> in_queues,
cv::GRunArgs in_constants,
std::vector< std::vector<Q*> > out_queues)
std::vector< std::vector<Q*> > out_queues,
const std::string& island_meta_info)
{
GAPI_Assert(in_queues.size() == in_rcs.size());
GAPI_Assert(out_queues.size() == out_rcs.size());
GAPI_Assert(out_queues.size() == out_metas.size());
QueueReader qr;
StreamingInput input(qr, in_queues, in_constants, in_rcs);
StreamingOutput output(out_metas, out_queues, out_rcs, island);
StreamingOutput output(out_metas, out_queues, out_rcs, island_exec);
GAPI_ITT_DYNAMIC_LOCAL_HANDLE(island_hndl, island_meta_info.c_str());
while (!output.done())
{
island->run(input, output);
GAPI_ITT_AUTO_TRACE_GUARD(island_hndl);
island_exec->run(input, output);
}
}
@@ -904,11 +945,21 @@ void collectorThread(std::vector<Q*> in_queues,
flags[idx] = true;
}
GAPI_ITT_STATIC_LOCAL_HANDLE(collector_hndl, "collector");
GAPI_ITT_STATIC_LOCAL_HANDLE(collector_get_results_hndl, "collector_get_results");
GAPI_ITT_STATIC_LOCAL_HANDLE(collector_push_hndl, "collector_push");
QueueReader qr;
while (true)
{
GAPI_ITT_AUTO_TRACE_GUARD(collector_hndl);
cv::GRunArgs this_result(out_size);
const bool ok = qr.getResultsVector(in_queues, in_mapping, out_size, this_result);
const bool ok = [&](){
GAPI_ITT_AUTO_TRACE_GUARD(collector_get_results_hndl);
return qr.getResultsVector(in_queues, in_mapping, out_size, this_result);
}();
if (!ok)
{
if (handle_stop)
@@ -918,7 +969,11 @@ void collectorThread(std::vector<Q*> in_queues,
// Terminate the thread anyway
return;
}
out_queue.push(Cmd{Result{std::move(this_result), flags}});
{
GAPI_ITT_AUTO_TRACE_GUARD(collector_push_hndl);
out_queue.push(Cmd{Result{std::move(this_result), flags}});
}
}
}
@@ -1379,11 +1434,8 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
m_sync->registerVideoEmitters(std::move(video_emitters));
// FIXME: The below code assumes our graph may have only one
// real video source (and so, only one stream which may really end)
// all other inputs are "constant" generators.
// Craft here a completion callback to notify Const emitters that
// a video source is over
// any of video sources is over
GAPI_Assert(m_const_emitter_queues.size() == m_const_vals.size());
auto real_video_completion_cb = [this]()
{
@@ -1431,7 +1483,7 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
for (auto &&op : m_ops)
{
// Prepare island thread parameters
auto island = m_gim.metadata(op.nh).get<IslandExec>().object;
auto island_exec = m_gim.metadata(op.nh).get<IslandExec>().object;
// Collect actor's input queues
auto in_queues = input_queues(*m_island_graph, op.nh);
@@ -1443,6 +1495,13 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
out_queues.push_back(reader_queues(*m_island_graph, out_eh));
}
// Create just empty island meta information
std::string island_meta_info { };
#if defined(OPENCV_WITH_ITT)
// In case if ITT tracing is enabled fill meta information with the built island name
island_meta_info = GIslandModel::traceIslandName(op.nh, m_gim);
#endif // OPENCV_WITH_ITT
// If Island Executable is recompiled, all its stuff including internal kernel states
// are recreated and re-initialized automatically.
// But if not, we should notify Island Executable about new started stream to let it update
@@ -1456,10 +1515,11 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
op.in_objects,
op.out_objects,
op.out_metas,
island,
island_exec,
in_queues,
op.in_constants,
out_queues);
out_queues,
island_meta_info);
}
// Finally, start collector thread(s).
@@ -1536,6 +1596,9 @@ void cv::gimpl::GStreamingExecutor::wait_shutdown()
bool cv::gimpl::GStreamingExecutor::pull(cv::GRunArgsP &&outs)
{
GAPI_ITT_STATIC_LOCAL_HANDLE(pull_hndl, "GStreamingExecutor::pull");
GAPI_ITT_AUTO_TRACE_GUARD(pull_hndl);
// This pull() can only be called when there's no desynchronized
// parts in the graph.
GAPI_Assert(!m_desync &&