Merge branch 4.x

modules/gapi/src/api/gbackend.cpp

@@ -36,7 +36,6 @@ cv::gapi::GBackend::Priv::compile(const ade::Graph&,
 {
     // ...and this method is here for the same reason!
     GAPI_Error("InternalError");
-    return {};
 }
 
 std::unique_ptr<cv::gimpl::GIslandExecutable>
@@ -224,7 +223,6 @@ void bindOutArg(Mag& mag, const RcDesc &rc, const GRunArgP &arg, HandleRMat hand
 
     default:
         util::throw_error(std::logic_error("Unsupported GShape type"));
-        break;
     }
 }
 
@@ -256,7 +254,6 @@ void resetInternalData(Mag& mag, const Data &d)
 
     default:
         util::throw_error(std::logic_error("Unsupported GShape type"));
-        break;
     }
 }
 
@@ -284,7 +281,6 @@ cv::GRunArg getArg(const Mag& mag, const RcDesc &ref)
                        mag.meta<cv::MediaFrame>().at(ref.id));
     default:
         util::throw_error(std::logic_error("Unsupported GShape type"));
-        break;
     }
 }
 
@@ -327,7 +323,6 @@ cv::GRunArgP getObjPtr(Mag& mag, const RcDesc &rc, bool is_umat)
 
     default:
         util::throw_error(std::logic_error("Unsupported GShape type"));
-        break;
     }
 }
 
@@ -359,7 +354,6 @@ void writeBack(const Mag& mag, const RcDesc &rc, GRunArgP &g_arg)
 
     default:
         util::throw_error(std::logic_error("Unsupported GShape type"));
-        break;
     }
 }
 

modules/gapi/src/compiler/gislandmodel.hpp

@@ -192,6 +192,7 @@ class GIslandEmitter
 public:
     // Obtain next value from the emitter
     virtual bool pull(GRunArg &) = 0;
+    virtual void halt() = 0;
     virtual ~GIslandEmitter() = default;
 };
 

modules/gapi/src/executor/gstreamingexecutor.cpp

@@ -41,6 +41,10 @@ using namespace cv::gimpl::stream;
 class VideoEmitter final: public cv::gimpl::GIslandEmitter {
     cv::gapi::wip::IStreamSource::Ptr src;
 
+    virtual void halt() override {
+        src->halt();
+    }
+
     virtual bool pull(cv::GRunArg &arg) override {
         // FIXME: probably we can maintain a pool of (then) pre-allocated
         // buffers to avoid runtime allocations.
@@ -62,6 +66,10 @@ public:
 class ConstEmitter final: public cv::gimpl::GIslandEmitter {
     cv::GRunArg m_arg;
 
+    virtual void halt() override {
+        // Not used here, but in fact can be used.
+    }
+
     virtual bool pull(cv::GRunArg &arg) override {
         arg = const_cast<const cv::GRunArg&>(m_arg); // FIXME: variant workaround
         return true;
@@ -1918,6 +1926,11 @@ void cv::gimpl::GStreamingExecutor::stop()
     for (auto &q : m_emitter_queues) {
         q.push(stream::Cmd{stream::Stop{}});
     }
+    // Also kindly ask emitter object to halt to break the blocking src->pull()
+    // loop
+    for (auto &nh : m_emitters) {
+        m_gim.metadata(nh).get<Emitter>().object->halt();
+    }
 
     // Pull messages from the final queue to ensure completion
     Cmd cmd;

modules/gapi/src/streaming/queue_source.cpp

@@ -0,0 +1,98 @@
+// 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) 2023 Intel Corporation
+
+#include <chrono>
+#include <atomic>
+
+#include <ade/util/zip_range.hpp>
+
+#include <opencv2/gapi/streaming/queue_source.hpp>
+#include <opencv2/gapi/streaming/meta.hpp>
+
+#include "executor/conc_queue.hpp"
+
+namespace cv {
+namespace gapi {
+namespace wip {
+
+class QueueSourceBase::Priv {
+public:
+    explicit Priv(const cv::GMetaArg &meta) {
+        m = meta;
+        halted = false;
+    }
+
+    cv::GMetaArg m;
+    cv::gapi::own::concurrent_bounded_queue<cv::GRunArg> q;
+    int64_t c = 0;
+    std::atomic<bool> halted;
+};
+
+QueueSourceBase::QueueSourceBase(const cv::GMetaArg &m)
+    : m_priv(new Priv(m)) {
+}
+
+void QueueSourceBase::push(Data &&data) {
+
+    // Tag data with seq_id/ts
+    const auto now = std::chrono::system_clock::now();
+    const auto dur = std::chrono::duration_cast<std::chrono::microseconds>
+        (now.time_since_epoch());
+    data.meta[cv::gapi::streaming::meta_tag::timestamp] = int64_t{dur.count()};
+    data.meta[cv::gapi::streaming::meta_tag::seq_id]    = int64_t{m_priv->c++};
+
+    m_priv->q.push(data);
+}
+
+bool QueueSourceBase::pull(Data &data) {
+    m_priv->q.pop(data);
+
+    if (m_priv->halted) {
+        return false;
+    }
+    return true;
+}
+
+void QueueSourceBase::halt() {
+    m_priv->halted.store(true);
+    m_priv->q.push(cv::GRunArg{});
+}
+
+cv::GMetaArg QueueSourceBase::descr_of() const {
+    return m_priv->m;
+}
+
+QueueInput::QueueInput(const cv::GMetaArgs &args) {
+    for (auto &&m : args) {
+        m_sources.emplace_back(new cv::gapi::wip::QueueSourceBase(m));
+    }
+}
+
+void QueueInput::push(cv::GRunArgs &&args) {
+    GAPI_Assert(m_sources.size() == args.size());
+    for (auto && it : ade::util::zip(ade::util::toRange(m_sources),
+                                     ade::util::toRange(args)))
+    {
+        auto &src = std::get<0>(it);
+        auto &obj = std::get<1>(it);
+
+        Data d;
+        d = obj;
+        src->push(std::move(d));
+    }
+}
+
+QueueInput::operator cv::GRunArgs () {
+    cv::GRunArgs args;
+    for (auto &&s : m_sources) {
+        args.push_back(s->ptr());
+    }
+    return args;
+}
+
+} // wip
+} // gapi
+} // cv