mirror of
https://github.com/opencv/opencv.git
synced 2026-07-29 15:23:05 +04:00
G-API: Introduce streaming::desync and infer(ROI)
- desync() is a new (and for now, the only one) intrinsic which splits the graph execution into asynchronous parts when running in Streaming mode; - desync() makes no effect when compiling in Traditional mode; - Added tests on desync() working in various scenarios; - Extended GStreamingExecutor to support desync(); also extended GStreamingCompiled() with a new version of pull() returning a vector of optional values; - Fixed various issues with storing the type information & proper construction callbacks for GArray<> and GOpaque; - Introduced a new infer(Roi,GMat) overload with a sample; - Introduced an internal API for Islands to control fusion procedure (to fuse or not to fuse); - Introduced handleStopStream() callback for island executables; - Added GCompileArgs to metadata of the graph (required for other features).
This commit is contained in:
Dmitry Matveev
@@ -119,8 +119,7 @@ void concurrent_bounded_queue<T>::set_capacity(std::size_t capacity) {
|
||||
// Clear the queue. Similar to the TBB version, this method is not
|
||||
// thread-safe.
|
||||
template<typename T>
|
||||
void concurrent_bounded_queue<T>::clear()
|
||||
{
|
||||
void concurrent_bounded_queue<T>::clear() {
|
||||
m_data = std::queue<T>{};
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
|
||||
#include "precomp.hpp"
|
||||
|
||||
#include <memory> // make_shared
|
||||
#include <iostream>
|
||||
|
||||
#include <ade/util/zip_range.hpp>
|
||||
@@ -60,14 +61,23 @@ public:
|
||||
|
||||
struct DataQueue {
|
||||
static const char *name() { return "StreamingDataQueue"; }
|
||||
enum tag { DESYNC }; // Enum of 1 element: purely a syntax sugar
|
||||
|
||||
explicit DataQueue(std::size_t capacity) {
|
||||
if (capacity) {
|
||||
q.set_capacity(capacity);
|
||||
// Note: `ptr` is shared<SyncQueue>, while the `q` is a shared<Q>
|
||||
auto ptr = std::make_shared<cv::gimpl::stream::SyncQueue>();
|
||||
if (capacity != 0) {
|
||||
ptr->set_capacity(capacity);
|
||||
}
|
||||
q = std::move(ptr);
|
||||
}
|
||||
explicit DataQueue(tag t)
|
||||
: q(new cv::gimpl::stream::DesyncQueue()) {
|
||||
GAPI_Assert(t == DESYNC);
|
||||
}
|
||||
|
||||
cv::gimpl::stream::Q q;
|
||||
// FIXME: ADE metadata requires types to be copiable
|
||||
std::shared_ptr<cv::gimpl::stream::Q> q;
|
||||
};
|
||||
|
||||
std::vector<cv::gimpl::stream::Q*> reader_queues( ade::Graph &g,
|
||||
@@ -77,7 +87,7 @@ std::vector<cv::gimpl::stream::Q*> reader_queues( ade::Graph &g,
|
||||
std::vector<cv::gimpl::stream::Q*> result;
|
||||
for (auto &&out_eh : obj->outEdges())
|
||||
{
|
||||
result.push_back(&qgr.metadata(out_eh).get<DataQueue>().q);
|
||||
result.push_back(qgr.metadata(out_eh).get<DataQueue>().q.get());
|
||||
}
|
||||
return result;
|
||||
}
|
||||
@@ -90,7 +100,7 @@ std::vector<cv::gimpl::stream::Q*> input_queues( ade::Graph &g,
|
||||
for (auto &&in_eh : obj->inEdges())
|
||||
{
|
||||
result.push_back(qgr.metadata(in_eh).contains<DataQueue>()
|
||||
? &qgr.metadata(in_eh).get<DataQueue>().q
|
||||
? qgr.metadata(in_eh).get<DataQueue>().q.get()
|
||||
: nullptr);
|
||||
}
|
||||
return result;
|
||||
@@ -133,6 +143,77 @@ void sync_data(cv::GRunArgs &results, cv::GRunArgsP &outputs)
|
||||
}
|
||||
}
|
||||
|
||||
// FIXME: Is there a way to derive function from its GRunArgsP version?
|
||||
template<class C> using O = cv::util::optional<C>;
|
||||
void sync_data(cv::gimpl::stream::Result &r, cv::GOptRunArgsP &outputs)
|
||||
{
|
||||
namespace own = cv::gapi::own;
|
||||
|
||||
for (auto && it : ade::util::zip(ade::util::toRange(outputs),
|
||||
ade::util::toRange(r.args),
|
||||
ade::util::toRange(r.flags)))
|
||||
{
|
||||
auto &out_obj = std::get<0>(it);
|
||||
auto &res_obj = std::get<1>(it);
|
||||
bool available = std::get<2>(it);
|
||||
|
||||
using T = cv::GOptRunArgP;
|
||||
#define HANDLE_CASE(Type) \
|
||||
case T::index_of<O<Type>*>(): \
|
||||
if (available) { \
|
||||
*cv::util::get<O<Type>*>(out_obj) \
|
||||
= cv::util::make_optional(std::move(cv::util::get<Type>(res_obj))); \
|
||||
} else { \
|
||||
cv::util::get<O<Type>*>(out_obj)->reset(); \
|
||||
}
|
||||
|
||||
// FIXME: this conversion should be unified
|
||||
switch (out_obj.index())
|
||||
{
|
||||
HANDLE_CASE(cv::Scalar); break;
|
||||
HANDLE_CASE(cv::RMat); break;
|
||||
|
||||
case T::index_of<O<cv::Mat>*>(): {
|
||||
// Mat: special handling.
|
||||
auto &mat_opt = *cv::util::get<O<cv::Mat>*>(out_obj);
|
||||
if (available) {
|
||||
auto q_map = cv::util::get<cv::RMat>(res_obj).access(cv::RMat::Access::R);
|
||||
// FIXME: Copy! Maybe we could do some optimization for this case!
|
||||
// e.g. don't handle RMat for last ilsand in the graph.
|
||||
// It is not always possible though.
|
||||
mat_opt = cv::util::make_optional(cv::gimpl::asMat(q_map).clone());
|
||||
} else {
|
||||
mat_opt.reset();
|
||||
}
|
||||
} break;
|
||||
case T::index_of<cv::detail::OptionalVectorRef>(): {
|
||||
// std::vector<>: special handling
|
||||
auto &vec_opt = cv::util::get<cv::detail::OptionalVectorRef>(out_obj);
|
||||
if (available) {
|
||||
vec_opt.mov(cv::util::get<cv::detail::VectorRef>(res_obj));
|
||||
} else {
|
||||
vec_opt.reset();
|
||||
}
|
||||
} break;
|
||||
case T::index_of<cv::detail::OptionalOpaqueRef>(): {
|
||||
// std::vector<>: special handling
|
||||
auto &opq_opt = cv::util::get<cv::detail::OptionalOpaqueRef>(out_obj);
|
||||
if (available) {
|
||||
opq_opt.mov(cv::util::get<cv::detail::OpaqueRef>(res_obj));
|
||||
} else {
|
||||
opq_opt.reset();
|
||||
}
|
||||
} break;
|
||||
default:
|
||||
// ...maybe because of STANDALONE mode.
|
||||
GAPI_Assert(false && "This value type is not supported!");
|
||||
break;
|
||||
}
|
||||
}
|
||||
#undef HANDLE_CASE
|
||||
}
|
||||
|
||||
|
||||
// Pops an item from every input queue and combine it to the final
|
||||
// result. Blocks the current thread. Returns true if the vector has
|
||||
// been obtained successfully and false if a Stop message has been
|
||||
@@ -206,12 +287,39 @@ class QueueReader
|
||||
bool m_finishing = false; // Set to true once a "soft" stop is received
|
||||
std::vector<Cmd> m_cmd;
|
||||
|
||||
void rewindToStop(std::vector<Q*> &in_queues,
|
||||
const std::size_t this_id);
|
||||
|
||||
public:
|
||||
bool getInputVector(std::vector<Q*> &in_queues,
|
||||
cv::GRunArgs &in_constants,
|
||||
cv::GRunArgs &isl_inputs);
|
||||
bool getInputVector (std::vector<Q*> &in_queues,
|
||||
cv::GRunArgs &in_constants,
|
||||
cv::GRunArgs &isl_inputs);
|
||||
|
||||
bool getResultsVector(std::vector<Q*> &in_queues,
|
||||
const std::vector<int> &in_mapping,
|
||||
const std::size_t out_size,
|
||||
cv::GRunArgs &out_results);
|
||||
};
|
||||
|
||||
// This method handles a stop sign got from some input
|
||||
// island. Reiterate through all _remaining valid_ queues (some of
|
||||
// them can be set to nullptr already -- see handling in
|
||||
// getInputVector) and rewind data to every Stop sign per queue.
|
||||
void QueueReader::rewindToStop(std::vector<Q*> &in_queues,
|
||||
const std::size_t this_id)
|
||||
{
|
||||
for (auto &&qit : ade::util::indexed(in_queues))
|
||||
{
|
||||
auto id2 = ade::util::index(qit);
|
||||
auto &q2 = ade::util::value(qit);
|
||||
if (this_id == id2) continue;
|
||||
|
||||
Cmd cmd;
|
||||
while (q2 && !cv::util::holds_alternative<Stop>(cmd))
|
||||
q2->pop(cmd);
|
||||
}
|
||||
}
|
||||
|
||||
bool QueueReader::getInputVector(std::vector<Q*> &in_queues,
|
||||
cv::GRunArgs &in_constants,
|
||||
cv::GRunArgs &isl_inputs)
|
||||
@@ -271,20 +379,7 @@ bool QueueReader::getInputVector(std::vector<Q*> &in_queues,
|
||||
else
|
||||
{
|
||||
GAPI_Assert(stop.kind == Stop::Kind::HARD);
|
||||
// Just got a stop sign. Reiterate through all
|
||||
// _remaining valid_ queues (some of them can be
|
||||
// set to nullptr already -- see above) and rewind
|
||||
// data to every Stop sign per queue
|
||||
for (auto &&qit : ade::util::indexed(in_queues))
|
||||
{
|
||||
auto id2 = ade::util::index(qit);
|
||||
auto &q2 = ade::util::value(qit);
|
||||
if (id == id2) continue;
|
||||
|
||||
Cmd cmd2;
|
||||
while (q2 && !cv::util::holds_alternative<Stop>(cmd2))
|
||||
q2->pop(cmd2);
|
||||
}
|
||||
rewindToStop(in_queues, id);
|
||||
// After queues are read to the proper indicator,
|
||||
// indicate end-of-stream
|
||||
return false;
|
||||
@@ -303,6 +398,60 @@ bool QueueReader::getInputVector(std::vector<Q*> &in_queues,
|
||||
return true; // A regular case - there is data to process.
|
||||
}
|
||||
|
||||
// This is a special method to obtain a result vector
|
||||
// for the entire pipeline's outputs.
|
||||
//
|
||||
// After introducing desync(), the pipeline output's vector
|
||||
// can be produced just partially. Also, if a desynchronized
|
||||
// path has multiple outputs for the pipeline, _these_ outputs
|
||||
// should still come synchronized to the end user (via pull())
|
||||
//
|
||||
//
|
||||
// This method handles all this.
|
||||
// It takes a number of input queues, which may or may not be
|
||||
// equal to the number of pipeline outputs (<=).
|
||||
// It also takes indexes saying which queue produces which
|
||||
// output in the resulting pipeline.
|
||||
//
|
||||
// `out_results` is always produced with the size of full output
|
||||
// vector. In the desync case, the number of in_queues will
|
||||
// be less than this size and some of the items won't be produced.
|
||||
// In the sync case, there will be a 1-1 mapping.
|
||||
//
|
||||
// In the desync case, there _will be_ multiple collector threads
|
||||
// calling this method, and pushing their whole-pipeline outputs
|
||||
// (_may be_ partially filled) to the same final output queue.
|
||||
// The receiver part at the GStreamingExecutor level won't change
|
||||
// because of that.
|
||||
bool QueueReader::getResultsVector(std::vector<Q*> &in_queues,
|
||||
const std::vector<int> &in_mapping,
|
||||
const std::size_t out_size,
|
||||
cv::GRunArgs &out_results)
|
||||
{
|
||||
m_cmd.resize(out_size);
|
||||
for (auto &&it : ade::util::indexed(in_queues))
|
||||
{
|
||||
auto ii = ade::util::index(it);
|
||||
auto oi = in_mapping[ii];
|
||||
auto &q = ade::util::value(it);
|
||||
q->pop(m_cmd[oi]);
|
||||
if (!cv::util::holds_alternative<Stop>(m_cmd[oi]))
|
||||
{
|
||||
out_results[oi] = std::move(cv::util::get<cv::GRunArg>(m_cmd[oi]));
|
||||
}
|
||||
else // A Stop sign
|
||||
{
|
||||
// In theory, the CNST should never reach here.
|
||||
// Collector thread never handles the inputs directly
|
||||
// (collector's input queues are always produced by
|
||||
// islands in the graph).
|
||||
rewindToStop(in_queues, ii);
|
||||
return false;
|
||||
} // if(Stop)
|
||||
} // for(in_queues)
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
// This thread is a plain dump source actor. What it do is just:
|
||||
// - Check input queue (the only one) for a control command
|
||||
@@ -603,22 +752,78 @@ void islandActorThread(std::vector<cv::gimpl::RcDesc> in_rcs, //
|
||||
// and then put the resulting vector into one single queue. While it
|
||||
// looks redundant, it simplifies dramatically the way how try_pull()
|
||||
// is implemented - we need to check one queue instead of many.
|
||||
void collectorThread(std::vector<Q*> in_queues,
|
||||
Q& out_queue)
|
||||
//
|
||||
// After desync() is added, there may be multiple collector threads
|
||||
// running, every thread producing its own part of the partial
|
||||
// pipeline output (optional<T>...). All partial outputs are pushed
|
||||
// to the same output queue and then picked by GStreamingExecutor
|
||||
// in the end.
|
||||
void collectorThread(std::vector<Q*> in_queues,
|
||||
std::vector<int> in_mapping,
|
||||
const std::size_t out_size,
|
||||
Q& out_queue)
|
||||
{
|
||||
// These flags are static now: regardless if the sync or
|
||||
// desync branch is collected by this thread, all in_queue
|
||||
// data should come in sync.
|
||||
std::vector<bool> flags(out_size, false);
|
||||
for (auto idx : in_mapping) {
|
||||
flags[idx] = true;
|
||||
}
|
||||
|
||||
QueueReader qr;
|
||||
while (true)
|
||||
{
|
||||
cv::GRunArgs this_result(in_queues.size());
|
||||
cv::GRunArgs this_const(in_queues.size());
|
||||
if (!qr.getInputVector(in_queues, this_const, this_result))
|
||||
cv::GRunArgs this_result(out_size);
|
||||
if (!qr.getResultsVector(in_queues, in_mapping, out_size, this_result))
|
||||
{
|
||||
out_queue.push(Cmd{Stop{}});
|
||||
return;
|
||||
}
|
||||
out_queue.push(Cmd{this_result});
|
||||
out_queue.push(Cmd{Result{std::move(this_result), flags}});
|
||||
}
|
||||
}
|
||||
|
||||
void check_DesyncObjectConsumedByMultipleIslands(const cv::gimpl::GIslandModel::Graph &gim) {
|
||||
using namespace cv::gimpl;
|
||||
|
||||
// Since the limitation exists only in this particular
|
||||
// implementation, the check is also done only here but not at the
|
||||
// graph compiler level.
|
||||
//
|
||||
// See comment in desync(GMat) src/api/kernels_streaming.cpp for details.
|
||||
for (auto &&nh : gim.nodes()) {
|
||||
if (gim.metadata(nh).get<NodeKind>().k == NodeKind::SLOT) {
|
||||
// SLOTs are read by ISLANDs, so look for the metadata
|
||||
// of the outbound edges
|
||||
std::unordered_map<int, GIsland*> out_desync_islands;
|
||||
for (auto &&out_eh : nh->outEdges()) {
|
||||
if (gim.metadata(out_eh).contains<DesyncIslEdge>()) {
|
||||
// This is a desynchronized edge
|
||||
// Look what Island it leads to
|
||||
const auto out_desync_idx = gim.metadata(out_eh)
|
||||
.get<DesyncIslEdge>().index;
|
||||
const auto out_island = gim.metadata(out_eh->dstNode())
|
||||
.get<FusedIsland>().object;
|
||||
|
||||
auto it = out_desync_islands.find(out_desync_idx);
|
||||
if (it != out_desync_islands.end()) {
|
||||
// If there's already an edge with this desync
|
||||
// id, it must point to the same island object
|
||||
GAPI_Assert(it->second == out_island.get()
|
||||
&& "A single desync object may only be used by a single island!");
|
||||
} else {
|
||||
// Store the island pointer for the further check
|
||||
out_desync_islands[out_desync_idx] = out_island.get();
|
||||
}
|
||||
} // if(desync)
|
||||
} // for(out_eh)
|
||||
// There must be only one backend in the end of the day
|
||||
// (under this desync path)
|
||||
} // if(SLOT)
|
||||
} // for(nodes)
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
// GStreamingExecutor expects compile arguments as input to have possibility to do
|
||||
@@ -630,20 +835,28 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
|
||||
.get<IslandModel>().model)
|
||||
, m_comp_args(comp_args)
|
||||
, m_gim(*m_island_graph)
|
||||
, m_desync(GModel::Graph(*m_orig_graph).metadata()
|
||||
.contains<Desynchronized>())
|
||||
{
|
||||
GModel::Graph gm(*m_orig_graph);
|
||||
// NB: Right now GIslandModel is acyclic, and all the below code assumes that.
|
||||
// NB: This naive execution code is taken from GExecutor nearly "as-is"
|
||||
// NB: This naive execution code is taken from GExecutor nearly
|
||||
// "as-is"
|
||||
|
||||
if (m_desync) {
|
||||
check_DesyncObjectConsumedByMultipleIslands(m_gim);
|
||||
}
|
||||
|
||||
const auto proto = gm.metadata().get<Protocol>();
|
||||
m_emitters .resize(proto.in_nhs.size());
|
||||
m_emitter_queues.resize(proto.in_nhs.size());
|
||||
m_sinks .resize(proto.out_nhs.size());
|
||||
m_sink_queues .resize(proto.out_nhs.size());
|
||||
m_sink_queues .resize(proto.out_nhs.size(), nullptr);
|
||||
m_sink_sync .resize(proto.out_nhs.size(), -1);
|
||||
|
||||
// Very rough estimation to limit internal queue sizes.
|
||||
// Pipeline depth is equal to number of its (pipeline) steps.
|
||||
const auto queue_capacity = std::count_if
|
||||
const auto queue_capacity = 3*std::count_if
|
||||
(m_gim.nodes().begin(),
|
||||
m_gim.nodes().end(),
|
||||
[&](ade::NodeHandle nh) {
|
||||
@@ -728,8 +941,12 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
|
||||
{
|
||||
// ...only if the data is not compile-const
|
||||
if (const_ins.count(eh->srcNode()) == 0) {
|
||||
qgr.metadata(eh).set(DataQueue(queue_capacity));
|
||||
m_internal_queues.insert(&qgr.metadata(eh).get<DataQueue>().q);
|
||||
if (m_gim.metadata(eh).contains<DesyncIslEdge>()) {
|
||||
qgr.metadata(eh).set(DataQueue(DataQueue::DESYNC));
|
||||
} else {
|
||||
qgr.metadata(eh).set(DataQueue(queue_capacity));
|
||||
}
|
||||
m_internal_queues.insert(qgr.metadata(eh).get<DataQueue>().q.get());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -760,7 +977,14 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
|
||||
ade::TypedGraph<DataQueue> qgr(*m_island_graph);
|
||||
GAPI_Assert(nh->inEdges().size() == 1u);
|
||||
qgr.metadata(nh->inEdges().front()).set(DataQueue(queue_capacity));
|
||||
m_sink_queues[sink_idx] = &qgr.metadata(nh->inEdges().front()).get<DataQueue>().q;
|
||||
m_sink_queues[sink_idx] = qgr.metadata(nh->inEdges().front()).get<DataQueue>().q.get();
|
||||
|
||||
// Assign a desync tag
|
||||
const auto sink_out_nh = gm.metadata().get<Protocol>().out_nhs[sink_idx];
|
||||
if (gm.metadata(sink_out_nh).contains<DesyncPath>()) {
|
||||
// metadata().get_or<> could make this thing better
|
||||
m_sink_sync[sink_idx] = gm.metadata(sink_out_nh).get<DesyncPath>().index;
|
||||
}
|
||||
}
|
||||
break;
|
||||
default:
|
||||
@@ -768,7 +992,23 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
|
||||
break;
|
||||
} // switch(kind)
|
||||
} // for(gim nodes)
|
||||
m_out_queue.set_capacity(queue_capacity);
|
||||
|
||||
// If there are desynchronized parts in the graph, there may be
|
||||
// multiple theads polling every separate (desynchronized)
|
||||
// branch in the graph individually. Prepare a mapping information
|
||||
// for any such thread
|
||||
for (auto &&idx : ade::util::iota(m_sink_queues.size())) {
|
||||
auto path_id = m_sink_sync[idx];
|
||||
auto &info = m_collector_map[path_id];
|
||||
info.queues.push_back(m_sink_queues[idx]);
|
||||
info.mapping.push_back(static_cast<int>(idx));
|
||||
}
|
||||
|
||||
// Reserve space in the final queue based on the number
|
||||
// of desync parts (they can generate output individually
|
||||
// per the same input frame, so the output traffic multiplies)
|
||||
GAPI_Assert(m_collector_map.size() > 0u);
|
||||
m_out_queue.set_capacity(queue_capacity * m_collector_map.size());
|
||||
}
|
||||
|
||||
cv::gimpl::GStreamingExecutor::~GStreamingExecutor()
|
||||
@@ -938,6 +1178,9 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
|
||||
real_video_completion_cb);
|
||||
}
|
||||
|
||||
for (auto &&op : m_ops) {
|
||||
op.isl_exec->handleNewStream();
|
||||
}
|
||||
|
||||
// Now do this for every island (in a topological order)
|
||||
for (auto &&op : m_ops)
|
||||
@@ -974,10 +1217,17 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
|
||||
out_queues);
|
||||
}
|
||||
|
||||
// Finally, start a collector thread.
|
||||
m_threads.emplace_back(collectorThread,
|
||||
m_sink_queues,
|
||||
std::ref(m_out_queue));
|
||||
// Finally, start collector thread(s).
|
||||
// If there are desynchronized parts in the graph, there may be
|
||||
// multiple theads polling every separate (desynchronized)
|
||||
// branch in the graph individually.
|
||||
for (auto &&info : m_collector_map) {
|
||||
m_threads.emplace_back(collectorThread,
|
||||
info.second.queues,
|
||||
info.second.mapping,
|
||||
m_sink_queues.size(),
|
||||
std::ref(m_out_queue));
|
||||
}
|
||||
state = State::READY;
|
||||
}
|
||||
|
||||
@@ -1018,15 +1268,25 @@ void cv::gimpl::GStreamingExecutor::wait_shutdown()
|
||||
for (auto &q : m_internal_queues) q->clear();
|
||||
m_out_queue.clear();
|
||||
|
||||
for (auto &&op : m_ops) {
|
||||
op.isl_exec->handleStopStream();
|
||||
}
|
||||
|
||||
state = State::STOPPED;
|
||||
}
|
||||
|
||||
bool cv::gimpl::GStreamingExecutor::pull(cv::GRunArgsP &&outs)
|
||||
{
|
||||
// This pull() can only be called when there's no desynchronized
|
||||
// parts in the graph.
|
||||
GAPI_Assert(!m_desync &&
|
||||
"This graph has desynchronized parts! Please use another pull()");
|
||||
|
||||
if (state == State::STOPPED)
|
||||
return false;
|
||||
GAPI_Assert(state == State::RUNNING);
|
||||
GAPI_Assert(m_sink_queues.size() == outs.size());
|
||||
GAPI_Assert(m_sink_queues.size() == outs.size() &&
|
||||
"Number of data objects in cv::gout() must match the number of graph outputs in cv::GOut()");
|
||||
|
||||
Cmd cmd;
|
||||
m_out_queue.pop(cmd);
|
||||
@@ -1036,12 +1296,39 @@ bool cv::gimpl::GStreamingExecutor::pull(cv::GRunArgsP &&outs)
|
||||
return false;
|
||||
}
|
||||
|
||||
GAPI_Assert(cv::util::holds_alternative<cv::GRunArgs>(cmd));
|
||||
cv::GRunArgs &this_result = cv::util::get<cv::GRunArgs>(cmd);
|
||||
GAPI_Assert(cv::util::holds_alternative<Result>(cmd));
|
||||
cv::GRunArgs &this_result = cv::util::get<Result>(cmd).args;
|
||||
sync_data(this_result, outs);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool cv::gimpl::GStreamingExecutor::pull(cv::GOptRunArgsP &&outs)
|
||||
{
|
||||
// This pull() can only be called in both cases: if there are
|
||||
// desyncrhonized parts or not.
|
||||
|
||||
// FIXME: so far it is a full duplicate of standard pull except
|
||||
// the sync_data version called.
|
||||
if (state == State::STOPPED)
|
||||
return false;
|
||||
GAPI_Assert(state == State::RUNNING);
|
||||
GAPI_Assert(m_sink_queues.size() == outs.size() &&
|
||||
"Number of data objects in cv::gout() must match the number of graph outputs in cv::GOut()");
|
||||
|
||||
Cmd cmd;
|
||||
m_out_queue.pop(cmd);
|
||||
if (cv::util::holds_alternative<Stop>(cmd))
|
||||
{
|
||||
wait_shutdown();
|
||||
return false;
|
||||
}
|
||||
|
||||
GAPI_Assert(cv::util::holds_alternative<Result>(cmd));
|
||||
sync_data(cv::util::get<Result>(cmd), outs);
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
bool cv::gimpl::GStreamingExecutor::try_pull(cv::GRunArgsP &&outs)
|
||||
{
|
||||
if (state == State::STOPPED)
|
||||
@@ -1059,8 +1346,8 @@ bool cv::gimpl::GStreamingExecutor::try_pull(cv::GRunArgsP &&outs)
|
||||
return false;
|
||||
}
|
||||
|
||||
GAPI_Assert(cv::util::holds_alternative<cv::GRunArgs>(cmd));
|
||||
cv::GRunArgs &this_result = cv::util::get<cv::GRunArgs>(cmd);
|
||||
GAPI_Assert(cv::util::holds_alternative<Result>(cmd));
|
||||
cv::GRunArgs &this_result = cv::util::get<Result>(cmd).args;
|
||||
sync_data(this_result, outs);
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -14,6 +14,8 @@
|
||||
|
||||
#include <memory> // unique_ptr, shared_ptr
|
||||
#include <thread> // thread
|
||||
#include <vector>
|
||||
#include <unordered_map>
|
||||
|
||||
#if defined(HAVE_TBB)
|
||||
# include <tbb/concurrent_queue.h> // FIXME: drop it from here!
|
||||
@@ -22,6 +24,7 @@ template<typename T> using QueueClass = tbb::concurrent_bounded_queue<T>;
|
||||
# include "executor/conc_queue.hpp"
|
||||
template<typename T> using QueueClass = cv::gapi::own::concurrent_bounded_queue<T>;
|
||||
#endif // TBB
|
||||
#include "executor/last_value.hpp"
|
||||
|
||||
#include <ade/graph.hpp>
|
||||
|
||||
@@ -40,14 +43,61 @@ struct Stop {
|
||||
cv::GRunArg cdata; // const data for CNST stop
|
||||
};
|
||||
|
||||
struct Result {
|
||||
cv::GRunArgs args; // Full results vector
|
||||
std::vector<bool> flags; // Availability flags (in case of desync)
|
||||
};
|
||||
|
||||
using Cmd = cv::util::variant
|
||||
< cv::util::monostate
|
||||
, Start // Tells emitters to start working. Not broadcasted to workers.
|
||||
, Stop // Tells emitters to stop working. Broadcasted to workers.
|
||||
, cv::GRunArg // Workers data payload to process.
|
||||
, cv::GRunArgs // Full results vector
|
||||
, Result // Pipeline's data for gout()
|
||||
>;
|
||||
using Q = QueueClass<Cmd>;
|
||||
|
||||
// Interface over a queue. The underlying queue implementation may be
|
||||
// different. This class is mainly introduced to bring some
|
||||
// abstraction over the real queues (bounded in-order) and a
|
||||
// desynchronized data slots (see required to implement
|
||||
// cv::gapi::desync)
|
||||
|
||||
class Q {
|
||||
public:
|
||||
virtual void push(const Cmd &cmd) = 0;
|
||||
virtual void pop(Cmd &cmd) = 0;
|
||||
virtual bool try_pop(Cmd &cmd) = 0;
|
||||
virtual void clear() = 0;
|
||||
virtual ~Q() = default;
|
||||
};
|
||||
|
||||
// A regular queue implementation
|
||||
class SyncQueue final: public Q {
|
||||
QueueClass<Cmd> m_q; // FIXME: OWN or WRAP??
|
||||
|
||||
public:
|
||||
virtual void push(const Cmd &cmd) override { m_q.push(cmd); }
|
||||
virtual void pop(Cmd &cmd) override { m_q.pop(cmd); }
|
||||
virtual bool try_pop(Cmd &cmd) override { return m_q.try_pop(cmd); }
|
||||
virtual void clear() override { m_q.clear(); }
|
||||
|
||||
void set_capacity(std::size_t c) { m_q.set_capacity(c);}
|
||||
};
|
||||
|
||||
// Desynchronized "queue" implementation
|
||||
// Every push overwrites value which is not yet popped
|
||||
// This container can hold 0 or 1 element
|
||||
// Special handling for Stop is implemented (FIXME: not really)
|
||||
class DesyncQueue final: public Q {
|
||||
cv::gapi::own::last_written_value<Cmd> m_v;
|
||||
|
||||
public:
|
||||
virtual void push(const Cmd &cmd) override { m_v.push(cmd); }
|
||||
virtual void pop(Cmd &cmd) override { m_v.pop(cmd); }
|
||||
virtual bool try_pop(Cmd &cmd) override { return m_v.try_pop(cmd); }
|
||||
virtual void clear() override { m_v.clear(); }
|
||||
};
|
||||
|
||||
} // namespace stream
|
||||
|
||||
// FIXME: Currently all GExecutor comments apply also
|
||||
@@ -87,6 +137,7 @@ protected:
|
||||
util::optional<bool> m_reshapable;
|
||||
|
||||
cv::gimpl::GIslandModel::Graph m_gim; // FIXME: make const?
|
||||
const bool m_desync;
|
||||
|
||||
// FIXME: Naive executor details are here for now
|
||||
// but then it should be moved to another place
|
||||
@@ -117,11 +168,27 @@ protected:
|
||||
std::vector<ade::NodeHandle> m_sinks;
|
||||
|
||||
std::vector<std::thread> m_threads;
|
||||
std::vector<stream::Q> m_emitter_queues;
|
||||
std::vector<stream::Q*> m_const_emitter_queues; // a view over m_emitter_queues
|
||||
std::vector<stream::Q*> m_sink_queues;
|
||||
std::unordered_set<stream::Q*> m_internal_queues;
|
||||
stream::Q m_out_queue;
|
||||
std::vector<stream::SyncQueue> m_emitter_queues;
|
||||
|
||||
// a view over m_emitter_queues
|
||||
std::vector<stream::SyncQueue*> m_const_emitter_queues;
|
||||
|
||||
std::vector<stream::Q*> m_sink_queues;
|
||||
|
||||
// desync path tags for outputs. -1 means that output
|
||||
// doesn't belong to a desync path
|
||||
std::vector<int> m_sink_sync;
|
||||
|
||||
std::unordered_set<stream::Q*> m_internal_queues;
|
||||
stream::SyncQueue m_out_queue;
|
||||
|
||||
// Describes mapping from desync paths to collector threads
|
||||
struct CollectorThreadInfo {
|
||||
std::vector<stream::Q*> queues;
|
||||
std::vector<int> mapping;
|
||||
};
|
||||
std::unordered_map<int, CollectorThreadInfo> m_collector_map;
|
||||
|
||||
|
||||
void wait_shutdown();
|
||||
|
||||
@@ -132,6 +199,7 @@ public:
|
||||
void setSource(GRunArgs &&args);
|
||||
void start();
|
||||
bool pull(cv::GRunArgsP &&outs);
|
||||
bool pull(cv::GOptRunArgsP &&outs);
|
||||
bool try_pull(cv::GRunArgsP &&outs);
|
||||
void stop();
|
||||
bool running() const;
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
// 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) 2020 Intel Corporation
|
||||
|
||||
#ifndef OPENCV_GAPI_EXECUTOR_LAST_VALUE_HPP
|
||||
#define OPENCV_GAPI_EXECUTOR_LAST_VALUE_HPP
|
||||
|
||||
#include <mutex>
|
||||
#include <condition_variable>
|
||||
|
||||
#include <opencv2/gapi/util/optional.hpp>
|
||||
#include <opencv2/gapi/own/assert.hpp>
|
||||
|
||||
namespace cv {
|
||||
namespace gapi {
|
||||
namespace own {
|
||||
|
||||
// This class implements a "Last Written Value" thing. Writer threads
|
||||
// (in our case, it is just one) can write as many values there as it
|
||||
// can.
|
||||
//
|
||||
// The reader thread gets only a value it gets at the time (or blocks
|
||||
// if there was no value written since the last read).
|
||||
//
|
||||
// Again, the implementation is highly inefficient right now.
|
||||
template<class T>
|
||||
class last_written_value {
|
||||
cv::util::optional<T> m_data;
|
||||
|
||||
std::mutex m_mutex;
|
||||
std::condition_variable m_cond_empty;
|
||||
|
||||
void unsafe_pop(T &t);
|
||||
|
||||
public:
|
||||
last_written_value() {}
|
||||
last_written_value(const last_written_value<T> &cc)
|
||||
: m_data(cc.m_data) {
|
||||
// FIXME: what to do with all that locks, etc?
|
||||
}
|
||||
last_written_value(last_written_value<T> &&cc)
|
||||
: m_data(std::move(cc.m_data)) {
|
||||
// FIXME: what to do with all that locks, etc?
|
||||
}
|
||||
|
||||
// FIXME: && versions
|
||||
void push(const T &t);
|
||||
void pop(T &t);
|
||||
bool try_pop(T &t);
|
||||
|
||||
// Not thread-safe
|
||||
void clear();
|
||||
};
|
||||
|
||||
// Internal: do shared pop things assuming the lock is already there
|
||||
template<typename T>
|
||||
void last_written_value<T>::unsafe_pop(T &t) {
|
||||
GAPI_Assert(m_data.has_value());
|
||||
t = std::move(m_data.value());
|
||||
m_data.reset();
|
||||
}
|
||||
|
||||
// Push an element to the queue. Blocking if there's no space left
|
||||
template<typename T>
|
||||
void last_written_value<T>::push(const T& t) {
|
||||
std::unique_lock<std::mutex> lock(m_mutex);
|
||||
m_data = cv::util::make_optional(t);
|
||||
lock.unlock();
|
||||
m_cond_empty.notify_one();
|
||||
}
|
||||
|
||||
// Pop an element from the queue. Blocking if there's no items
|
||||
template<typename T>
|
||||
void last_written_value<T>::pop(T &t) {
|
||||
std::unique_lock<std::mutex> lock(m_mutex);
|
||||
if (!m_data.has_value()) {
|
||||
// if there is no data, wait
|
||||
m_cond_empty.wait(lock, [&](){return m_data.has_value();});
|
||||
}
|
||||
unsafe_pop(t);
|
||||
}
|
||||
|
||||
// Try pop an element from the queue. Returns false if queue is empty
|
||||
template<typename T>
|
||||
bool last_written_value<T>::try_pop(T &t) {
|
||||
std::unique_lock<std::mutex> lock(m_mutex);
|
||||
if (!m_data.has_value()) {
|
||||
// if there is no data, return
|
||||
return false;
|
||||
}
|
||||
unsafe_pop(t);
|
||||
return true;
|
||||
}
|
||||
|
||||
// Clear the value holder. This method is not thread-safe.
|
||||
template<typename T>
|
||||
void last_written_value<T>::clear() {
|
||||
m_data.reset();
|
||||
}
|
||||
|
||||
}}} // namespace cv::gapi::own
|
||||
|
||||
#endif // OPENCV_GAPI_EXECUTOR_CONC_QUEUE_HPP
|
||||
Reference in New Issue
Block a user