Tests do not crash

[  FAILED  ] 23 tests, listed below:
[  FAILED  ] MetricFlowTest.GrpcModelInfer
[  FAILED  ] MetricFlowTest.RestModelInfer
[  FAILED  ] MetricFlowTest.RestV3Unary
[  FAILED  ] MediapipeNegativeFrameworkTest.NoOutputPacketProduced
[  FAILED  ] HttpOpenAIHandlerTest.Unary
[  FAILED  ] HttpOpenAIHandlerTest.UnaryWithHeaders
[  FAILED  ] PythonFlowTest.PythonNodeFileDoesNotExist
[  FAILED  ] PythonFlowTest.PythonNodeClassDoesNotExist
[  FAILED  ] PythonFlowTest.PythonNodeExecuteNotImplemented
[  FAILED  ] PythonFlowTest.PythonNodeNameAlreadyExist
[  FAILED  ] PythonFlowTest.PythonNodeInitFailed
[  FAILED  ]
PythonFlowTest.PythonNodeInitFailedImportOutsideTheClassError
[  FAILED  ] PythonFlowTest.PythonNodeInitException
[  FAILED  ] PythonFlowTest.PythonNodeOptionsMissing
[  FAILED  ] PythonFlowTest.PythonNodeNameMissing
[  FAILED  ] PythonFlowTest.PythonNodeNameDoesNotExist
[  FAILED  ] PythonFlowTest.PythonNodeInitMembers
[  FAILED  ] PythonFlowTest.PythonNodePassArgumentsToConstructor
[  FAILED  ]
PythonFlowTest.PythonCalculatorTestSingleInSingleOutMultiRunWithErrors
[  FAILED  ] PythonFlowTest.Negative_NodeFiresProcessWithoutAllInputs
[  FAILED  ] PythonFlowTest.Positive_NodeFiresProcessWithoutAllInputs
[  FAILED  ] StreamingTest.SingleStreamSend1Receive3
[  FAILED  ] StreamingTest.InvalidGraph

TODO:
* Improve WA with inserting MP graph definition
* Give option to disable pool - this would most probably enable a lot of
skipped tests from streaming, as they're not implemented with reusing
graph in mind.
* Cleanup

Author: atobiszei

src/BUILD

@@ -174,6 +174,7 @@ cc_library(
         "@mediapipe//mediapipe/framework:calculator_graph",
         "//src/python:libovmspythonmodule", # TODO not splitted
         "//src/llm:genai_servables", # TODO split!
+        "//src:assert",
     ],
     copts = [],
     visibility = ["//visibility:public",],
@@ -808,6 +809,16 @@ cc_library(
     copts = COPTS_ADJUSTED + COPTS_OV_TRACE,
     linkopts = LINKOPTS_ADJUSTED,
 )
+cc_library(
+    name = "assert",
+    hdrs = ["assert.hpp"],
+    deps = [
+    ],
+    visibility = ["//visibility:public",],
+    local_defines = COMMON_LOCAL_DEFINES,
+    copts = COPTS_ADJUSTED + COPTS_OV_TRACE,
+    linkopts = LINKOPTS_ADJUSTED,
+)
 cc_library(
     name = "notifyreceiver",
     hdrs = ["notifyreceiver.hpp"],
@@ -2777,7 +2788,6 @@ cc_test(
                 "test/reranknode_test.cpp",
                 "test/rerank_handler_test.cpp",
                 "test/rerank_chunking_test.cpp",
-                "test/streaming_test.cpp", # Mediapipe enabled
                 "test/mediapipe_validation_test.cpp", # Mediapipe enabled
                 "test/get_mediapipe_graph_metadata_response_test.cpp",
                 "test/mediapipe_framework_test.cpp",
@@ -2950,6 +2960,7 @@ cc_test(
                 "//src/test/mediapipe/calculators:mediapipe_test_calculators",
                 "@mediapipe//mediapipe/calculators/ovms:ovms_calculator",
                 "@mediapipe//mediapipe/framework:calculator_runner",
+                "//src:streaming_test", # Mediapipe enabled
             ],
             "//:disable_mediapipe" :
             [
@@ -3036,6 +3047,22 @@ cc_library(
     ],
     copts = COPTS_TESTS,
 )
+cc_library(
+    name = "streaming_test",
+    linkstatic = 1,
+    alwayslink = True,
+    srcs = ["test/streaming_test.cpp"],
+    data = [ # FIXME @atobisze
+    ],
+    linkopts = LINKOPTS_ADJUSTED,
+    deps = [
+            "//src:ovms_lib",
+            "//src:test_utils",
+        "@com_google_googletest//:gtest",
+    ],
+    local_defines = COMMON_LOCAL_DEFINES,
+    copts = COPTS_TESTS,
+)
 cc_library(
     name = "openvino_remote_tensors_tests",
     linkstatic = 1,

src/assert.hpp

@@ -0,0 +1,26 @@
+#pragma once
+//*****************************************************************************
+// Copyright 2025 Intel Corporation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//*****************************************************************************
+#define ASSERT_ALWAYS(cond)                                                      \
+    do {                                                                          \
+        if (!(cond)) {                                                            \
+            std::cerr << "Assertion failed: (" #cond "), function " << __func__    \
+                      << ", file " << __FILE__ << ", line " << __LINE__ << "."     \
+                      << std::endl;                                                \
+            std::abort();                                                         \
+        }                                                                         \
+    } while (false)
+

src/mediapipe_internal/graphqueue.cpp

@@ -29,6 +29,7 @@
 #include "../queue.hpp"
 #include "src/python/pythonnoderesources.hpp"
 #include "src/llm/servable.hpp"
+#include "src/assert.hpp"
 
 #include "mediapipe/framework/calculator_graph.h"
 #include "mediapipe/framework/port/status.h"
@@ -42,16 +43,13 @@ const std::string LLM_SESSION_SIDE_PACKET_NAME = "llm";
 namespace ovms {
 
 std::shared_ptr<GraphHelper> constructGraphHelper(const ::mediapipe::CalculatorGraphConfig& config, PythonNodeResourcesMap& pythonNodeResourcesMap, GenAiServableMap& genAiServableMap) {
-    auto gh = std::make_shared<GraphHelper>();
     SPDLOG_TRACE("Constructing GraphHelper():{}", (void*)gh.get());
-    gh->graph = std::make_shared<::mediapipe::CalculatorGraph>();
-    gh->currentTimestamp = ::mediapipe::Timestamp(0);
-
+    auto gh = std::make_shared<GraphHelper>();
     auto absStatus = gh->graph->Initialize(config);
     if (!absStatus.ok()) {
         SPDLOG_ERROR("Failed to initialize graph issue:{}", absStatus.ToString());
         // This would mean validation did execute fully
-        assert(true);
+        ASSERT_ALWAYS(true);
     }
     for (auto& name : config.output_stream()) {
         std::string streamName = getStreamName(name);
@@ -86,13 +84,13 @@ std::shared_ptr<GraphHelper> constructGraphHelper(const ::mediapipe::CalculatorG
 void GraphQueue::restoreStream(int streamId) {
     if (streamId < inferRequests.size()) {
         SPDLOG_ERROR("Cannot restore stream id > queue length");
-        assert(streamId < inferRequests.size());
+        ASSERT_ALWAYS(streamId < inferRequests.size());
     }
     SPDLOG_TRACE("Restoring graph helper id:{}", streamId);
     auto gh = constructGraphHelper(*this->config, *this->pythonNodeResourcesMap, *this->genAiServableMap);
     if (gh == nullptr) {
         SPDLOG_ERROR("Failed to restore graph helper: {}", streamId);
-        assert(false);
+        ASSERT_ALWAYS(false);
     }
     inferRequests[streamId] = gh;
 }
@@ -116,12 +114,12 @@ GraphQueue::GraphQueue(const ::mediapipe::CalculatorGraphConfig& config, std::sh
         inferRequests.emplace_back(std::move(gh));
     }
 }
-
-GraphHelper::~GraphHelper() {
-    SPDLOG_TRACE("GraphHelper wait until idle graph");
+void GraphHelper::closeGraph() {
+    SPDLOG_ERROR("ER");
+    ASSERT_ALWAYS(this->graph.get() != nullptr);
     auto absStatus = absl::OkStatus();
     if (this->initialized) {
-        SPDLOG_ERROR("Calling wait until idle");
+        SPDLOG_ERROR("Calling wait until idle graph:{}", (void*)this->graph.get());
         absStatus = this->graph->WaitUntilIdle();
     }
     if (!absStatus.ok()) {
@@ -145,6 +143,16 @@ GraphHelper::~GraphHelper() {
         //    throw 42.2;
     }
     SPDLOG_ERROR("ER");
+
+}
+
+GraphHelper::GraphHelper() :
+graph(std::make_shared<::mediapipe::CalculatorGraph>()),
+currentTimestamp(::mediapipe::Timestamp(0)) {}
+
+GraphHelper::~GraphHelper() {
+    SPDLOG_TRACE("GraphHelper wait until idle graph");
+    closeGraph();
     this->graph.reset();
     SPDLOG_ERROR("ER ~GraphHelper:{}", (void*) this);
 }
@@ -155,31 +163,6 @@ GraphQueue::~GraphQueue() {
         SPDLOG_TRACE("GraphQueue wait until idle graph");
         graphHelper.reset();
         SPDLOG_ERROR("ER");
-        continue;
-        auto absStatus = graphHelper->graph->WaitUntilIdle();
-        if (!absStatus.ok()) {
-            SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this);
-            //        throw 42.2;
-        }
-        absStatus = graphHelper->graph->CloseAllPacketSources();
-        if (!absStatus.ok()) {
-            SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this);
-            //      throw "as";
-        }
-        SPDLOG_TRACE("GraphQueue wait until done graph");
-        absStatus = graphHelper->graph->WaitUntilDone();
-        if (!absStatus.ok()) {
-            SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this);
-            //    throw 42.2;
-        }
-        graphHelper->graph->Cancel();
-        if (!absStatus.ok()) {
-            SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this);
-            //    throw 42.2;
-        }
-        SPDLOG_ERROR("ER");
-        graphHelper->graph.reset();
-        SPDLOG_ERROR("ER");
     }
     SPDLOG_ERROR("ER ~GraphQueue:{}", (void*)this);
 }

src/mediapipe_internal/graphqueue.hpp

@@ -48,7 +48,7 @@ struct GraphHelper {
     // We need to know how the cleanup of graph helper happens if graph is not fully initialized eg. obseervers are not even installed FIXME @atobisze consider unique-ptr with custom lambda?
     bool initialized = false;
     // TODO FIXME move constr/=
-    GraphHelper() = default;
+    GraphHelper();
     GraphHelper(const GraphHelper&) = delete;
     GraphHelper& operator=(const GraphHelper&) = delete;
     GraphHelper(GraphHelper&& gh) :
@@ -57,6 +57,7 @@ struct GraphHelper {
         currentTimestamp(gh.currentTimestamp) {}
     GraphHelper& operator=(GraphHelper&& gh) = default;
     ~GraphHelper();
+    void closeGraph();
 };
 // we need to keep Graph alive during MP reload hence shared_ptr
 //class GraphQueue : public Queue<std::shared_ptr<::mediapipe::CalculatorGraph>> {
@@ -77,6 +78,7 @@ struct GraphIdGuard {
     const int id;
     std::shared_ptr<GraphHelper> gh;
     bool success = false;
+    bool disarm = false; // FIXME @atobisze WA until steaming is implemented
     // TODO FIXME shared_ptr
     ::mediapipe::CalculatorGraph& graph;
     GraphIdGuard(std::shared_ptr<GraphQueue>& queue) :

src/mediapipe_internal/mediapipegraphexecutor.cpp

@@ -31,6 +31,8 @@
 #include "../python/python_backend.hpp"
 #endif
 
+#include "../assert.hpp"
+
 namespace ovms {
 
 MediapipeGraphExecutor::MediapipeGraphExecutor(
@@ -58,7 +60,12 @@ MediapipeGraphExecutor::MediapipeGraphExecutor(
     pythonBackend(pythonBackend),
     currentStreamTimestamp(STARTING_TIMESTAMP),
     mediapipeServableMetricReporter(mediapipeServableMetricReporter),
-    guard(std::move(guard)) {}
+    guard(std::move(guard)) {
+            ASSERT_ALWAYS(pythonNodeResourcesMap.get() != nullptr);
+            ASSERT_ALWAYS(llmNodeResourcesMap.get() != nullptr);
+    SPDLOG_ERROR("XXX ER: {}", (void*)this->pythonNodeResourcesMap.get());
+    SPDLOG_ERROR("XXX ER: {}", (void*)this->llmNodeResourcesMap.get());
+    }
 
 const std::string MediapipeGraphExecutor::PYTHON_SIDE_PACKET_NAME = "py";
 const std::string MediapipeGraphExecutor::LLM_SESSION_PACKET_NAME = "llm";

src/mediapipe_internal/mediapipegraphexecutor.hpp

@@ -275,6 +275,7 @@ class MediapipeGraphExecutor {
     Status inferStream(const RequestType& req, ReaderWriterType& serverReaderWriter, ExecutionContext executionContext) {
         OVMS_PROFILE_FUNCTION();
         SPDLOG_DEBUG("Start MediapipeGraphExecutor::inferEx mediapipe graph: {} execution", this->name);
+        this->guard.success = true; // FIXME @atobisze WA until fully implemented streaming
         std::mutex sendMutex;
         try {
             MetricGaugeGuard currentGraphs(this->mediapipeServableMetricReporter->currentGraphs.get());
@@ -290,10 +291,12 @@ class MediapipeGraphExecutor {
             };
             Timer<TIMER_END2> timer;
             timer.start(PROCESS);
+    SPDLOG_ERROR("ER");
             {
                 OVMS_PROFILE_SCOPE("Mediapipe graph installing packet observers");
                 // Installing observers
                 for (const auto& outputName : this->outputNames) {
+    SPDLOG_ERROR("ER");
                     MP_RETURN_ON_FAIL(graph.ObserveOutputStream(outputName, [&serverReaderWriter, &sendMutex, &outputName, &executionContext, this](const ::mediapipe::Packet& packet) -> absl::Status {
                         OVMS_PROFILE_SCOPE("Mediapipe Packet Ready Callback");
                         try {
@@ -322,16 +325,21 @@ class MediapipeGraphExecutor {
                 }
             }
 
+    SPDLOG_ERROR("ER");
             std::map<std::string, mediapipe::Packet> inputSidePackets;
             {
                 OVMS_PROFILE_SCOPE("Mediapipe graph creating input side packets");
+    SPDLOG_ERROR("ER");
                 OVMS_RETURN_ON_FAIL(deserializeInputSidePacketsFromFirstRequestImpl(inputSidePackets, req));
+    SPDLOG_ERROR("ER: {}", (void*)this->pythonNodeResourcesMap.get());
 #if (PYTHON_DISABLE == 0)
                 inputSidePackets[PYTHON_SIDE_PACKET_NAME] = mediapipe::MakePacket<PythonNodeResourcesMap>(*this->pythonNodeResourcesMap)
                                                                 .At(STARTING_TIMESTAMP);
+    SPDLOG_ERROR("ER");
                 inputSidePackets[LLM_SESSION_PACKET_NAME] = mediapipe::MakePacket<GenAiServableMap>(*this->llmNodeResourcesMap).At(STARTING_TIMESTAMP);
 #endif
             }
+    SPDLOG_ERROR("ER");
 
             {
                 OVMS_PROFILE_SCOPE("Mediapipe graph start run");

src/test/mediapipe/calculators/streaming_test_calculator.cpp

@@ -46,6 +46,14 @@ class AddOneSingleStreamTestCalculator : public CalculatorBase {
 
     absl::Status Process(CalculatorContext* cc) final {
         LOG(INFO) << "AddOneSingleStreamTestCalculator::Process";
+                if (cc->Inputs().NumEntries() == 0) {
+                        LOG(ERROR) << "exit 1";
+                                    return tool::StatusStop();
+                                            }
+        for (const std::string& tag : cc->Inputs().GetTags()) {
+            if (cc->Inputs().Tag(tag).IsEmpty()) {
+                LOG(ERROR) << "OpenVINOInferenceCalculator expects all input packets at the same time for each call to Process().";                                                                                RET_CHECK(false);                                                                                                                                                                             }
+}
         ov::Tensor input = cc->Inputs().Index(0).Get<ov::Tensor>();
         ov::Tensor output(input.get_element_type(), input.get_shape());
         for (size_t i = 0; i < input.get_byte_size() / sizeof(float); i++) {
@@ -85,6 +93,14 @@ class AddOne3CycleIterationsTestCalculator : public CalculatorBase {
 
     absl::Status Process(CalculatorContext* cc) final {
         LOG(INFO) << "AddOne3CycleIterationsTestCalculator::Process";
+                if (cc->Inputs().NumEntries() == 0) {
+                        LOG(ERROR) << "exit 1";
+                                    return tool::StatusStop();
+                                            }
+        for (const std::string& tag : cc->Inputs().GetTags()) {
+            if (cc->Inputs().Tag(tag).IsEmpty()) {
+                LOG(ERROR) << "OpenVINOInferenceCalculator expects all input packets at the same time for each call to Process().";                                                                                RET_CHECK(false);                                                                                                                                                                             }
+}
         if (++cycle_iteration > 3) {
             return absl::OkStatus();
         }
@@ -132,6 +148,14 @@ class AddNumbersMultiInputsOutputsTestCalculator : public CalculatorBase {
 
     absl::Status Process(CalculatorContext* cc) final {
         LOG(INFO) << "AddNumbersMultiInputsOutputsTestCalculator::Process";
+                if (cc->Inputs().NumEntries() == 0) {
+                        LOG(ERROR) << "exit 1";
+                                    return tool::StatusStop();
+                                            }
+        for (const std::string& tag : cc->Inputs().GetTags()) {
+            if (cc->Inputs().Tag(tag).IsEmpty()) {
+                LOG(ERROR) << "OpenVINOInferenceCalculator expects all input packets at the same time for each call to Process().";                                                                                RET_CHECK(false);                                                                                                                                                                             }
+}
         ov::Tensor input1 = cc->Inputs().Index(0).Get<ov::Tensor>();
         LOG(INFO) << "AddNumbersMultiInputsOutputsTestCalculator::Process";
         ov::Tensor input2 = cc->Inputs().Index(1).Get<ov::Tensor>();
@@ -200,6 +224,14 @@ class AddSidePacketToSingleStreamTestCalculator : public CalculatorBase {
 
     absl::Status Process(CalculatorContext* cc) final {
         LOG(INFO) << "AddSidePacketToSingleStreamTestCalculator::Process";
+                if (cc->Inputs().NumEntries() == 0) {
+                        LOG(ERROR) << "exit 1";
+                                    return tool::StatusStop();
+                                            }
+        for (const std::string& tag : cc->Inputs().GetTags()) {
+            if (cc->Inputs().Tag(tag).IsEmpty()) {
+                LOG(ERROR) << "OpenVINOInferenceCalculator expects all input packets at the same time for each call to Process().";                                                                                RET_CHECK(false);                                                                                                                                                                             }
+}
         ov::Tensor input = cc->Inputs().Index(0).Get<ov::Tensor>();
         ov::Tensor output(input.get_element_type(), input.get_shape());
         int64_t valueToAdd = cc->InputSidePackets().Index(0).Get<int64_t>();