Streaming (#18)

* Adding support for streaming

* Updating kernel attributes

* Updates to kernel members

* Minor readability update

* Current progress on streaming

* Fix to test env in bazelrc to lessen overhead

* Enabling pipelining conformance tests

* Successful streaming conformance

* One more attempt to enable pipelining

* Fixing uninitialized parameter that caused overflow

Author: amikhail48

.bazelrc

@@ -11,7 +11,7 @@ build --experimental_ui_max_stdouterr_bytes=-1
 
 # Test settings
 test --test_output=errors  # Show errors from tests
-test --test_env=PATH=/usr/local/bin:/usr/bin:/bin  # Example of setting environment variables for tests
+# test --test_env=PATH=/usr/local/bin:/usr/bin:/bin  # Example of setting environment variables for tests
 
 # General settings
 build --jobs=4             # Use 4 parallel jobs for builds

framework/include/vx_graph.h

@@ -17,6 +17,7 @@
 #define VX_GRAPH_H
 
 #include <COREVX/execution_queue.hpp>
+#include <atomic>
 
 #include "vx_internal.h"
 #include "vx_reference.h"
@@ -171,6 +172,12 @@ class Graph : public Reference
     vx_status pipelineValidateRefsList(
         const vx_graph_parameter_queue_params_t graph_parameters_queue_param);
 
+    /**
+     * @brief Streaming loop function
+     *
+     */
+    void streamingLoop();
+
     /**
      * @brief Destruct function for the Graph object
      * @ingroup group_int_graph
@@ -207,9 +214,9 @@ class Graph : public Reference
         /*! \brief the max buffers that can be enqueued */
         vx_uint32 numBufs;
         /*! \brief The internal data ref queue */
-        ExecutionQueue<vx_reference, VX_INT_MAX_QUEUE_DEPTH> queue;
+        ExecutionQueue<vx_reference, VX_INT_MAX_PARAM_QUEUE_DEPTH> queue;
         /*! \brief references that can be queued into data ref queue */
-        vx_reference refs_list[VX_INT_MAX_QUEUE_DEPTH];
+        vx_reference refs_list[VX_INT_MAX_PARAM_QUEUE_DEPTH];
 #endif
     } parameters[VX_INT_MAX_PARAMS];
     /*! \brief The number of graph parameters. */
@@ -228,6 +235,16 @@ class Graph : public Reference
     /*! \brief The number of times to schedule a graph  */
     vx_size scheduleCount;
 #endif
+#ifdef OPENVX_USE_STREAMING
+    /*! \brief This indicates that the graph is streaming enabled */
+    std::atomic<vx_bool> isStreamingEnabled;
+    /*! \brief This indicates that the graph is currently streaming */
+    std::atomic<vx_bool> isStreaming;
+    /*! \brief The index of the trigger node */
+    vx_uint32 triggerNodeIndex;
+    /*! \brief The thread used for streaming */
+    vx_thread streamingThread;
+#endif
 };
 
 #endif /* VX_GRAPH_H */

framework/include/vx_internal.h

@@ -329,6 +329,10 @@ typedef struct vx_sem_t
  */
 typedef sem_t vx_sem_t;
 #endif
+/*! \brief A C++ STL thread
+ * \ingroup group_int_osal
+ */
+typedef std::thread vx_thread;
 /*! \brief A POSIX thread
  * \ingroup group_int_osal
  */

framework/include/vx_kernel.h

@@ -142,27 +142,27 @@ class Kernel : public Reference
      */
     static void printKernel(vx_kernel kernel);
 
-    /*! \brief */
+    /*! \brief The name of the kernel */
     vx_char        name[VX_MAX_KERNEL_NAME];
-    /*! \brief */
+    /*! \brief The kernel enum ID */
     vx_enum        enumeration;
-    /*! \brief */
+    /*! \brief The kernel function pointer */
     vx_kernel_f    function;
-    /*! \brief */
+    /*! \brief The kernel signature */
     vx_signature_t signature;
-    /*! Indicates that the kernel is not yet enabled. */
+    /*! \brief Indicates that the kernel is not yet enabled. */
     vx_bool        enabled;
-    /*! Indicates that this kernel is added by user. */
+    /*! \brief Indicates that this kernel is added by user. */
     vx_bool        user_kernel;
-    /*! \brief */
+    /*! \brief The kernel validate function pointer */
     vx_kernel_validate_f validate;
-    /*! \brief */
+    /*! \brief The kernel input validate function pointer */
     vx_kernel_input_validate_f validate_input;
-    /*! \brief */
+    /*! \brief The kernel output validate function pointer */
     vx_kernel_output_validate_f validate_output;
-    /*! \brief */
+    /*! \brief The kernel init function pointer */
     vx_kernel_initialize_f initialize;
-    /*! \brief */
+    /*! \brief The kernel deinit function pointer */
     vx_kernel_deinitialize_f deinitialize;
     /*! \brief The collection of attributes of a kernel */
     vx_kernel_attr_t attributes;
@@ -172,9 +172,15 @@ class Kernel : public Reference
     /*! \brief The tiling function pointer interface */
     vx_tiling_kernel_f tilingfast_function;
     vx_tiling_kernel_f tilingflexible_function;
-#endif
+#endif /* OPENVX_KHR_TILING */
     /*! \brief The pointer to the kernel object deinitializer. */
     vx_kernel_object_deinitialize_f kernel_object_deinitialize;
+    /*! \brief The kernel's input depth required to start */
+    vx_uint32 input_depth;
+    /*! \brief The kernel's output depth required to start */
+    vx_uint32 output_depth;
+    /*! \brief Indicates whether kernel has piped up */
+    vx_uint32 pipeUpCounter;
 };
 
 #endif /* VX_KERNEL_H */

framework/include/vx_node.h

@@ -17,6 +17,7 @@
 #define VX_NODE_H
 
 #include <VX/vx.h>
+#include <vector>
 
 /*!
  * \file
@@ -132,6 +133,8 @@ class Node : public Reference
     vx_bool             is_replicated;
     /*! \brief The replicated parameters flags */
     vx_bool             replicated_flags[VX_INT_MAX_PARAMS];
+    /*! \brief The node state */
+    vx_node_state_e     state;
 };
 
 #endif /* VX_NODE_H */

framework/src/vx_graph.cpp

@@ -230,21 +230,33 @@ void vxContaminateGraphs(vx_reference ref)
 /* INTERNAL FUNCTIONS */
 /******************************************************************************/
 
-Graph::Graph(vx_context context, vx_reference scope) : Reference(context, VX_TYPE_GRAPH, scope),
-nodes(),
-perf(),
-numNodes(0),
-heads(),
-numHeads(0),
-state(VX_FAILURE),
-verified(vx_false_e),
-reverify(vx_false_e),
-lock(),
-parameters(),
-numParams(0),
-shouldSerialize(vx_false_e),
-parentGraph(nullptr),
-delays()
+Graph::Graph(vx_context context, vx_reference scope)
+    : Reference(context, VX_TYPE_GRAPH, scope),
+      nodes(),
+      perf(),
+      numNodes(0),
+      heads(),
+      numHeads(0),
+      state(VX_FAILURE),
+      verified(vx_false_e),
+      reverify(vx_false_e),
+      lock(),
+      parameters(),
+      numParams(0),
+      shouldSerialize(vx_false_e),
+      parentGraph(nullptr),
+      delays(),
+      scheduleMode(VX_GRAPH_SCHEDULE_MODE_NORMAL),
+#ifdef OPENVX_USE_PIPELINING
+      numEnqueableParams(0),
+      scheduleCount(0),
+#endif /* OPENVX_USE_PIPELINING */
+#ifdef OPENVX_USE_STREAMING
+      isStreamingEnabled(vx_false_e),
+      isStreaming(vx_false_e),
+      triggerNodeIndex(0),
+      streamingThread()
+#endif /* OPENVX_USE_STREAMING */
 {
 }
 
@@ -1723,6 +1735,35 @@ vx_status Graph::pipelineValidateRefsList(
     return status;
 }
 
+void Graph::streamingLoop()
+{
+#ifdef OPENVX_USE_STREAMING
+    while (isStreaming)
+    {
+        /* Wait for trigger node event if set */
+        // if (triggerNodeIndex != UINT32_MAX)
+        // {
+        //     /* 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));
+        //         /* Allow clean exit */
+        //         if (!isStreaming) return;
+        //     }
+        //     /* Reset the event for the next iteration */
+        //     nodes[triggerNodeIndex]->executed = vx_false_e;
+        // }
+
+        /* Schedule and wait for the graph */
+        vx_status status = vxScheduleGraph(this);
+        if (status != VX_SUCCESS) break;
+        status = vxWaitGraph(this);
+        if (status != VX_SUCCESS) break;
+    }
+#endif /* OPENVX_USE_STREAMING */
+}
+
 void Graph::destruct()
 {
     while (numNodes)
@@ -2068,7 +2109,8 @@ VX_API_ENTRY vx_status VX_API_CALL vxVerifyGraph(vx_graph graph)
                 {
                     if (((graph->nodes[n]->kernel->signature.directions[p] == VX_BIDIRECTIONAL) ||
                          (graph->nodes[n]->kernel->signature.directions[p] == VX_INPUT)) &&
-                        (graph->nodes[n]->parameters[p] != nullptr))
+                        (graph->nodes[n]->parameters[p] != nullptr) &&
+                        (graph->nodes[n]->kernel->validate_input != nullptr))
                     {
                         vx_status input_validation_status = graph->nodes[n]->kernel->validate_input((vx_node)graph->nodes[n], p);
                         if (input_validation_status != VX_SUCCESS)
@@ -2100,25 +2142,30 @@ VX_API_ENTRY vx_status VX_API_CALL vxVerifyGraph(vx_graph graph)
                         if (graph->setupOutput(n, p, &vref, &metas[p], &status, &num_errors) ==
                             vx_false_e)
                             break;
-                        output_validation_status = graph->nodes[n]->kernel->validate_output(
-                            (vx_node)graph->nodes[n], p, metas[p]);
-                        if (output_validation_status == VX_SUCCESS)
+                        if (graph->nodes[n]->kernel->validate_output != nullptr)
                         {
-                            if (graph->postprocessOutput(n, p, &vref, metas[p], &status,
-                                                         &num_errors) == vx_false_e)
+                            output_validation_status = graph->nodes[n]->kernel->validate_output(
+                                (vx_node)graph->nodes[n], p, metas[p]);
+                            if (output_validation_status == VX_SUCCESS)
                             {
-                                break;
+                                if (graph->postprocessOutput(n, p, &vref, metas[p], &status,
+                                                             &num_errors) == vx_false_e)
+                                {
+                                    break;
+                                }
+                            }
+                            else
+                            {
+                                status = output_validation_status;
+                                vxAddLogEntry(reinterpret_cast<vx_reference>(graph), status,
+                                              "Node %s: parameter[%u] failed output validation! "
+                                              "(status = %d)\n",
+                                              graph->nodes[n]->kernel->name, p, status);
+                                VX_PRINT(VX_ZONE_ERROR,
+                                         "Failed on validation of output parameter[%u] on kernel "
+                                         "%s, status=%d\n",
+                                         p, graph->nodes[n]->kernel->name, status);
                             }
-                        }
-                        else
-                        {
-                            status = output_validation_status;
-                            vxAddLogEntry(reinterpret_cast<vx_reference>(graph), status, "Node %s: parameter[%u] failed output validation! (status = %d)\n",
-                                          graph->nodes[n]->kernel->name, p, status);
-                            VX_PRINT(VX_ZONE_ERROR,"Failed on validation of output parameter[%u] on kernel %s, status=%d\n",
-                                     p,
-                                     graph->nodes[n]->kernel->name,
-                                     status);
                         }
                     }
                 }
@@ -2511,6 +2558,7 @@ static vx_status vxExecuteGraph(vx_graph graph, vx_uint32 depth)
     vx_uint32 next_nodes[VX_INT_MAX_REF];
     vx_uint32 left_nodes[VX_INT_MAX_REF];
     vx_context context = vxGetContext((vx_reference)graph);
+    vx_uint32 max_pipeup_depth = 1;
     (void)depth;
 
 #if defined(OPENVX_USE_SMP)
@@ -2640,6 +2688,34 @@ static vx_status vxExecuteGraph(vx_graph graph, vx_uint32 depth)
                              next_nodes[n],
                              target->name, node->kernel->name);
 
+                    /* Check for pipeup phase:
+                     * If this is the first time we are executing the graph, we need to pipeup
+                     * all nodes with kernels in the graph that need pipeup of refs.
+                     */
+                    max_pipeup_depth = std::max(
+                        {max_pipeup_depth, node->kernel->input_depth, node->kernel->output_depth});
+                    if (node->kernel->pipeUpCounter < max_pipeup_depth - 1)
+                    {
+                        node->state = VX_NODE_STATE_PIPEUP;
+                        std::cout << "max_pipeup_depth: " << max_pipeup_depth << std::endl;
+                        node->kernel->pipeUpCounter++;
+                        // Retain input buffers during PIPEUP
+                        for (vx_uint32 i = 0; i < node->kernel->output_depth - 1; i++)
+                        {
+                            action = target->funcs.process(target, &node, 0, 1);
+                            node->kernel->pipeUpCounter++;
+                        }
+                        // For source nodes, provide new output buffers during PIPEUP
+                        for (vx_uint32 i = 0; i < node->kernel->input_depth - 1; i++)
+                        {
+                            action = target->funcs.process(target, &node, 0, 1);
+                            node->kernel->pipeUpCounter++;
+                        }
+                    }
+
+                    /* If this node was in pipeup, update its state */
+                    node->state = VX_NODE_STATE_STEADY;
+
                     action = target->funcs.process(target, &node, 0, 1);
 
                     VX_PRINT(VX_ZONE_GRAPH, "Returned Node[%u] %s:%s Action %d\n",