Support for large samples sizes / signal lists

CMakeLists.txt

@@ -103,7 +103,7 @@ pkg_check_modules(LIB60870 IMPORTED_TARGET lib60870>=2.3.1)
 pkg_check_modules(LIBCONFIG IMPORTED_TARGET libconfig>=1.4.9)
 pkg_check_modules(MOSQUITTO IMPORTED_TARGET libmosquitto>=1.6.9)
 pkg_check_modules(MODBUS IMPORTED_TARGET libmodbus>=3.1.0)
-pkg_check_modules(RDKAFKA IMPORTED_TARGET rdkafka>=1.5.0)
+pkg_check_modules(RDKAFKAPP IMPORTED_TARGET rdkafka++>=1.5.0)
 pkg_check_modules(HIREDIS IMPORTED_TARGET hiredis>=1.0.0)
 pkg_check_modules(REDISPP IMPORTED_TARGET redis++>=1.2.0)
 pkg_check_modules(RABBITMQ_C IMPORTED_TARGET librabbitmq>=0.8.0)
@@ -193,7 +193,7 @@ cmake_dependent_option(WITH_NODE_IEC60870       "Build with iec60870 node-types"
 cmake_dependent_option(WITH_NODE_IEC61850       "Build with iec61850 node-types"                        "${WITH_DEFAULTS}" "LIBIEC61850_FOUND; NOT WITHOUT_GPL" OFF)
 cmake_dependent_option(WITH_NODE_INFINIBAND     "Build with infiniband node-type"                       "${WITH_DEFAULTS}" "IBVerbs_FOUND; RDMACM_FOUND" OFF) # Infiniband node-type is currenly broken
 cmake_dependent_option(WITH_NODE_INFLUXDB       "Build with influxdb node-type"                         "${WITH_DEFAULTS}" "" OFF)
-cmake_dependent_option(WITH_NODE_KAFKA          "Build with kafka node-type"                            "${WITH_DEFAULTS}" "RDKAFKA_FOUND" OFF)
+cmake_dependent_option(WITH_NODE_KAFKA          "Build with kafka node-type"                            "${WITH_DEFAULTS}" "RDKAFKAPP_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_LOOPBACK       "Build with loopback node-type"                         "${WITH_DEFAULTS}" "" OFF)
 cmake_dependent_option(WITH_NODE_MODBUS         "Build with modbus node-type"                           "${WITH_DEFAULTS}" "MODBUS_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_MQTT           "Build with mqtt node-type"                             "${WITH_DEFAULTS}" "MOSQUITTO_FOUND" OFF)
@@ -309,7 +309,7 @@ add_feature_info(NODE_MODBUS            WITH_NODE_MODBUS            "Build with
 add_feature_info(NODE_MQTT              WITH_NODE_MQTT              "Build with mqtt node-type")
 add_feature_info(NODE_NANOMSG           WITH_NODE_NANOMSG           "Build with nanomsg node-type")
 add_feature_info(NODE_NGSI              WITH_NODE_NGSI              "Build with ngsi node-type")
-add_feature_info(NODE_OPAL_AYSNC        WITH_NODE_OPAL_ASYNC        "Build with opal.async node-type")
+add_feature_info(NODE_OPAL_ASYNC        WITH_NODE_OPAL_ASYNC        "Build with opal.async node-type")
 add_feature_info(NODE_OPAL_ORCHESTRA    WITH_NODE_OPAL_ORCHESTRA    "Build with opal.orchestra node-type")
 add_feature_info(NODE_OPENDSS           WITH_NODE_OPENDSS           "Build with opendss node-type")
 add_feature_info(NODE_REDIS             WITH_NODE_REDIS             "Build with redis node-type")

doc/openapi/components/schemas/config/nodes/kafka.yaml

@@ -56,7 +56,7 @@ allOf:
     in:
       type: object
       properties:
-        consume:
+        topic:
           type: string
           description: The Kafka topic to which this node-type will subscribe for receiving messages.
 
@@ -67,7 +67,7 @@ allOf:
     out:
       type: object
       properties:
-        produce:
+        topic:
           type: string
           description: The Kafka topic to which this node-type will publish messages.
 

doc/openapi/components/schemas/config/nodes/opal_orchestra.yaml

@@ -59,12 +59,6 @@ allOf:
       description: >-
         If true, the DDF file provided in the 'dff' setting will be overwriting with settings and signals from the VILLASnode configuration.
 
-    ddf_overwrite_only:
-      type: boolean
-      default: false
-      description: >-
-        If true, VILLASnode will overwrite the file provided in the 'ddf' setting, and terminate immediately afterwards.
-
     rate:
       type: number
       default: 1

doc/openapi/components/schemas/formats/edgeflex.yaml

@@ -34,9 +34,9 @@ additionalProperties:
 
 example:
 - created: 1633791645123
-  signal0: 123.456
-  signal1: true
-  signal2: 1234
-  signal3:
+  signal_0: 123.456
+  signal_1: true
+  signal_2: 1234
+  signal_3:
     real: 1234.4556
     imag: 23232.12312

doc/openapi/paths/nodes.yaml

@@ -41,28 +41,28 @@ get:
                     signals:
                       - type: float
                         enabled: true
-                        name: signal0
+                        name: signal_0
                       - type: float
                         enabled: true
-                        name: signal1
+                        name: signal_1
                       - type: float
                         enabled: true
-                        name: signal2
+                        name: signal_2
                       - type: float
                         enabled: true
-                        name: signal3
+                        name: signal_3
                   out:
                     vectorize: 2
                     signals:
                       - type: float
                         enabled: true
-                        name: signal0
+                        name: signal_0
                       - type: float
                         enabled: true
-                        name: signal1
+                        name: signal_1
                       - type: float
                         enabled: true
-                        name: signal2
+                        name: signal_2
                   type: websocket
                   vectorize: 2
                   series:

etc/examples/nodes/kafka.conf

@@ -7,27 +7,39 @@ nodes = {
 
         format = "json.kafka"
 
-        server = "localhost:9094"
-        protocol = "SASL_SSL"
+        server = "localhost:9092"
+        protocol = "PLAINTEXT"
         client_id = "villas-node"
 
         in = {
-            consume = "test-topic"
+            topic = "test-topic"
             group_id = "villas-node"
         }
 
         out = {
-            produce = "test-topic"
+            topic = "test-topic"
         }
+    }
 
-        ssl = {
-            ca = "/etc/ssl/certs/ca.pem"
-        }
+    siggen = {
+        type = "signal"
 
-        sasl = {
-            mechanisms = "SCRAM-SHA-512"
-            username = "scram-sha-512-usr"
-            password = "scram-sha-512-pwd"
-        }
+        rate = 20
+        values = 5
+        signal = "mixed"
     }
 }
+
+paths = (
+    {
+        in = "siggen"
+        out = "kafka_node"
+    },
+    {
+        in = "kafka_node"
+
+        hooks = [
+            "print"
+        ]
+    }
+)

etc/examples/nodes/opal_orchestra.conf

@@ -74,7 +74,6 @@ nodes = {
 
                     orchestra_name = "pub_signal_float"
                     orchestra_type = "float64"
-                    orchestra_index = 0
                 },
                 {
                     name = "pub_signal2"
@@ -83,7 +82,7 @@ nodes = {
                     orchestra_name = "pub_signal_float"
                     orchestra_type = "float64"
                 },
-                { name = "signal_float", orchestra_name = "some_bus/signal_float", orchestra_type = "float64", orchestra_index = 2 },
+                { name = "signal_float", orchestra_name = "some_bus/signal_float", orchestra_type = "float64" },
                 { name = "signal_bool", orchestra_name = "some_bus/signal_bool", orchestra_type = "boolean" },
                 { name = "signal_uint8", orchestra_name = "some_bus/some_nested_bus/signal_uint8", orchestra_type = "unsigned int8" },
                 { name = "signal_uint8_2", orchestra_type = "unsigned int8" }

etc/examples/typhoon.conf

@@ -21,11 +21,11 @@ nodes = {
             )
 
             signals = (
-                { name = "signal0", type="float" },
-                { name = "signal1", type="integer" },
-                { name = "signal2", type="boolean" },
-                { name = "signal3", type="float" },
-                { name = "signal4", type="complex" }
+                { name = "signal_0", type="float" },
+                { name = "signal_1", type="integer" },
+                { name = "signal_2", type="boolean" },
+                { name = "signal_3", type="float" },
+                { name = "signal_4", type="complex" }
             )
         }
 

include/villas/node.hpp

@@ -7,6 +7,7 @@
 
 #pragma once
 
+#include <functional>
 #include <iostream>
 
 #include <fmt/ostream.h>
@@ -106,6 +107,13 @@ class Node {
 
   virtual json_t *_readStatus() const { return nullptr; }
 
+  int parseCommon(
+      json_t *json,
+      std::function<Signal::Ptr(json_t *json, NodeDirection::Direction dir, unsigned index)>
+          parse_signal = [](json_t *json, NodeDirection::Direction dir, unsigned index) {
+            return Signal::fromJson(json);
+          });
+
 public:
   // Initialize node with default values
   Node(const uuid_t &id = {}, const std::string &name = "");
@@ -277,7 +285,6 @@ class Node {
 };
 
 class NodeFactory : public villas::plugin::Plugin {
-
   friend Node;
 
 protected:

include/villas/node/config.hpp.in

@@ -16,7 +16,7 @@
 #define DEFAULT_SAMPLE_LENGTH 64u
 #define DEFAULT_QUEUE_LENGTH 1024u
 #define MAX_SAMPLE_LENGTH 512u
-#define DEFAULT_FORMAT_BUFFER_LENGTH 4096u
+#define DEFAULT_FORMAT_BUFFER_LENGTH (4096u*1024u)
 
 /* Number of hugepages which are requested from the the kernel.
  * @see https://www.kernel.org/doc/Documentation/vm/hugetlbpage.txt

include/villas/node_direction.hpp

@@ -7,6 +7,8 @@
 
 #pragma once
 
+#include <functional>
+
 #include <jansson.h>
 
 #include <villas/common.hpp>
@@ -50,7 +52,7 @@ class NodeDirection {
 
   NodeDirection(enum NodeDirection::Direction dir, Node *n);
 
-  int parse(json_t *json);
+  int parse(json_t *json, std::function<Signal::Ptr(json_t *json, unsigned index)> parse_signal);
   void check();
   int prepare();
   int start();

include/villas/nodes/nanomsg.hpp

@@ -16,9 +16,6 @@ namespace node {
 // Forward declarations
 class NodeCompat;
 
-// The maximum length of a packet which contains stuct msg.
-#define NANOMSG_MAX_PACKET_LEN 1500
-
 struct nanomsg {
   struct {
     int socket;

include/villas/nodes/socket.hpp

@@ -17,8 +17,7 @@ namespace node {
 // Forward declarations
 class NodeCompat;
 
-// The maximum length of a packet which contains stuct msg.
-#define SOCKET_INITIAL_BUFFER_LEN (64 * 1024)
+#define SOCKET_INITIAL_BUFFER_LEN DEFAULT_FORMAT_BUFFER_LENGTH
 
 struct Socket {
   int sd;     // The socket descriptor

include/villas/signal.hpp

@@ -8,6 +8,7 @@
 #pragma once
 
 #include <memory>
+#include <optional>
 
 #include <jansson.h>
 
@@ -35,19 +36,21 @@ class Signal {
   union SignalData init; // The initial value of the signal.
   enum SignalType type;
 
+  Signal::Ptr previous; // Previous signal in vector.
+
   // Initialize a signal with default values.
   Signal(const std::string &n = "", const std::string &u = "",
          enum SignalType t = SignalType::INVALID);
 
   // Parse signal description.
   int parse(json_t *json);
 
-  std::string toString(const union SignalData *d = nullptr) const;
+  virtual std::string toString(const union SignalData *d = nullptr) const;
 
   // Produce JSON representation of signal.
-  json_t *toJson() const;
+  virtual json_t *toJson() const;
 
-  bool isNext(const Signal &sig);
+  static Signal::Ptr fromJson(json_t *json);
 };
 
 } // namespace node

include/villas/signal_list.hpp

@@ -7,7 +7,10 @@
 
 #pragma once
 
+#include <functional>
 #include <memory>
+#include <string>
+#include <string_view>
 
 #include <jansson.h>
 
@@ -24,16 +27,12 @@ class SignalList : public std::vector<Signal::Ptr> {
   using Ptr = std::shared_ptr<SignalList>;
 
   SignalList() {}
-
+  SignalList(json_t *json, std::function<Signal::Ptr(json_t *json, unsigned index)> parse_signal =
+                               [](json_t *json, unsigned index) {
+                                return Signal::fromJson(json);
+                              });
+  SignalList(std::string_view dt);
   SignalList(unsigned len, enum SignalType fmt);
-  SignalList(const char *dt);
-  SignalList(json_t *json) {
-    int ret = parse(json);
-    if (ret)
-      throw RuntimeError("Failed to parse signal list");
-  }
-
-  int parse(json_t *json);
 
   Ptr clone();
 
@@ -42,9 +41,15 @@ class SignalList : public std::vector<Signal::Ptr> {
 
   json_t *toJson() const;
 
-  int getIndexByName(const std::string &name);
-  Signal::Ptr getByName(const std::string &name);
+  int getIndexByName(std::string_view name);
+  Signal::Ptr getByName(std::string_view name);
   Signal::Ptr getByIndex(unsigned idx);
+
+  void
+  parse(json_t *json_signals,
+        std::function<Signal::Ptr(json_t *json, unsigned index)> parse_signal = [](json_t *json, unsigned index) {
+          return Signal::fromJson(json);
+         });
 };
 
 } // namespace node