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

Merge pull request #18793 from dmatveev:dm/in_graph_metadata

Browse Source 
G-API: Introduce runtime in-graph metadata

* G-API: In-graph metadata -- initial implementation

* G-API: Finish the in-graph metadata implementation for Streaming

* G-API: Fix standalone build & warnings for in-graph metadata

* G-API: In-graph meta -- fixed review comments

* G-API: Fix issues with desync causing failing tests
This commit is contained in:
Dmitry Matveev
2020-11-17 17:04:19 +03:00
committed by GitHub
parent b5c162175b
commit b866d0dc38
16 changed files with 681 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -350,16 +350,14 @@ bool QueueReader::getInputVector(std::vector<Q*> &in_queues,
// value-initialized scalar)
// It can also hold a constant value received with
// Stop::Kind::CNST message (see above).
// FIXME: Variant move problem
isl_inputs[id] = const_cast<const cv::GRunArg&>(in_constants[id]);
isl_inputs[id] = in_constants[id];
continue;
}
q->pop(m_cmd[id]);
if (!cv::util::holds_alternative<Stop>(m_cmd[id]))
{
// FIXME: Variant move problem
isl_inputs[id] = const_cast<const cv::GRunArg &>(cv::util::get<cv::GRunArg>(m_cmd[id]));
isl_inputs[id] = cv::util::get<cv::GRunArg>(m_cmd[id]);
}
else // A Stop sign
{
@@ -382,7 +380,7 @@ bool QueueReader::getInputVector(std::vector<Q*> &in_queues,
// NEXT time (on a next call to getInputVector()), the
// "q==nullptr" check above will be triggered, but now
// we need to make it manually:
isl_inputs[id] = const_cast<const cv::GRunArg&>(in_constants[id]);
isl_inputs[id] = in_constants[id];
}
else
{
@@ -666,8 +664,7 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
Cmd cmd;
if (cv::util::holds_alternative<cv::GRunArg>(post_iter->data))
{
// FIXME: That ugly VARIANT problem
cmd = Cmd{const_cast<const cv::GRunArg&>(cv::util::get<cv::GRunArg>(post_iter->data))};
cmd = Cmd{cv::util::get<cv::GRunArg>(post_iter->data)};
}
else
{
@@ -677,8 +674,7 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
}
for (auto &&q : m_out_queues[out_idx])
{
// FIXME: This ugly VARIANT problem
q->push(const_cast<const Cmd&>(cmd));
q->push(cmd);
}
post_iter = m_postings[out_idx].erase(post_iter);
}
@@ -708,6 +704,15 @@ class StreamingOutput final: public cv::gimpl::GIslandExecutable::IOutput
}
}
}
void meta(const cv::GRunArgP &out, const cv::GRunArg::Meta &m) override
{
const auto it = m_postIdx.find(cv::gimpl::proto::ptr(out));
GAPI_Assert(it != m_postIdx.end());
const auto out_iter = it->second.second;
cv::util::get<cv::GRunArg>(out_iter->data).meta = m;
}
public:
explicit StreamingOutput(const cv::GMetaArgs &metas,
std::vector< std::vector<Q*> > &out_queues,
@@ -769,6 +774,7 @@ void islandActorThread(std::vector<cv::gimpl::RcDesc> in_rcs, //
void collectorThread(std::vector<Q*> in_queues,
std::vector<int> in_mapping,
const std::size_t out_size,
const bool handle_stop,
Q& out_queue)
{
// These flags are static now: regardless if the sync or
@@ -783,9 +789,14 @@ void collectorThread(std::vector<Q*> in_queues,
while (true)
{
cv::GRunArgs this_result(out_size);
if (!qr.getResultsVector(in_queues, in_mapping, out_size, this_result))
const bool ok = qr.getResultsVector(in_queues, in_mapping, out_size, this_result);
if (!ok)
{
out_queue.push(Cmd{Stop{}});
if (handle_stop)
{
out_queue.push(Cmd{Stop{}});
}
// Terminate the thread anyway
return;
}
out_queue.push(Cmd{Result{std::move(this_result), flags}});
@@ -1263,12 +1274,22 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
// If there are desynchronized parts in the graph, there may be
// multiple theads polling every separate (desynchronized)
// branch in the graph individually.
const bool has_main_path = m_sink_sync.end() !=
std::find(m_sink_sync.begin(), m_sink_sync.end(), -1);
for (auto &&info : m_collector_map) {
m_threads.emplace_back(collectorThread,
info.second.queues,
info.second.mapping,
m_sink_queues.size(),
has_main_path ? info.first == -1 : true, // see below (*)
std::ref(m_out_queue));
// (*) - there may be a problem with desynchronized paths when those work
// faster than the main path. In this case, the desync paths get "Stop" message
// earlier and thus broadcast it down to pipeline gets stopped when there is
// some "main path" data to process. This new collectorThread's flag regulates it:
// - desync paths should never post Stop message if there is a main path.
// - if there is no main path, than any desync path can terminate the execution.
}
state = State::READY;
}