Core: support for Data Record extensions [WIP]

Author: Lukas955

src/core/CMakeLists.txt

@@ -4,6 +4,8 @@ set(CORE_SOURCE
     configurator/config_file.hpp
     configurator/configurator.cpp
     configurator/configurator.hpp
+    configurator/extensions.cpp
+    configurator/extensions.hpp
     configurator/instance.hpp
     configurator/instance_input.cpp
     configurator/instance_input.hpp

src/core/configurator/extensions.cpp

@@ -0,0 +1,138 @@
+
+
+#include <algorithm>
+#include "extensions.hpp"
+
+/** Identification of this component (for log) */
+static const char *comp_str = "Extensions";
+
+void
+ipx_cfg_extensions::add_extension(ipx_instance *inst, unsigned int idx, struct ipx_ctx_ext *ext)
+{
+    auto ext_info = &m_extensions[ext->data_type][ext->data_name];
+    std::vector<struct plugin_rec> &vec_plugins = (ext->etype == IPX_EXTENSION_PRODUCER)
+        ? ext_info->producers
+        : ext_info->consumers;
+    vec_plugins.push_back({inst->get_name(), idx, ext});
+}
+
+void
+ipx_cfg_extensions::check_dependencies(const std::string &ident, const struct ext_rec &rec)
+{
+    std::string name_producers;
+    std::string name_consumers;
+
+    // Prepare list of produces and consumers for log
+    for (const auto &producer : rec.producers) {
+        if (!name_producers.empty()) {
+            name_producers.append(", ");
+        }
+        name_producers.append("'" + producer.name + "'");
+    }
+    for (const auto &consumer : rec.consumers) {
+        if (!name_consumers.empty()) {
+            name_consumers.append(", ");
+        }
+        name_consumers.append("'" + consumer.name + "'");
+    }
+
+    // No producers?
+    if (rec.producers.empty()) {
+        throw std::runtime_error("No provider of Data Record extension " + ident + "found. "
+            "The extension is required by " + name_consumers);
+    }
+
+    // Multiple producers?
+    if (rec.producers.size() > 1) {
+        throw std::runtime_error("Data Record extension " + ident + " is provided by "
+            "multiple instances (" + name_producers + ")");
+    }
+
+    // No consumer
+    if (rec.consumers.empty()) {
+        IPX_WARNING(comp_str, "Extension %s is provided by %s, but no other plugins use it. "
+            "The provider can be probably removed.", ident.c_str(), name_producers.c_str());
+        return;
+    }
+
+    // Check if the producer is placed before all consumers
+    const struct plugin_rec &producer_rec = rec.producers.front();
+    const auto min_cons = std::min_element(rec.consumers.cbegin(), rec.consumers.cend());
+    if (producer_rec.plugin_idx > min_cons->plugin_idx) {
+        throw std::runtime_error("Instance '" + producer_rec.name + "', which is a provider "
+            "of Data Record extension " + ident + ", is placed in the collector pipeline "
+            "after '" + min_cons->name + "' instance, which depends on the extension. "
+            "Please, swap the order of the plugin instances");
+    }
+}
+
+
+void
+ipx_cfg_extensions::resolve(std::vector<ipx_instance *> &plugins)
+{
+    // Add extensions and dependencies from all plugins
+    unsigned int pos = 0;
+    for (auto &it : plugins) {
+        if (dynamic_cast<ipx_instance_input *>(it)) {
+            // Input plugins cannot register extensions
+            continue;
+        }
+
+        struct ipx_ctx_ext *arr_ptr = nullptr;
+        size_t arr_size = 0;
+
+        std::tie(arr_ptr, arr_size) = it->get_extensions();
+        for (size_t i = 0; i < arr_size; ++i) {
+            add_extension(it, pos, &arr_ptr[i]);
+        }
+
+        if (!dynamic_cast<ipx_instance_output *>(it)) {
+            // Increment position for all plugins except outputs
+            pos++;
+        }
+    };
+
+    if (m_extensions.empty()) {
+        // No extensions, no dependencies
+        return;
+    }
+
+    size_t offset = 0;
+    uint64_t mask = 1U;
+
+    for (auto &ext_type : m_extensions) {
+        for (auto &ext_name : ext_type.second) {
+            if (!mask) {
+                // No more bits in the mask!
+                throw std::runtime_error("Maximum number of Data Record extensions has been reached!");
+            }
+
+            struct ext_rec &ext = ext_name.second;
+            assert(ext.producers.size() == 1 && "Exactly one producer");
+
+            // Check the extension
+            std::string ident = "'" + ext_type.first + "/" + ext_name.first + "'";
+            check_dependencies(ident, ext_name.second);
+
+            // Determine size, offset and bitset mask
+            ext.size = ext.producers[0].rec->size;
+            ext.offset = offset;
+            ext.mask = mask;
+
+            // Align the offset to multiple of 8
+            ext.offset += (ext.size % 8U == 0) ? ext.size : (((ext.size / 8U) + 1U) * 8U);
+            ext.mask <<= 1U;
+
+            IPX_DEBUG(comp_str, "Extension %s registered (size: %zu, offset: %zu)",
+                ext.size, ext.offset);
+        }
+    }
+
+
+
+    // TODO: save total size of all extensions
+
+
+
+}
+

src/core/configurator/extensions.hpp

@@ -0,0 +1,56 @@
+
+#ifndef IPX_CFG_EXTENSIONS_H
+#define IPX_CFG_EXTENSIONS_H
+
+#include <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "instance.hpp"
+#include "instance_intermediate.hpp"
+#include "instance_output.hpp"
+
+extern "C" {
+#include "../extension.h"
+#include "../verbose.h"
+}
+
+class ipx_cfg_extensions {
+private:
+    struct plugin_rec {
+        std::string name;
+        unsigned int plugin_idx;
+        const struct ipx_ctx_ext *rec;
+
+        bool operator<(const plugin_rec &other) const {
+            return plugin_idx < other.plugin_idx;
+        }
+    };
+
+    struct ext_rec {
+        std::vector<plugin_rec> producers;
+        std::vector<plugin_rec> consumers;
+
+        size_t size;
+        size_t offset;
+        uint64_t mask;
+    };
+
+    // Extensions [extension type][extension name]
+    std::map<std::string, std::map<std::string, struct ext_rec>> m_extensions;
+
+    void
+    add_extension(ipx_instance *inst, unsigned int idx, struct ipx_ctx_ext *ext);
+    void
+    check_dependencies(const std::string &ident, const struct ext_rec &rec);
+
+public:
+    ipx_cfg_extensions() = default;
+    ~ipx_cfg_extensions() = default;
+
+    void
+    resolve(std::vector<ipx_instance *> &plugins);
+};
+
+#endif // IPX_CFG_EXTENSIONS_H

src/core/configurator/instance.hpp

@@ -116,6 +116,25 @@ class ipx_instance {
 
     /** \brief Start a thread of the instance                                                    */
     virtual void start() = 0;
+
+    /**
+     * \brief Get name of the instance
+     * \return Name
+     */
+    const std::string &
+    get_name() {
+        return _name;
+    }
+
+    /**
+     * \brief Get registered extensions and dependencies
+     * \return An array of extensions and dependencies and the size of the array
+     * \throw runtime_error if extension are not supported by the instance type
+     */
+    virtual std::tuple<struct ipx_ctx_ext *, size_t>
+    get_extensions() {
+        throw std::runtime_error("Extensions are not supported the plugin");
+    }
 };
 
 #endif //IPFIXCOL_INSTANCE_H

src/core/configurator/instance_intermediate.cpp

@@ -134,4 +134,13 @@ ipx_instance_intermediate::connect_to(ipx_instance_intermediate &intermediate)
 {
     assert(_state == state::NEW); // Only configuration of an uninitialized instance can be changed!
     ipx_ctx_ring_dst_set(_ctx, intermediate.get_input());
+}
+
+std::tuple<struct ipx_ctx_ext *, size_t>
+ipx_instance_intermediate::get_extensions()
+{
+    struct ipx_ctx_ext *ext_arr = nullptr;
+    size_t ext_size = 0;
+    ipx_ctx_ext_defs(_ctx, &ext_arr, &ext_size);
+    return std::make_tuple(ext_arr, ext_size);
 }

src/core/configurator/instance_intermediate.hpp

@@ -152,6 +152,14 @@ class ipx_instance_intermediate : public ipx_instance {
      * \param[in] intermediate Intermediate plugin to receive our messages
      */
     virtual void connect_to(ipx_instance_intermediate &intermediate);
+
+    /**
+     * \brief Get registered extensions and dependencies
+     * \return An array of extensions and dependencies and the size of the array
+     * \throw runtime_error if extension are not supported by the instance type
+     */
+    virtual std::tuple<struct ipx_ctx_ext *, size_t>
+    get_extensions() override;
 };
 
 #endif //IPFIXCOL_INSTANCE_INTERMEDIATE_HPP

src/core/configurator/instance_output.cpp

@@ -141,3 +141,12 @@ ipx_instance_output::get_input()
 {
     return std::make_tuple(_instance_buffer, _type, _filter);
 }
+
+std::tuple<struct ipx_ctx_ext *, size_t>
+ipx_instance_output::get_extensions()
+{
+    struct ipx_ctx_ext *ext_arr = nullptr;
+    size_t ext_size = 0;
+    ipx_ctx_ext_defs(_ctx, &ext_arr, &ext_size);
+    return std::make_tuple(ext_arr, ext_size);
+}

src/core/configurator/instance_output.hpp

@@ -137,6 +137,16 @@ class ipx_instance_output : public ipx_instance {
      */
     std::tuple<ipx_ring_t *, enum ipx_odid_filter_type, const ipx_orange_t *>
     get_input();
+
+    /**
+     * \brief Get registered extension dependencies
+     *
+     * \note Since output instances cannot produce extensions, the array contains only registered
+     *   dependencies.
+     * \return An array of dependencies and the size of the array
+     */
+    std::tuple<struct ipx_ctx_ext *, size_t>
+    get_extensions() override;
 };
 
 #endif //IPFIXCOL_INSTANCE_OUTPUT_HPP

src/core/context.c

@@ -375,6 +375,18 @@ ipx_ctx_ring_dst_set(ipx_ctx_t *ctx, ipx_ring_t *ring)
     ctx->pipeline.dst = ring;
 }
 
+void
+ipx_ctx_ext_defs(ipx_ctx_t *ctx, struct ipx_ctx_ext **arr, size_t *arr_size)
+{
+    if (ctx->cfg_extension.items_cnt == 0) {
+        *arr = NULL;
+        *arr_size = 0;
+        return;
+    }
+
+    *arr = ctx->cfg_extension.items;
+    *arr_size = ctx->cfg_extension.items_cnt;
+}
 
 // -------------------------------------------------------------------------------------------------
 
@@ -448,6 +460,7 @@ ipx_ctx_ext_producer(ipx_ctx_t *ctx, const char *type, const char *name, size_t
         return rc;
     }
 
+    IPX_CTX_DEBUG(ctx, "Data Record extension '%s/%s' has been registered.", type, name);
     *ext = rec;
     return IPX_OK;
 }
@@ -474,6 +487,7 @@ ipx_ctx_ext_consumer(ipx_ctx_t *ctx, const char *type, const char *name, ipx_ctx
         return rc;
     }
 
+    IPX_CTX_DEBUG(ctx, "Dependency on Data Record extension '%s/%s' has been added.", type, name);
     *ext = rec;
     return IPX_OK;
 }

src/core/context.h

@@ -264,4 +264,19 @@ ipx_ctx_verb_set(ipx_ctx_t *ctx, enum ipx_verb_level verb);
 IPX_API int
 ipx_ctx_term_cnt_set(ipx_ctx_t *ctx, unsigned int cnt);
 
+/**
+ * \brief Get registered extensions and dependencies
+ *
+ * \note
+ *   Keep on mind that the array is filled only after plugin initialization. Moreover,
+ *   the most plugins don't use extension at all, so the array is usually empty.
+ * \warning
+ *   Don't change the size and offset of extensions if the plugin is already running!
+ * \param[in] ctx       Plugin context
+ * \param[out] arr      Array with extensions and dependencies
+ * \param[out] arr_size Size of the array
+ */
+IPX_API void
+ipx_ctx_ext_defs(ipx_ctx_t *ctx, struct ipx_ctx_ext **arr, size_t *arr_size);
+
 #endif // IPFIXCOL_CONTEXT_INTERNAL_H