Adding integration tests for pipelining

Author: amikhail48

.github/workflows/ci.yml

@@ -38,7 +38,7 @@ jobs:
           curl -fsSL https://ollama.com/install.sh | sh
           ollama serve &
           sleep 5
-          ollama pull llama3:8b
+          ollama pull llama3.2:3b
           curl -s http://localhost:11434/api/tags || exit 1
 
       # Checks-out your repository under $GITHUB_WORKSPACE, which is the CWD for

COPYRIGHT

@@ -0,0 +1 @@
+Copyright (c) 2025 Edge AI, LLC. All rights reserved.

framework/src/vx_graph.cpp

@@ -253,7 +253,7 @@ Graph::Graph(vx_context context, vx_reference scope)
 #ifdef OPENVX_USE_STREAMING
       isStreamingEnabled(vx_false_e),
       isStreaming(vx_false_e),
-      triggerNodeIndex(0),
+      triggerNodeIndex(UINT32_MAX),
       streamingThread(),
 #endif /* OPENVX_USE_STREAMING */
       scheduleMode(VX_GRAPH_SCHEDULE_MODE_NORMAL)
@@ -1743,11 +1743,12 @@ void Graph::streamingLoop()
         /* Wait for trigger node event if set */
         // if (triggerNodeIndex != UINT32_MAX)
         // {
+        //     VX_PRINT(VX_ZONE_INFO, "Trigger node defined -- waiting on it to trigger\n");
         //     /* Wait for the trigger node to complete */
         //     while (!nodes[triggerNodeIndex]->executed)
         //     {
         //         std::cout << "Waiting for trigger node to complete" << std::endl;
-        //         std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        //         std::this_thread::sleep_for(std::chrono::milliseconds(10));
         //         /* Allow clean exit */
         //         if (!isStreaming) return;
         //     }
@@ -1999,7 +2000,7 @@ VX_API_ENTRY vx_status VX_API_CALL vxVerifyGraph(vx_graph graph)
                         VX_PRINT(VX_ZONE_ERROR, "Node " VX_FMT_REF " (%s) Parameter[%u] was required and not supplied!\n",
                             graph->nodes[n],
                             graph->nodes[n]->kernel->name,p);
-                        status = VX_ERROR_NOT_SUFFICIENT;
+                        status = VX_ERROR_INVALID_PARAMETERS;
                         num_errors++;
                     }
                     else if (graph->nodes[n]->parameters[p]->internal_count == 0)

framework/src/vx_graph_pipeline.cpp

@@ -178,7 +178,11 @@ VX_API_ENTRY vx_status VX_API_CALL vxGraphParameterEnqueueReadyRef(vx_graph grap
                     graph->parameters[i].queue.movePendingToReady();
                 }
 
-                if (graph->scheduleMode == VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO)
+                if (graph->scheduleMode == VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO
+#ifdef OPENVX_USE_STREAMING
+                    && vx_false_e == graph->isStreaming
+#endif
+                )
                 {
                     /* Schedule the graph */
                     status = vxScheduleGraph(graph);

kernels/ai_server/chatbot.hpp

@@ -15,7 +15,7 @@
 #include <curl/curl.h>
 #include <nlohmann/json.hpp>
 
-#define DEFAULT_MODEL "llama3:8b"
+#define DEFAULT_MODEL "llama3.2:3b"
 #define SERVER_URL "http://localhost:11434"
 #define API_KEY "hardcoded-api-key"
 

tests/integration_test/BUILD

@@ -65,6 +65,35 @@ cc_test(
     size="small"
 )
 
+cc_test(
+    name="test_pipelining",
+    srcs=[
+        "test_pipelining.cpp"
+    ],
+    deps=[
+        "//:corevx",
+        "@googletest//:gtest_main",
+        "//targets/ai_server:imported_openvx_ai_server",
+        "//targets/c_model:imported_openvx_c_model",
+        "//targets/debug:imported_openvx_debug",
+        "//targets/extras:imported_openvx_extras",
+        "//targets/liteRT:imported_openvx_liteRT",
+        "//targets/opencl:imported_openvx_opencl",
+        "//targets/onnxRT:imported_openvx_onnxRT",
+        "//targets/executorch:imported_openvx_torch",
+
+    ],
+    linkopts=select({
+        "@platforms//os:linux": ["-Wl,-rpath,$ORIGIN"],
+        "@platforms//os:macos": ["-Wl,-rpath,@executable_path"],
+        "//conditions:default": [],
+    }),
+    data=[
+        "//tests/raw:models",
+    ],
+    size="small"
+)
+
 cc_test(
     name = "test_tflite",
     srcs = [