Merge pull request #16031 from aDanPin:dm/streaming_auto_meta

G-API-NG/Streaming: don't require explicit metadata in compileStreaming()

* First probably working version
Hardcode gose to setSource() :)

* Pre final version of move metadata declaration from compileStreaming() to setSource().

* G-API-NG/Streaming: recovered the existing Streaming functionality

- The auto-meta test is disabling since it crashes.
- Restored .gitignore

* G-API-NG/Streaming: Made the meta-less compileStreaming() work

- Works fine even with OpenCV backend;
- Fluid doesn't support such kind of compilation so far - to be fixed

* G-API-NG/Streaming: Fix Fluid to support meta-less compilation

- Introduced a notion of metadata-sensitive passes and slightly
  refactored GCompiler and GFluidBackend to support that
- Fixed a TwoVideoSourcesFail test on streaming

* Add three smoke streaming tests to gapi_streaming_tests.
All three teste run pipeline with two different input sets
1) SmokeTest_Two_Const_Mats test run pipeline with two const Mats
2) SmokeTest_One_Video_One_Const_Scalar test run pipleline with Mat(video source) and const Scalar
3) SmokeTest_One_Video_One_Const_Vector test run pipeline with Mat(video source) and const Vector
 # Please enter the commit message for your changes. Lines starting

* style fix

* Some review stuff

* Some review stuff

modules/gapi/include/opencv2/gapi/gcomputation.hpp

@@ -410,11 +410,12 @@ public:
      *
      * @param in_metas vector of input metadata configuration. Grab
      * metadata from real data objects (like cv::Mat or cv::Scalar)
-     * using cv::descr_of(), or create it on your own.  @param args
-     * compilation arguments for this compilation process. Compilation
-     * arguments directly affect what kind of executable object would
-     * be produced, e.g. which kernels (and thus, devices) would be
-     * used to execute computation.
+     * using cv::descr_of(), or create it on your own.
+     *
+     * @param args compilation arguments for this compilation
+     * process. Compilation arguments directly affect what kind of
+     * executable object would be produced, e.g. which kernels (and
+     * thus, devices) would be used to execute computation.
      *
      * @return GStreamingCompiled, a streaming-oriented executable
      * computation compiled specifically for the given input
@@ -424,6 +425,27 @@ public:
      */
     GStreamingCompiled compileStreaming(GMetaArgs &&in_metas, GCompileArgs &&args = {});
 
+    /**
+     * @brief Compile the computation for streaming mode.
+     *
+     * This method triggers compilation process and produces a new
+     * GStreamingCompiled object which then can process video stream
+     * data in any format. Underlying mechanisms will be adjusted to
+     * every new input video stream automatically, but please note that
+     * _not all_ existing backends support this (see reshape()).
+     *
+     * @param args compilation arguments for this compilation
+     * process. Compilation arguments directly affect what kind of
+     * executable object would be produced, e.g. which kernels (and
+     * thus, devices) would be used to execute computation.
+     *
+     * @return GStreamingCompiled, a streaming-oriented executable
+     * computation compiled for any input image format.
+     *
+     * @sa @ref gapi_compile_args
+     */
+    GStreamingCompiled compileStreaming(GCompileArgs &&args = {});
+
     // 2. Direct metadata version
     /**
      * @overload

modules/gapi/include/opencv2/gapi/streaming/cap.hpp

@@ -45,31 +45,62 @@ namespace wip {
 class GCaptureSource: public IStreamSource
 {
 public:
-    explicit GCaptureSource(int id) : cap(id) {}
-    explicit GCaptureSource(const std::string &path) : cap(path) {}
+    explicit GCaptureSource(int id) : cap(id) { prep(); }
+    explicit GCaptureSource(const std::string &path) : cap(path) { prep(); }
 
     // TODO: Add more constructor overloads to make it
     // fully compatible with VideoCapture's interface.
 
 protected:
     cv::VideoCapture cap;
+    cv::Mat first;
+    bool first_pulled = false;
+
+    void prep()
+    {
+        // Prepare first frame to report its meta to engine
+        // when needed
+        GAPI_Assert(first.empty());
+        cv::Mat tmp;
+        if (!cap.read(tmp))
+        {
+            GAPI_Assert(false && "Couldn't grab the very first frame");
+        }
+        // NOTE: Some decode/media VideoCapture backends continue
+        // owning the video buffer under cv::Mat so in order to
+        // process it safely in a highly concurrent pipeline, clone()
+        // is the only right way.
+        first = tmp.clone();
+    }
+
     virtual bool pull(cv::gapi::wip::Data &data) override
     {
+        if (!first_pulled)
+        {
+            GAPI_Assert(!first.empty());
+            first_pulled = true;
+            data = first; // no need to clone here since it was cloned already
+            return true;
+        }
+
         if (!cap.isOpened()) return false;
+
         cv::Mat frame;
         if (!cap.read(frame))
         {
             // end-of-stream happened
             return false;
         }
-
-        // NOTE: Some decode/media VideoCapture backends continue
-        // owning the video buffer under cv::Mat so in order to
-        // process it safely in a highly concurrent pipeline, clone()
-        // is the only right way.
+        // Same reason to clone as in prep()
         data = frame.clone();
         return true;
     }
+
+    virtual GMetaArg descr_of() const override
+    {
+        GAPI_Assert(!first.empty());
+        return cv::GMetaArg{cv::descr_of(first)};
+    }
 };
 
 } // namespace wip

modules/gapi/include/opencv2/gapi/streaming/source.hpp

@@ -7,8 +7,11 @@
 #ifndef OPENCV_GAPI_STREAMING_SOURCE_HPP
 #define OPENCV_GAPI_STREAMING_SOURCE_HPP
 
-#include <memory> // shared_ptr
-#include <type_traits> // is_base_of
+#include <memory>                      // shared_ptr
+#include <type_traits>                 // is_base_of
+
+#include <opencv2/gapi/gmetaarg.hpp>   // GMetaArg
+
 
 namespace cv {
 namespace gapi {
@@ -38,6 +41,7 @@ public:
     using Ptr = std::shared_ptr<IStreamSource>;
     Ptr ptr() { return shared_from_this(); }
     virtual bool pull(Data &data) = 0;
+    virtual GMetaArg descr_of() const = 0;
     virtual ~IStreamSource() = default;
 };
 

modules/gapi/src/api/gbackend.cpp

@@ -56,6 +56,13 @@ void cv::gapi::GBackend::Priv::addBackendPasses(ade::ExecutionEngineSetupContext
     // add custom (backend-specific) graph transformations
 }
 
+void cv::gapi::GBackend::Priv::addMetaSensitiveBackendPasses(ade::ExecutionEngineSetupContext &)
+{
+    // Do nothing by default, plugins may override this to
+    // add custom (backend-specific) graph transformations
+    // which are sensitive to metadata
+}
+
 cv::gapi::GKernelPackage cv::gapi::GBackend::Priv::auxiliaryKernels() const
 {
     return {};

modules/gapi/src/api/gbackend_priv.hpp

@@ -50,14 +50,22 @@ public:
                          const GCompileArgs &args,
                          const std::vector<ade::NodeHandle> &nodes) const;
 
+
     virtual EPtr compile(const ade::Graph   &graph,
                          const GCompileArgs &args,
                          const std::vector<ade::NodeHandle> &nodes,
                          const std::vector<cv::gimpl::Data>& ins_data,
                          const std::vector<cv::gimpl::Data>& outs_data) const;
 
+    // Ask backend to provide general backend-specific compiler passes
     virtual void addBackendPasses(ade::ExecutionEngineSetupContext &);
 
+    // Ask backend to put extra meta-sensitive backend passes Since
+    // the inception of Streaming API one can compile graph without
+    // meta information, so if some passes depend on this information,
+    // they are called when meta information becomes available.
+    virtual void addMetaSensitiveBackendPasses(ade::ExecutionEngineSetupContext &);
+
     virtual cv::gapi::GKernelPackage auxiliaryKernels() const;
 
     virtual ~Priv() = default;

modules/gapi/src/api/gcomputation.cpp

@@ -82,6 +82,12 @@ cv::GStreamingCompiled cv::GComputation::compileStreaming(GMetaArgs &&metas, GCo
     return comp.compileStreaming();
 }
 
+cv::GStreamingCompiled cv::GComputation::compileStreaming(GCompileArgs &&args)
+{
+    cv::gimpl::GCompiler comp(*this, {}, std::move(args));
+    return comp.compileStreaming();
+}
+
 // FIXME: Introduce similar query/test method for GMetaArgs as a building block
 // for functions like this?
 static bool formats_are_same(const cv::GMetaArgs& metas1, const cv::GMetaArgs& metas2)

modules/gapi/src/api/gproto.cpp

@@ -110,6 +110,9 @@ cv::GMetaArg cv::descr_of(const cv::GRunArg &arg)
         case GRunArg::index_of<cv::detail::VectorRef>():
             return cv::GMetaArg(util::get<cv::detail::VectorRef>(arg).descr_of());
 
+        case GRunArg::index_of<cv::gapi::wip::IStreamSource::Ptr>():
+            return cv::util::get<cv::gapi::wip::IStreamSource::Ptr>(arg)->descr_of();
+
         default: util::throw_error(std::logic_error("Unsupported GRunArg type"));
     }
 }

modules/gapi/src/backends/fluid/gfluidbackend.cpp

@@ -128,7 +128,7 @@ namespace
             ;
         }
 
-        virtual void addBackendPasses(ade::ExecutionEngineSetupContext &ectx) override;
+        virtual void addMetaSensitiveBackendPasses(ade::ExecutionEngineSetupContext &ectx) override;
 
     };
 }
@@ -1418,7 +1418,7 @@ void cv::gimpl::GParallelFluidExecutable::run(std::vector<InObj>  &&input_objs,
 
 // FIXME: these passes operate on graph global level!!!
 // Need to fix this for heterogeneous (island-based) processing
-void GFluidBackendImpl::addBackendPasses(ade::ExecutionEngineSetupContext &ectx)
+void GFluidBackendImpl::addMetaSensitiveBackendPasses(ade::ExecutionEngineSetupContext &ectx)
 {
     using namespace cv::gimpl;
 

modules/gapi/src/compiler/gcompiler.cpp

@@ -259,13 +259,16 @@ cv::gimpl::GCompiler::GCompiler(const cv::GComputation &c,
                                                       // (no compound backend present here)
     m_e.addPass("kernels", "check_islands_content", passes::checkIslandsContent);
 
+    //Input metas may be empty when a graph is compiled for streaming
     m_e.addPassStage("meta");
-    m_e.addPass("meta", "initialize",   std::bind(passes::initMeta, _1, std::ref(m_metas)));
-    m_e.addPass("meta", "propagate",    std::bind(passes::inferMeta, _1, false));
-    m_e.addPass("meta", "finalize",     passes::storeResultingMeta);
-    // moved to another stage, FIXME: two dumps?
-    //    m_e.addPass("meta", "dump_dot",     passes::dumpDotStdout);
-
+    if (!m_metas.empty())
+    {
+        m_e.addPass("meta", "initialize",   std::bind(passes::initMeta, _1, std::ref(m_metas)));
+        m_e.addPass("meta", "propagate",    std::bind(passes::inferMeta, _1, false));
+        m_e.addPass("meta", "finalize",     passes::storeResultingMeta);
+        // moved to another stage, FIXME: two dumps?
+        //    m_e.addPass("meta", "dump_dot",     passes::dumpDotStdout);
+    }
     // Special stage for backend-specific transformations
     // FIXME: document passes hierarchy and order for backend developers
     m_e.addPassStage("transform");
@@ -280,6 +283,10 @@ cv::gimpl::GCompiler::GCompiler(const cv::GComputation &c,
     // FIXME: add a better way to do that!
     m_e.addPass("exec", "add_streaming",    passes::addStreaming);
 
+    // Note: Must be called after addStreaming as addStreaming pass
+    // can possibly add new nodes to the IslandModel
+    m_e.addPass("exec", "sort_islands",     passes::topoSortIslands);
+
     if (dump_path.has_value())
     {
         m_e.addPass("exec", "dump_dot", std::bind(passes::dumpGraph, _1,
@@ -294,6 +301,10 @@ cv::gimpl::GCompiler::GCompiler(const cv::GComputation &c,
     for (auto &b : backends)
     {
         b.priv().addBackendPasses(ectx);
+        if (!m_metas.empty())
+        {
+            b.priv().addMetaSensitiveBackendPasses(ectx);
+        }
     }
 }
 
@@ -361,9 +372,18 @@ void cv::gimpl::GCompiler::validateOutProtoArgs()
 
 cv::gimpl::GCompiler::GPtr cv::gimpl::GCompiler::generateGraph()
 {
-    validateInputMeta();
+    if (!m_metas.empty())
+    {
+        // Metadata may be empty if we're compiling our graph for streaming
+        validateInputMeta();
+    }
     validateOutProtoArgs();
-    return makeGraph(m_c.priv().m_ins, m_c.priv().m_outs);
+    auto g = makeGraph(m_c.priv().m_ins, m_c.priv().m_outs);
+    if (!m_metas.empty())
+    {
+        GModel::Graph(*g).metadata().set(OriginalInputMeta{m_metas});
+    }
+    return g;
 }
 
 void cv::gimpl::GCompiler::runPasses(ade::Graph &g)
@@ -373,17 +393,17 @@ void cv::gimpl::GCompiler::runPasses(ade::Graph &g)
 }
 
 void cv::gimpl::GCompiler::compileIslands(ade::Graph &g)
+{
+    compileIslands(g, m_args);
+}
+
+void cv::gimpl::GCompiler::compileIslands(ade::Graph &g, const cv::GCompileArgs &args)
 {
     GModel::Graph gm(g);
     std::shared_ptr<ade::Graph> gptr(gm.metadata().get<IslandModel>().model);
     GIslandModel::Graph gim(*gptr);
 
-    // Run topological sort on GIslandModel first
-    auto pass_ctx = ade::passes::PassContext{*gptr};
-    ade::passes::TopologicalSort{}(pass_ctx);
-
-    // Now compile islands
-    GIslandModel::compileIslands(gim, g, m_args);
+    GIslandModel::compileIslands(gim, g, args);
 }
 
 cv::GCompiled cv::gimpl::GCompiler::produceCompiled(GPtr &&pg)
@@ -417,12 +437,27 @@ cv::GCompiled cv::gimpl::GCompiler::produceCompiled(GPtr &&pg)
 
 cv::GStreamingCompiled cv::gimpl::GCompiler::produceStreamingCompiled(GPtr &&pg)
 {
-    const auto &outMetas = GModel::ConstGraph(*pg).metadata()
-        .get<OutputMeta>().outMeta;
-    std::unique_ptr<GStreamingExecutor> pE(new GStreamingExecutor(std::move(pg)));
-
     GStreamingCompiled compiled;
-    compiled.priv().setup(m_metas, outMetas, std::move(pE));
+    GMetaArgs outMetas;
+
+    // FIXME: the whole below construct is ugly, need to revise
+    // how G*Compiled learns about its meta.
+    if (!m_metas.empty())
+    {
+        outMetas = GModel::ConstGraph(*pg).metadata().get<OutputMeta>().outMeta;
+    }
+
+    std::unique_ptr<GStreamingExecutor> pE(new GStreamingExecutor(std::move(pg)));
+    if (!m_metas.empty() && !outMetas.empty())
+    {
+        compiled.priv().setup(m_metas, outMetas, std::move(pE));
+    }
+    else if (m_metas.empty() && outMetas.empty())
+    {
+        // Otherwise, set it up with executor object only
+        compiled.priv().setup(std::move(pE));
+    }
+    else GAPI_Assert(false && "Impossible happened -- please report a bug");
     return compiled;
 }
 
@@ -440,6 +475,34 @@ cv::GStreamingCompiled cv::gimpl::GCompiler::compileStreaming()
     std::unique_ptr<ade::Graph> pG = generateGraph();
     GModel::Graph(*pG).metadata().set(Streaming{});
     runPasses(*pG);
-    compileIslands(*pG);
+    if (!m_metas.empty())
+    {
+        // If the metadata has been passed, compile our islands!
+        compileIslands(*pG);
+    }
     return produceStreamingCompiled(std::move(pG));
 }
+
+void cv::gimpl::GCompiler::runMetaPasses(ade::Graph &g, const cv::GMetaArgs &metas)
+{
+    auto pass_ctx = ade::passes::PassContext{g};
+    cv::gimpl::passes::initMeta(pass_ctx, metas);
+    cv::gimpl::passes::inferMeta(pass_ctx, true);
+    cv::gimpl::passes::storeResultingMeta(pass_ctx);
+
+    // Also run meta-sensitive backend-specific passes, if there's any.
+    // FIXME: This may be hazardous if our backend are not very robust
+    // in their passes -- how can we guarantee correct functioning in the
+    // future?
+    ade::ExecutionEngine engine;
+    engine.addPassStage("exec"); // FIXME: Need a better decision on how we replicate
+                                 // our main compiler stages here.
+    ade::ExecutionEngineSetupContext ectx(engine);
+
+    // NB: &&b or &b doesn't work here since "backends" is a set. Nevermind
+    for (auto b : GModel::Graph(g).metadata().get<ActiveBackends>().backends)
+    {
+        b.priv().addMetaSensitiveBackendPasses(ectx);
+    }
+    engine.runPasses(g);
+}

modules/gapi/src/compiler/gcompiler.hpp

@@ -29,12 +29,16 @@ class GAPI_EXPORTS GCompiler
     cv::gapi::GKernelPackage m_all_kernels;
     cv::gapi::GNetPackage    m_all_networks;
 
-    std::vector<std::unique_ptr<ade::Graph>> m_all_patterns;  // built patterns from transformations
+    // Patters built from transformations
+    std::vector<std::unique_ptr<ade::Graph>> m_all_patterns;
+
 
     void validateInputMeta();
     void validateOutProtoArgs();
 
 public:
+    // Metas may be empty in case when graph compiling for streaming
+    // In this case graph get metas from first frame
     explicit GCompiler(const GComputation &c,
                              GMetaArgs    &&metas,
                              GCompileArgs &&args);
@@ -47,11 +51,13 @@ public:
 
     // But those are actually composed of this:
     using GPtr = std::unique_ptr<ade::Graph>;
-    GPtr       generateGraph();               // Unroll GComputation into a GModel
-    void       runPasses(ade::Graph &g);      // Apply all G-API passes on a GModel
-    void       compileIslands(ade::Graph &g); // Instantiate GIslandExecutables in GIslandModel
-    GCompiled  produceCompiled(GPtr &&pg);    // Produce GCompiled from processed GModel
+    GPtr        generateGraph();               // Unroll GComputation into a GModel
+    void        runPasses(ade::Graph &g);      // Apply all G-API passes on a GModel
+    void        compileIslands(ade::Graph &g); // Instantiate GIslandExecutables in GIslandModel
+    static void compileIslands(ade::Graph &g, const cv::GCompileArgs &args);
+    GCompiled   produceCompiled(GPtr &&pg);    // Produce GCompiled from processed GModel
     GStreamingCompiled  produceStreamingCompiled(GPtr &&pg); // Produce GStreamingCompiled from processed GMbodel
+    static void runMetaPasses(ade::Graph &g, const cv::GMetaArgs &metas);
 };
 
 }}

modules/gapi/src/compiler/gislandmodel.cpp

@@ -290,6 +290,7 @@ void GIslandModel::compileIslands(Graph &g, const ade::Graph &orig_g, const GCom
             g.metadata(nh).set(IslandExec{std::move(island_exe)});
         }
     }
+    g.metadata().set(IslandsCompiled{});
 }
 
 ade::NodeHandle GIslandModel::producerOf(const ConstGraph &g, ade::NodeHandle &data_nh)

modules/gapi/src/compiler/gislandmodel.hpp

@@ -166,6 +166,12 @@ struct Sink
     std::size_t proto_index;
 };
 
+// This flag is set in graph's own metadata if compileIsland was successful
+struct IslandsCompiled
+{
+    static const char *name() { return "IslandsCompiled"; }
+};
+
 namespace GIslandModel
 {
     using Graph = ade::TypedGraph
@@ -175,6 +181,7 @@ namespace GIslandModel
         , IslandExec
         , Emitter
         , Sink
+        , IslandsCompiled
         , ade::passes::TopologicalSortData
         >;
 
@@ -186,6 +193,7 @@ namespace GIslandModel
         , IslandExec
         , Emitter
         , Sink
+        , IslandsCompiled
         , ade::passes::TopologicalSortData
         >;
 

modules/gapi/src/compiler/gmodel.hpp

@@ -109,6 +109,17 @@ struct Protocol
     std::vector<ade::NodeHandle> out_nhs;
 };
 
+// The original metadata the graph has been compiled for.
+// - For regular GCompiled, this information always present and
+//   is NOT updated on reshape()
+// - For GStreamingCompiled, this information may be missing.
+//   It means that compileStreaming() was called without meta.
+struct OriginalInputMeta
+{
+    static const char *name() { return "OriginalInputMeta"; }
+    GMetaArgs inputMeta;
+};
+
 struct OutputMeta
 {
     static const char *name() { return "OutputMeta"; }
@@ -193,6 +204,7 @@ namespace GModel
         , ConstValue
         , Island
         , Protocol
+        , OriginalInputMeta
         , OutputMeta
         , Journal
         , ade::passes::TopologicalSortData
@@ -213,6 +225,7 @@ namespace GModel
         , ConstValue
         , Island
         , Protocol
+        , OriginalInputMeta
         , OutputMeta
         , Journal
         , ade::passes::TopologicalSortData

modules/gapi/src/compiler/gstreaming.cpp

@@ -27,6 +27,11 @@ void cv::GStreamingCompiled::Priv::setup(const GMetaArgs &_metaArgs,
     m_exec     = std::move(_pE);
 }
 
+void cv::GStreamingCompiled::Priv::setup(std::unique_ptr<cv::gimpl::GStreamingExecutor> &&_pE)
+{
+    m_exec = std::move(_pE);
+}
+
 bool cv::GStreamingCompiled::Priv::isEmpty() const
 {
     return !m_exec;
@@ -47,9 +52,7 @@ const cv::GMetaArgs& cv::GStreamingCompiled::Priv::outMetas() const
 // the G*Compiled's priv?
 void cv::GStreamingCompiled::Priv::setSource(cv::GRunArgs &&args)
 {
-    // FIXME: This metadata checking should be removed at all
-    // for the streaming case.
-    if (!can_describe(m_metas, args))
+    if (!m_metas.empty() && !can_describe(m_metas, args))
     {
         util::throw_error(std::logic_error("This object was compiled "
                                            "for different metadata!"));

modules/gapi/src/compiler/gstreaming_priv.hpp

@@ -32,6 +32,7 @@ public:
     void setup(const GMetaArgs &metaArgs,
                const GMetaArgs &outMetas,
                std::unique_ptr<cv::gimpl::GStreamingExecutor> &&pE);
+    void setup(std::unique_ptr<cv::gimpl::GStreamingExecutor> &&pE);
     bool isEmpty() const;
 
     const GMetaArgs& metas() const;

modules/gapi/src/compiler/passes/exec.cpp

@@ -638,4 +638,12 @@ void passes::syncIslandTags(ade::passes::PassContext &ctx)
     GIslandModel::Graph gim(*gptr);
     GIslandModel::syncIslandTags(gim, ctx.graph);
 }
+
+void passes::topoSortIslands(ade::passes::PassContext &ctx)
+{
+    GModel::Graph gm(ctx.graph);
+    std::shared_ptr<ade::Graph> gptr(gm.metadata().get<IslandModel>().model);
+    auto pass_ctx = ade::passes::PassContext{*gptr};
+    ade::passes::TopologicalSort{}(pass_ctx);
+}
 }} // namespace cv::gimpl

modules/gapi/src/compiler/passes/passes.hpp

@@ -58,6 +58,7 @@ void resolveKernels(ade::passes::PassContext   &ctx,
 
 void fuseIslands(ade::passes::PassContext &ctx);
 void syncIslandTags(ade::passes::PassContext &ctx);
+void topoSortIslands(ade::passes::PassContext &ctx);
 
 void applyTransformations(ade::passes::PassContext &ctx,
                           const gapi::GKernelPackage &pkg,

modules/gapi/src/executor/gstreamingexecutor.cpp

@@ -15,6 +15,7 @@
 #include "executor/gstreamingexecutor.hpp"
 #include "compiler/passes/passes.hpp"
 #include "backends/common/gbackend.hpp" // createMat
+#include "compiler/gcompiler.hpp" // for compileIslands
 
 namespace
 {
@@ -420,6 +421,10 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
             return m_gim.metadata(nh).get<NodeKind>().k == NodeKind::ISLAND;
          });
 
+    // If metadata was not passed to compileStreaming, Islands are not compiled at this point.
+    // It is fine -- Islands are then compiled in setSource (at the first valid call).
+    const bool islands_compiled = m_gim.metadata().contains<IslandsCompiled>();
+
     auto sorted = m_gim.metadata().get<ade::passes::TopologicalSortData>();
     for (auto nh : sorted.nodes())
     {
@@ -440,7 +445,8 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
 
                 // FIXME: THIS ORDER IS IRRELEVANT TO PROTOCOL OR ANY OTHER ORDER!
                 // FIXME: SAME APPLIES TO THE REGULAR GEEXECUTOR!!
-                auto xtract_in = [&](ade::NodeHandle slot_nh, std::vector<RcDesc> &vec) {
+                auto xtract_in = [&](ade::NodeHandle slot_nh, std::vector<RcDesc> &vec)
+                {
                     const auto orig_data_nh
                         = m_gim.metadata(slot_nh).get<DataSlot>().original_data_node;
                     const auto &orig_data_info
@@ -458,7 +464,8 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
                                            , orig_data_info.shape
                                            , orig_data_info.ctor});
                 };
-                auto xtract_out = [&](ade::NodeHandle slot_nh, std::vector<RcDesc> &vec, cv::GMetaArgs &metas) {
+                auto xtract_out = [&](ade::NodeHandle slot_nh, std::vector<RcDesc> &vec, cv::GMetaArgs &metas)
+                {
                     const auto orig_data_nh
                         = m_gim.metadata(slot_nh).get<DataSlot>().original_data_node;
                     const auto &orig_data_info
@@ -476,13 +483,16 @@ cv::gimpl::GStreamingExecutor::GStreamingExecutor(std::unique_ptr<ade::Graph> &&
                 for (auto in_slot_nh  : nh->inNodes())  xtract_in(in_slot_nh,  input_rcs);
                 for (auto out_slot_nh : nh->outNodes()) xtract_out(out_slot_nh, output_rcs, output_metas);
 
+                std::shared_ptr<GIslandExecutable> isl_exec = islands_compiled
+                    ? m_gim.metadata(nh).get<IslandExec>().object
+                    : nullptr;
                 m_ops.emplace_back(OpDesc{ std::move(input_rcs)
                                          , std::move(output_rcs)
                                          , std::move(output_metas)
                                          , nh
                                          , in_constants
-                                         , m_gim.metadata(nh).get<IslandExec>().object});
-
+                                         , isl_exec
+                                         });
                 // Initialize queues for every operation's input
                 ade::TypedGraph<DataQueue> qgr(*m_island_graph);
                 for (auto eh : nh->inEdges())
@@ -542,7 +552,8 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
 {
     GAPI_Assert(state == State::READY || state == State::STOPPED);
 
-    const auto is_video = [](const GRunArg &arg) {
+    const auto is_video = [](const GRunArg &arg)
+    {
         return util::holds_alternative<cv::gapi::wip::IStreamSource::Ptr>(arg);
     };
     const auto num_videos = std::count_if(ins.begin(), ins.end(), is_video);
@@ -554,6 +565,67 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
                                            " currently supported!"));
     }
 
+    GModel::ConstGraph gm(*m_orig_graph);
+    // Now the tricky-part: completing Islands compilation if compileStreaming
+    // has been called without meta arguments.
+    // The logic is basically the following:
+    // - (0) Collect metadata from input vector;
+    // - (1) If graph is compiled with meta
+    //   - (2) Just check if the passed objects have correct meta.
+    // - (3) Otherwise:
+    //   - (4) Run metadata inference;
+    //   - (5) If islands are not compiled at this point OR are not reshapeable:
+    //     - (6) Compile them for a first time with this meta;
+    //     - (7) Update internal structures with this island information
+    //   - (8) Otherwise:
+    //     - (9) Reshape islands to this new metadata.
+    //     - (10) Update internal structures again
+    const auto update_int_metas = [&]()
+    {
+        for (auto& op : m_ops)
+        {
+            op.out_metas.resize(0);
+            for (auto out_slot_nh : op.nh->outNodes())
+            {
+                const auto &orig_nh = m_gim.metadata(out_slot_nh).get<DataSlot>().original_data_node;
+                const auto &orig_info = gm.metadata(orig_nh).get<Data>();
+                op.out_metas.emplace_back(orig_info.meta);
+            }
+        }
+    };
+    const auto new_meta = cv::descr_of(ins); // 0
+    if (gm.metadata().contains<OriginalInputMeta>()) // (1)
+    {
+        // NB: Metadata is tested in setSource() already - just put an assert here
+        GAPI_Assert(new_meta == gm.metadata().get<OriginalInputMeta>().inputMeta); // (2)
+    }
+    else // (3)
+    {
+        GCompiler::runMetaPasses(*m_orig_graph.get(), new_meta); // (4)
+        if (!m_gim.metadata().contains<IslandsCompiled>()
+            || (m_reshapable.has_value() && m_reshapable.value() == false)) // (5)
+        {
+            bool is_reshapable = true;
+            GCompiler::compileIslands(*m_orig_graph.get(), m_comp_args); // (6)
+            for (auto& op : m_ops)
+            {
+                op.isl_exec = m_gim.metadata(op.nh).get<IslandExec>().object;
+                is_reshapable &= op.isl_exec->canReshape();
+            }
+            update_int_metas(); // (7)
+            m_reshapable = util::make_optional(is_reshapable);
+        }
+        else // (8)
+        {
+            for (auto& op : m_ops)
+            {
+                op.isl_exec->reshape(*m_orig_graph, m_comp_args); // (9)
+            }
+            update_int_metas(); // (10)
+        }
+    }
+    // Metadata handling is done!
+
     // Walk through the protocol, set-up emitters appropriately
     // There's a 1:1 mapping between emitters and corresponding data inputs.
     for (auto it : ade::util::zip(ade::util::toRange(m_emitters),
@@ -592,7 +664,8 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
     // all other inputs are "constant" generators.
     // Craft here a completion callback to notify Const emitters that
     // a video source is over
-    auto real_video_completion_cb = [this]() {
+    auto real_video_completion_cb = [this]()
+    {
         for (auto q : m_const_emitter_queues) q->push(Cmd{Stop{}});
     };
 
@@ -624,6 +697,7 @@ void cv::gimpl::GStreamingExecutor::setSource(GRunArgs &&ins)
                                real_video_completion_cb);
     }
 
+
     // Now do this for every island (in a topological order)
     for (auto &&op : m_ops)
     {

modules/gapi/src/executor/gstreamingexecutor.hpp

@@ -76,6 +76,9 @@ protected:
 
     std::unique_ptr<ade::Graph> m_orig_graph;
     std::shared_ptr<ade::Graph> m_island_graph;
+    cv::GCompileArgs m_comp_args;
+    cv::GMetaArgs m_last_metas;
+    util::optional<bool> m_reshapable;
 
     cv::gimpl::GIslandModel::Graph m_gim; // FIXME: make const?
 
@@ -90,7 +93,6 @@ protected:
 
         std::vector<GRunArg> in_constants;
 
-        // FIXME: remove it as unused
         std::shared_ptr<GIslandExecutable> isl_exec;
     };
     std::vector<OpDesc> m_ops;

modules/gapi/test/streaming/gapi_streaming_tests.cpp

@@ -302,6 +302,102 @@ TEST_P(GAPI_Streaming, SmokeTest_VideoConstSource_NoHang)
     EXPECT_EQ(ref_frames, test_frames);
 }
 
+TEST_P(GAPI_Streaming, SmokeTest_AutoMeta)
+{
+    cv::GMat in;
+    cv::GMat in2;
+    cv::GMat roi = cv::gapi::crop(in2, cv::Rect{1,1,256,256});
+    cv::GMat blr = cv::gapi::blur(roi, cv::Size(3,3));
+    cv::GMat out = blr - in;
+
+    auto testc = cv::GComputation(cv::GIn(in, in2), cv::GOut(out))
+        .compileStreaming(cv::compile_args(cv::gapi::use_only{GetParam()}));
+
+    cv::Mat in_const = cv::Mat::eye(cv::Size(256,256), CV_8UC3);
+    cv::Mat tmp;
+
+    // Test with one video source
+    auto in_src = gapi::wip::make_src<cv::gapi::wip::GCaptureSource>(findDataFile("cv/video/768x576.avi"));
+    testc.setSource(cv::gin(in_const, in_src));
+    testc.start();
+
+    std::size_t test_frames = 0u;
+    while (testc.pull(cv::gout(tmp))) test_frames++;
+    EXPECT_EQ(100u, test_frames);
+
+    // Now test with another one
+    in_src = gapi::wip::make_src<cv::gapi::wip::GCaptureSource>(findDataFile("cv/video/1920x1080.avi"));
+    testc.setSource(cv::gin(in_const, in_src));
+    testc.start();
+
+    test_frames = 0u;
+    while (testc.pull(cv::gout(tmp))) test_frames++;
+    EXPECT_EQ(165u, test_frames);
+}
+
+
+TEST_P(GAPI_Streaming, SmokeTest_AutoMeta_2xConstMat)
+{
+    cv::GMat in;
+    cv::GMat in2;
+    cv::GMat roi = cv::gapi::crop(in2, cv::Rect{1,1,256,256});
+    cv::GMat blr = cv::gapi::blur(roi, cv::Size(3,3));
+    cv::GMat out = blr - in;
+
+    auto testc = cv::GComputation(cv::GIn(in, in2), cv::GOut(out))
+        .compileStreaming(cv::compile_args(cv::gapi::use_only{GetParam()}));
+
+    cv::Mat in_const = cv::Mat::eye(cv::Size(256,256), CV_8UC3);
+    cv::Mat tmp;
+
+    // Test with first image
+    auto in_src = cv::imread(findDataFile("cv/edgefilter/statue.png"));
+    testc.setSource(cv::gin(in_const, in_src));
+    testc.start();
+
+    ASSERT_TRUE(testc.pull(cv::gout(tmp)));
+
+    testc.stop();
+
+    // Now test with second image
+    in_src = cv::imread(findDataFile("cv/edgefilter/kodim23.png"));
+    testc.setSource(cv::gin(in_const, in_src));
+    testc.start();
+
+    ASSERT_TRUE(testc.pull(cv::gout(tmp)));
+
+    testc.stop();
+}
+
+TEST_P(GAPI_Streaming, SmokeTest_AutoMeta_VideoScalar)
+{
+    cv::GMat in_m;
+    cv::GScalar in_s;
+    cv::GMat out_m = in_m * in_s;
+
+    auto testc = cv::GComputation(cv::GIn(in_m, in_s), cv::GOut(out_m))
+        .compileStreaming(cv::compile_args(cv::gapi::use_only{GetParam()}));
+
+    cv::Mat tmp;
+    // Test with one video source and scalar
+    auto in_src = gapi::wip::make_src<cv::gapi::wip::GCaptureSource>(findDataFile("cv/video/768x576.avi"));
+    testc.setSource(cv::gin(in_src, cv::Scalar{1.25}));
+    testc.start();
+
+    std::size_t test_frames = 0u;
+    while (testc.pull(cv::gout(tmp))) test_frames++;
+    EXPECT_EQ(100u, test_frames);
+
+    // Now test with another one video source and scalar
+    in_src = gapi::wip::make_src<cv::gapi::wip::GCaptureSource>(findDataFile("cv/video/1920x1080.avi"));
+    testc.setSource(cv::gin(in_src, cv::Scalar{0.75}));
+    testc.start();
+
+    test_frames = 0u;
+    while (testc.pull(cv::gout(tmp))) test_frames++;
+    EXPECT_EQ(165u, test_frames);
+}
+
 INSTANTIATE_TEST_CASE_P(TestStreaming, GAPI_Streaming,
                         Values(  OCV_KERNELS()
                                  //, OCL_KERNELS() // FIXME: Fails bit-exactness check, maybe relax it?
@@ -377,6 +473,38 @@ namespace TypesTest
     };
 } // namespace TypesTest
 
+TEST_P(GAPI_Streaming, SmokeTest_AutoMeta_VideoArray)
+{
+    cv::GMat in_m;
+    cv::GArray<int> in_v;
+    cv::GMat out_m = TypesTest::AddV::on(in_m, in_v) - in_m;
+
+    // Run pipeline
+    auto testc = cv::GComputation(cv::GIn(in_m, in_v), cv::GOut(out_m))
+                    .compileStreaming(cv::compile_args(cv::gapi::kernels<TypesTest::OCVAddV>()));
+
+    cv::Mat tmp;
+    // Test with one video source and vector
+    auto in_src = gapi::wip::make_src<cv::gapi::wip::GCaptureSource>(findDataFile("cv/video/768x576.avi"));
+    std::vector<int> first_in_vec(768*3, 1);
+    testc.setSource(cv::gin(in_src, first_in_vec));
+    testc.start();
+
+    std::size_t test_frames = 0u;
+    while (testc.pull(cv::gout(tmp))) test_frames++;
+    EXPECT_EQ(100u, test_frames);
+
+    // Now test with another one
+    in_src = gapi::wip::make_src<cv::gapi::wip::GCaptureSource>(findDataFile("cv/video/1920x1080.avi"));
+    std::vector<int> second_in_vec(1920*3, 1);
+    testc.setSource(cv::gin(in_src, second_in_vec));
+    testc.start();
+
+    test_frames = 0u;
+    while (testc.pull(cv::gout(tmp))) test_frames++;
+    EXPECT_EQ(165u, test_frames);
+}
+
 TEST(GAPI_Streaming_Types, InputScalar)
 {
     // This test verifies if Streaming works with Scalar data @ input.
@@ -662,7 +790,6 @@ struct GAPI_Streaming_Unit: public ::testing::Test {
     {
         initTestDataPath();
 
-
         const auto a_desc = cv::descr_of(m);
         const auto b_desc = cv::descr_of(m);
         sc  = cc.compileStreaming(a_desc, b_desc);
@@ -672,10 +799,17 @@ struct GAPI_Streaming_Unit: public ::testing::Test {
 
 TEST_F(GAPI_Streaming_Unit, TestTwoVideoSourcesFail)
 {
-    // FIXME: Meta check doesn't fail here (but ideally it should)
     const auto c_ptr = gapi::wip::make_src<cv::gapi::wip::GCaptureSource>(findDataFile("cv/video/768x576.avi"));
+    auto c_desc = cv::GMatDesc{CV_8U,3,{768,576}};
+    auto m_desc = cv::descr_of(m);
+
+    sc = cc.compileStreaming(c_desc, m_desc);
     EXPECT_NO_THROW(sc.setSource(cv::gin(c_ptr, m)));
+
+    sc = cc.compileStreaming(m_desc, c_desc);
     EXPECT_NO_THROW(sc.setSource(cv::gin(m, c_ptr)));
+
+    sc = cc.compileStreaming(c_desc, c_desc);
     EXPECT_ANY_THROW(sc.setSource(cv::gin(c_ptr, c_ptr)));
 }
 
@@ -823,5 +957,4 @@ TEST_F(GAPI_Streaming_Unit, SetSource_After_Completion)
     EXPECT_EQ(0., cv::norm(out, out_ref, cv::NORM_INF));
 }
 
-
 } // namespace opencv_test