potok/opencv
1
0
Fork 0
mirror of https://github.com/opencv/opencv.git synced 2026-07-29 15:23:05 +04:00
Code Packages 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
modules/gapi/src/executor/gexecutor.cpp
+47 -6
View File
@@ -12,6 +12,8 @@
#include <ade/util/zip_range.hpp>
#include <opencv2/gapi/opencv_includes.hpp>
#include "api/gproto_priv.hpp" // ptr(GRunArgP)
#include "executor/gexecutor.hpp"
#include "compiler/passes/passes.hpp"
@@ -105,6 +107,9 @@ void bindInArgExec(Mag& mag, const RcDesc &rc, const GRunArg &arg)
mag_rmat = util::get<cv::RMat>(arg); break;
default: util::throw_error(std::logic_error("content type of the runtime argument does not match to resource description ?"));
}
// FIXME: has to take extra care about meta here for this particuluar
// case, just because this function exists at all
mag.meta<cv::RMat>()[rc.id] = arg.meta;
}
void bindOutArgExec(Mag& mag, const RcDesc &rc, const GRunArgP &arg)
@@ -131,7 +136,7 @@ cv::GRunArgP getObjPtrExec(Mag& mag, const RcDesc &rc)
{
return getObjPtr(mag, rc);
}
return GRunArgP(&mag.template slot<cv::RMat>()[rc.id]);
return GRunArgP(&mag.slot<cv::RMat>()[rc.id]);
}
void writeBackExec(const Mag& mag, const RcDesc &rc, GRunArgP &g_arg)
@@ -155,6 +160,25 @@ void writeBackExec(const Mag& mag, const RcDesc &rc, GRunArgP &g_arg)
default: util::throw_error(std::logic_error("content type of the runtime argument does not match to resource description ?"));
}
}
void assignMetaStubExec(Mag& mag, const RcDesc &rc, const cv::GRunArg::Meta &meta) {
switch (rc.shape)
{
case GShape::GARRAY: mag.meta<cv::detail::VectorRef>()[rc.id] = meta; break;
case GShape::GOPAQUE: mag.meta<cv::detail::OpaqueRef>()[rc.id] = meta; break;
case GShape::GSCALAR: mag.meta<cv::Scalar>()[rc.id] = meta; break;
case GShape::GFRAME: mag.meta<cv::MediaFrame>()[rc.id] = meta; break;
case GShape::GMAT:
mag.meta<cv::Mat>() [rc.id] = meta;
mag.meta<cv::RMat>()[rc.id] = meta;
#if !defined(GAPI_STANDALONE)
mag.meta<cv::UMat>()[rc.id] = meta;
#endif
break;
default: util::throw_error(std::logic_error("Unsupported GShape type")); break;
}
}
} // anonymous namespace
}}} // namespace cv::gimpl::magazine
@@ -231,11 +255,28 @@ public:
class cv::gimpl::GExecutor::Output final: public cv::gimpl::GIslandExecutable::IOutput
{
cv::gimpl::Mag &mag;
virtual GRunArgP get(int idx) override { return magazine::getObjPtrExec(mag, desc()[idx]); }
virtual void post(GRunArgP&&) override { } // Do nothing here
virtual void post(EndOfStream&&) override {} // Do nothing here too
std::unordered_map<const void*, int> out_idx;
GRunArgP get(int idx) override
{
auto r = magazine::getObjPtrExec(mag, desc()[idx]);
// Remember the output port for this output object
out_idx[cv::gimpl::proto::ptr(r)] = idx;
return r;
}
void post(GRunArgP&&) override { } // Do nothing here
void post(EndOfStream&&) override {} // Do nothing here too
void meta(const GRunArgP &out, const GRunArg::Meta &m) override
{
const auto idx = out_idx.at(cv::gimpl::proto::ptr(out));
magazine::assignMetaStubExec(mag, desc()[idx], m);
}
public:
Output(cv::gimpl::Mag &m, const std::vector<RcDesc> &rcs) : mag(m) { set(rcs); }
Output(cv::gimpl::Mag &m, const std::vector<RcDesc> &rcs)
: mag(m)
{
set(rcs);
}
};
void cv::gimpl::GExecutor::run(cv::gimpl::GRuntimeArgs &&args)
@@ -330,7 +371,7 @@ void cv::gimpl::GExecutor::run(cv::gimpl::GRuntimeArgs &&args)
// Run the script
for (auto &op : m_ops)
{
// (5)
// (5), (6)
Input i{m_res, op.in_objects};
Output o{m_res, op.out_objects};
op.isl_exec->run(i, o);
modules/gapi/src/executor/gstreamingexecutor.cpp
+32 -11
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;
}