Add new OPAL-RT Asynchronous Process node-type

CMakeLists.txt

@@ -79,6 +79,7 @@ find_package(FileSystem)
 find_package(Criterion)
 find_package(OpalOrchestra)
 find_package(LibXml2)
+find_package(OpalAsyncApi)
 
 # Check for tools
 find_program(PROTOBUFC_COMPILER NAMES protoc-c)
@@ -198,7 +199,8 @@ cmake_dependent_option(WITH_NODE_MODBUS         "Build with modbus node-type"
 cmake_dependent_option(WITH_NODE_MQTT           "Build with mqtt node-type"                             "${WITH_DEFAULTS}" "MOSQUITTO_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_NANOMSG        "Build with nanomsg node-type"                          "${WITH_DEFAULTS}" "NANOMSG_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_NGSI           "Build with ngsi node-type"                             "${WITH_DEFAULTS}" "" OFF)
-cmake_dependent_option(WITH_NODE_OPAL_ORCHESTRA "Build with the opal-orchestra node-type"               "${WITH_DEFAULTS}" "OpalOrchestra_FOUND; LibXml2_FOUND" OFF)
+cmake_dependent_option(WITH_NODE_OPAL_ORCHESTRA "Build with opal-orchestra node-type"                   "${WITH_DEFAULTS}" "OpalOrchestra_FOUND; LibXml2_FOUND" OFF)
+cmake_dependent_option(WITH_NODE_OPAL_ASYNC     "Build with opal.async node-type"                       "${WITH_DEFAULTS}" "OpalAsyncApi_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_REDIS          "Build with redis node-type"                            "${WITH_DEFAULTS}" "HIREDIS_FOUND; REDISPP_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_RTP            "Build with rtp node-type"                              "${WITH_DEFAULTS}" "re_FOUND" OFF)
 cmake_dependent_option(WITH_NODE_SHMEM          "Build with shmem node-type"                            "${WITH_DEFAULTS}" "HAS_SEMAPHORE; HAS_MMAN" OFF)

cmake/FindOpalAsyncApi.cmake

@@ -0,0 +1,34 @@
+# CMakeLists.txt
+#
+# Author: Steffen Vogel <[email protected]>
+# SPDX-FileCopyrightText: 2023-2025 OPAL-RT Germany GmbH
+# SPDX-License-Identifier: Apache-2.0
+
+set(TARGET_RTLAB_ROOT "/usr/opalrt" CACHE STRING "RT-LAB Root directory")
+
+if(EXISTS "${TARGET_RTLAB_ROOT}/common/bin/opalmodelmk")
+    set(ENV{TARGET_RTLAB_ROOT} ${TARGET_RTLAB_ROOT})
+
+    file(WRITE "${CMAKE_CURRENT_BINARY_DIR}/get_vars.mk"
+        "all:\n\techo $(OPAL_LIBS) $(OPAL_LIBPATH)\n")
+
+    execute_process(
+        COMMAND make -sf ${TARGET_RTLAB_ROOT}/common/bin/opalmodelmk
+        WORKING_DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}"
+        OUTPUT_VARIABLE OPAL_VARS
+    )
+
+    string(STRIP ${OPAL_VARS} OPAL_VARS)
+    string(REPLACE " " ";" OPAL_VARS ${OPAL_VARS})
+
+    set(OPAL_LIBRARIES -lOpalCore -lOpalUtils ${OPAL_VARS} -lirc -ldl -pthread -lrt)
+    set(OPAL_INCLUDE_DIR ${TARGET_RTLAB_ROOT}/common/include_target)
+
+    add_library(OpalAsyncApi INTERFACE)
+    target_include_directories(OpalAsyncApi INTERFACE ${OPAL_INCLUDE_DIR})
+    target_link_libraries(OpalAsyncApi INTERFACE ${OPAL_LIBRARIES})
+endif()
+
+find_package_handle_standard_args(OpalAsyncApi DEFAULT_MSG OPAL_LIBRARIES OPAL_INCLUDE_DIR)
+
+mark_as_advanced(OPAL_LIBRARIES OPAL_INCLUDE_DIR)

doc/openapi/components/schemas/config/node_obj.yaml

@@ -39,6 +39,7 @@ discriminator:
     mqtt: nodes/_mqtt.yaml
     nanomsg: nodes/_nanomsg.yaml
     ngsi: nodes/_ngsi.yaml
+    opal_async: nodes/_opal_async.yaml
     opendss: nodes/_opendss.yaml
     opal.orchestra: nodes/_opal_orchestra.yaml
     redis: nodes/_redis.yaml

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

@@ -0,0 +1,7 @@
+# yaml-language-server: $schema=http://json-schema.org/draft-07/schema
+# SPDX-FileCopyrightText: 2025 Institute for Automation of Complex Power Systems, RWTH Aachen University
+# SPDX-License-Identifier: Apache-2.0
+---
+allOf:
+- $ref: ../node_obj.yaml
+- $ref: opal_async.yaml

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

@@ -0,0 +1,42 @@
+# yaml-language-server: $schema=http://json-schema.org/draft-07/schema
+# Author: Steffen Vogel <[email protected]>
+# SPDX-FileCopyrightText: 2023-2025 OPAL-RT Germany GmbH
+# SPDX-License-Identifier: Apache-2.0
+---
+allOf:
+  - type: object
+    properties:
+      id:
+        description: The Send/Recv ID of the RT-Lab OpAsyncSend/Recv blocks.
+        min: 1
+        default: 1
+        type: integer
+
+      in:
+        type: object
+        properties:
+          reply:
+            description: Send a confirmation to the Simulink model that signals have been received and processed.
+            default: false
+            type: boolean
+
+      shmem:
+        description: Shared-memory parameters for communication with OpAsyncGenCtrl block of Simulink model.
+        type: object
+        required:
+          - async_name
+          - async_size
+          - system_ctrl_name
+        properties:
+          async_name:
+            description: Name of the shared memory region used for data exchange with the Simulink model.
+            type: string
+          async_size:
+            description: Size of the shared memory region used for data exchange with the Simulink model.
+            type: integer
+          system_ctrl_name:
+            description: Name of the shared memory region used for logging.
+            type: string
+
+  - $ref: ../node_signals.yaml
+  - $ref: ../node.yaml

etc/examples/nodes/opal_async.conf

@@ -0,0 +1,21 @@
+# Author: Steffen Vogel <[email protected]>
+# SPDX-FileCopyrightText: 2023-2025 OPAL-RT Germany GmbH
+# SPDX-License-Identifier: Apache-2.0
+
+nodes = {
+    opal_async_node1 = {
+        type = "opal.async"
+
+        # The Send/Recv ID of the RT-Lab OpAsyncSend/Recv blocks.
+        id = 1
+
+        in = {
+            # Send a confirmation to the Simulink model that signals have been received and processed.
+            reply = false
+
+            hooks = (
+                "stats"
+            )
+        }
+    }
+}

include/villas/mapping.hpp

@@ -42,9 +42,7 @@ class MappingEntry {
   using Ptr = std::shared_ptr<MappingEntry>;
 
   enum class Type { UNKNOWN, DATA, STATS, HEADER, TIMESTAMP };
-
   enum class HeaderType { LENGTH, SEQUENCE };
-
   enum class TimestampType { ORIGIN, RECEIVED };
 
   Node *node;     // The node to which this mapping refers.

include/villas/nodes/opal_async.hpp

@@ -0,0 +1,146 @@
+/* Node type: OPAL-RT Asynchronous Process (libOpalAsyncApi)
+ *
+ * Author: Steffen Vogel <[email protected]>
+ * SPDX-FileCopyrightText: 2023-2025 OPAL-RT Germany GmbH
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <condition_variable>
+#include <mutex>
+
+#include <spdlog/details/log_msg_buffer.h>
+#include <spdlog/details/null_mutex.h>
+#include <spdlog/sinks/base_sink.h>
+
+// Define RTLAB before including OpalPrint.h for messages to be sent
+// to the OpalDisplay. Otherwise stdout will be used.
+#define RTLAB
+
+#include <AsyncApi.h>
+#include <OpalGenAsyncParamCtrl.h>
+#include <OpalPrint.h>
+
+#include <villas/format.hpp>
+#include <villas/node.hpp>
+#include <villas/popen.hpp>
+#include <villas/sample.hpp>
+
+namespace villas {
+namespace node {
+namespace opal {
+
+class LogSink final : public spdlog::sinks::base_sink<std::mutex> {
+private:
+  std::string shmemSystemCtrlName;
+
+public:
+  explicit LogSink(const std::string &shmemName);
+
+  ~LogSink() override;
+
+protected:
+  void sink_it_(const spdlog::details::log_msg &msg) override;
+
+  void flush_() override {}
+};
+
+} // namespace opal
+
+// Forward declarations
+struct Sample;
+
+class OpalAsyncNode : public Node {
+
+protected:
+  // RT-LAB -> VILLASnode
+  // Corresponds to AsyncAPI's *Send* direction
+  struct {
+    unsigned id;
+    bool present;
+    unsigned length;
+
+    bool reply;
+    int mode;
+
+    Opal_SendAsyncParam params;
+  } in;
+
+  // VILLASnode -> RT-LAB
+  // Corresponds to AsyncAPI's *Recv* direction
+  struct {
+    unsigned id;
+    bool present;
+    unsigned length;
+
+    Opal_RecvAsyncParam params;
+  } out;
+
+  bool ready;
+  std::mutex readyLock;
+  std::condition_variable readyCv;
+
+  virtual int _read(struct Sample *smps[], unsigned cnt) override;
+
+  virtual int _write(struct Sample *smps[], unsigned cnt) override;
+
+public:
+  OpalAsyncNode(const uuid_t &id = {}, const std::string &name = "")
+      : Node(id, name), ready(false) {
+    in.id = 1;
+    in.present = false;
+    in.length = 0;
+    in.reply = true;
+
+    out.id = 1;
+    out.present = false;
+    out.length = 0;
+  }
+
+  virtual const std::string &getDetails() override;
+
+  virtual int start() override;
+  virtual int stop() override;
+
+  virtual int parse(json_t *json) override;
+
+  void markReady();
+  void waitReady();
+};
+
+class OpalAsyncNodeFactory : public NodeFactory {
+
+public:
+  using NodeFactory::NodeFactory;
+
+  virtual Node *make(const uuid_t &id = {},
+                     const std::string &nme = "") override {
+    auto *n = new OpalAsyncNode(id, nme);
+
+    init(n);
+
+    return n;
+  }
+
+  virtual int getFlags() const override {
+    return (int)NodeFactory::Flags::SUPPORTS_READ |
+           (int)NodeFactory::Flags::SUPPORTS_WRITE |
+           (int)NodeFactory::Flags::PROVIDES_SIGNALS;
+  }
+
+  virtual std::string getName() const override { return "opal.async"; }
+
+  virtual std::string getDescription() const override {
+    return "OPAL Asynchronous Process (libOpalAsyncApi)";
+  }
+
+  virtual int getVectorize() const override { return 1; }
+
+  virtual int start(SuperNode *sn) override;
+
+  virtual int stop() override;
+};
+
+} // namespace node
+} // namespace villas

include/villas/nodes/opal_orchestra/ddf.hpp

@@ -38,7 +38,9 @@ class DataItem : public Item {
   unsigned short length;
   double defaultValue;
 
-  explicit DataItem(std::string name) : name(std::move(name)) {}
+  explicit DataItem(std::string name)
+      : name(std::move(name)), type(SignalType::BOOLEAN), length(0),
+        defaultValue(0) {}
 
   static const unsigned int IDENTIFIER_NAME_LENGTH = 64;
 

include/villas/nodes/opal_orchestra/signal.hpp

@@ -42,11 +42,11 @@ orchestra::SignalType signalTypeFromString(const std::string &t);
 
 std::string signalTypeToString(orchestra::SignalType t);
 
-node::SignalData toNodeSignalData(const char *orchestraData,
+node::SignalData toNodeSignalData(const void *orchestraData,
                                   orchestra::SignalType orchestraType,
                                   node::SignalType &villasType);
 
-void toOrchestraSignalData(char *orchestraData,
+void toOrchestraSignalData(void *orchestraData,
                            orchestra::SignalType orchestraType,
                            const SignalData &villasData,
                            node::SignalType villasType);

include/villas/nodes/test_rtt.hpp

@@ -65,12 +65,12 @@ class TestRTT : public Node {
     double getEstimatedDuration() const;
   };
 
-  Task task;             // The periodic task for test_rtt_read()
-  Format::Ptr formatter; // The format of the output file
+  Task task;             // The periodic task for test_rtt_read().
+  Format::Ptr formatter; // The format of the output file.
   FILE *stream;
 
-  std::list<Case> cases;             // List of test cases
-  std::list<Case>::iterator current; // Currently running test case
+  std::list<Case> cases;             // List of test cases.
+  std::list<Case>::iterator current; // Currently running test case.
 
   std::string output; // The directory where we place the results.
   std::string prefix; // An optional prefix in the filename.
@@ -92,7 +92,7 @@ class TestRTT : public Node {
   };
 
   TestRTT(const uuid_t &id = {}, const std::string &name = "")
-      : Node(id, name), task(), formatter(nullptr), stream(nullptr),
+      : Node(id, name), task(), formatter(nullptr), stream(nullptr), current(),
         shutdown(false) {}
 
   virtual ~TestRTT(){};

include/villas/pool.hpp

@@ -20,23 +20,23 @@
 namespace villas {
 namespace node {
 
-// A thread-safe memory pool
+// A thread-safe memory pool.
 struct Pool {
   enum State state;
 
   off_t
-      buffer_off; // Offset from the struct address to the underlying memory area
+      buffer_off; // Offset from the struct address to the underlying memory area.
 
-  size_t len;       // Length of the underlying memory area
-  size_t blocksz;   // Length of a block in bytes
-  size_t alignment; // Alignment of a block in bytes
+  size_t len;       // Length of the underlying memory area.
+  size_t blocksz;   // Length of a block in bytes.
+  size_t alignment; // Alignment of a block in bytes.
 
-  struct CQueue queue; // The queue which is used to keep track of free blocks
+  struct CQueue queue; // The queue which is used to keep track of free blocks.
 };
 
-#define pool_buffer(p) ((char *)(p) + (p)->buffer_off)
+const char *pool_buffer(const struct Pool *pool);
 
-/* Initiazlize a pool
+/* Initialize a pool.
  *
  * @param[inout] p The pool data structure.
  * @param[in] cnt The total number of blocks which are reserverd by this pool.

include/villas/sample.hpp

@@ -13,6 +13,7 @@
 #include <cstdlib>
 #include <ctime>
 #include <memory>
+#include <optional>
 
 #include <villas/log.hpp>
 #include <villas/signal.hpp>
@@ -95,10 +96,7 @@ struct Sample {
 #define SAMPLE_NON_POOL PTRDIFF_MIN
 
 // Get the address of the pool to which the sample belongs.
-#define sample_pool(s)                                                         \
-  ((s)->pool_off == SAMPLE_NON_POOL                                            \
-       ? nullptr                                                               \
-       : (struct Pool *)((char *)(s) + (s)->pool_off))
+std::optional<struct Pool *> sample_pool(const struct Sample *smp);
 
 struct Sample *sample_alloc(struct Pool *p);