From bd65bc7c3a358c7d1f54c92049fa27a90bbbdb77 Mon Sep 17 00:00:00 2001 From: Parker Liu Date: Mon, 2 Aug 2021 11:20:37 +0800 Subject: [PATCH] Implement the plugin manager and adapter manager --- CMakeLists.txt | 10 +- src/adapter/adapter.c | 68 +++--- src/adapter/adapter_internal.h | 9 +- src/core/neu_manager.c | 361 +++++++++++++++++++++++++------ src/core/plugin_info.h | 27 +++ src/core/plugin_manager.c | 375 ++++++++++++++++++++++++++++++++- src/core/plugin_manager.h | 46 +++- src/main.c | 2 +- 8 files changed, 784 insertions(+), 114 deletions(-) create mode 100644 src/core/plugin_info.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 724262e0..4e2c21dd 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -61,8 +61,8 @@ target_link_libraries(neuron-base vector-static ${CMAKE_THREAD_LIBS_INIT}) # define for plugins set(SAMPLE_PLUGIN_SOURCES plugins/plugin_sample.c) -add_library(sample-plugin SHARED) -target_sources(sample-plugin PRIVATE ${SAMPLE_PLUGIN_SOURCES}) -target_include_directories(sample-plugin PRIVATE include) -target_link_libraries(sample-plugin neuron-base) -target_link_libraries(sample-plugin ${CMAKE_THREAD_LIBS_INIT}) +add_library(plugin-sample SHARED) +target_sources(plugin-sample PRIVATE ${SAMPLE_PLUGIN_SOURCES}) +target_include_directories(plugin-sample PRIVATE include) +target_link_libraries(plugin-sample neuron-base) +target_link_libraries(plugin-sample ${CMAKE_THREAD_LIBS_INIT}) diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 9f46d609..aef84922 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -55,6 +55,7 @@ struct neu_adapter { nng_socket sock; nng_thread * thrd; uint32_t new_req_id; + plugin_id_t plugin_id; char * plugin_lib_name; void * plugin_lib; // handle of dynamic lib neu_plugin_module_t *plugin_module; @@ -168,13 +169,15 @@ static void adapter_loop(void *arg) const neu_plugin_intf_funs_t *intf_funs; neu_request_t req; uint32_t req_code; - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = adapter_get_req_id(adapter); - req.req_type = NEU_REQRESP_READ; - req.buf_len = sizeof(uint32_t); - req_code = 1; - req.buf = (char *) &req_code; - intf_funs->request(adapter->plugin, &req); + if (adapter->plugin_module) { + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = adapter_get_req_id(adapter); + req.req_type = NEU_REQRESP_READ; + req.buf_len = sizeof(uint32_t); + req_code = 1; + req.buf = (char *) &req_code; + intf_funs->request(adapter->plugin, &req); + } break; } @@ -276,12 +279,21 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) return NULL; } - adapter->id = info->id; - adapter->type = info->type; - adapter->name = strdup(info->name); - adapter->plugin_lib_name = strdup(info->plugin_lib_name); - adapter->new_req_id = 0; + adapter->id = info->id; + adapter->type = info->type; + adapter->name = strdup(info->name); + adapter->new_req_id = 0; + adapter->plugin_id = info->plugin_id; + if (adapter->plugin_id.id_val == 0) { + adapter->plugin_lib_name = NULL; + adapter->plugin_lib = NULL; + adapter->plugin_module = NULL; + adapter->plugin = NULL; + log_info("Create a adapter without plugin"); + goto open_pipe; + } + adapter->plugin_lib_name = strdup(info->plugin_lib_name); if (adapter->name == NULL || adapter->plugin_lib_name == NULL) { if (adapter->name != NULL) { free(adapter->name); @@ -303,21 +315,21 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) adapter->plugin_lib_name, adapter->name); } + neu_plugin_t *plugin; adapter->plugin_lib = handle; adapter->plugin_module = plugin_module; - neu_plugin_t *plugin; - plugin_module = adapter->plugin_module; - plugin = plugin_module->intf_funs->open(adapter, &callback_funs); + plugin = plugin_module->intf_funs->open(adapter, &callback_funs); if (plugin == NULL) { neu_panic("Can't to open plugin(%s)", plugin_module->module_name); } + adapter->plugin = plugin; +open_pipe: rv = nng_pair1_open(&adapter->sock); if (rv != 0) { neu_panic("The adapter(%s) can't open pipe", adapter->name); } - adapter->plugin = plugin; return adapter; } @@ -328,8 +340,12 @@ void neu_adapter_destroy(neu_adapter_t *adapter) } nng_close(adapter->sock); - adapter->plugin_module->intf_funs->close(adapter->plugin); - unload_plugin_library(adapter->plugin_lib); + if (adapter->plugin_module != NULL) { + adapter->plugin_module->intf_funs->close(adapter->plugin); + } + if (adapter->plugin_lib != NULL) { + unload_plugin_library(adapter->plugin_lib); + } if (adapter->name != NULL) { free(adapter->name); } @@ -350,9 +366,11 @@ int neu_adapter_start(neu_adapter_t *adapter, neu_manager_t *manager) return (-1); } - const neu_plugin_intf_funs_t *intf_funs; - intf_funs = adapter->plugin_module->intf_funs; - intf_funs->init(adapter->plugin); + if (adapter->plugin_module != NULL) { + const neu_plugin_intf_funs_t *intf_funs; + intf_funs = adapter->plugin_module->intf_funs; + intf_funs->init(adapter->plugin); + } adapter->manager = manager; nng_thread_create(&adapter->thrd, adapter_loop, adapter); @@ -369,9 +387,11 @@ int neu_adapter_stop(neu_adapter_t *adapter, neu_manager_t *manager) nng_mtx_unlock(adapter->mtx); nng_thread_destroy(adapter->thrd); - const neu_plugin_intf_funs_t *intf_funs; - intf_funs = adapter->plugin_module->intf_funs; - intf_funs->uninit(adapter->plugin); + if (adapter->plugin_module != NULL) { + const neu_plugin_intf_funs_t *intf_funs; + intf_funs = adapter->plugin_module->intf_funs; + intf_funs->uninit(adapter->plugin); + } return rv; } diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index e164143e..4dc78472 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -22,6 +22,7 @@ #include "core/databuf.h" #include "core/neu_manager.h" +#include "core/plugin_info.h" typedef enum adapter_type { ADAPTER_TYPE_UNKNOW, @@ -33,11 +34,13 @@ typedef enum adapter_type { ADAPTER_TYPE_MAX, } adapter_type_e; +typedef uint32_t adapter_id_t; typedef struct neu_adapter_info { - uint32_t id; + adapter_id_t id; adapter_type_e type; - char * name; - char * plugin_lib_name; + plugin_id_t plugin_id; + const char * name; + const char * plugin_lib_name; } neu_adapter_info_t; neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info); diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index b23a28a1..34a9093c 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -31,25 +31,24 @@ #include "neu_log.h" #include "neu_manager.h" #include "neu_panic.h" +#include "neu_vector.h" +#include "plugin_manager.h" -typedef struct adapter_bind_info { - uint32_t adapter_id; +#define DEFAULT_ADAPTER_REG_COUNT 8 + +typedef struct adapter_reg_entity { + adapter_id_t adapter_id; neu_adapter_t *adapter; nng_pipe adapter_pipe; int bind_count; -} adapter_bind_info_t; - -// TODO: The array of bound adapters should not be fixed -#define MAX_BIND_ADAPTERS 100 -static adapter_bind_info_t g_bind_adapters[MAX_BIND_ADAPTERS]; +} adapter_reg_entity_t; typedef struct manager_bind_info { - nng_mtx * mtx; - nng_cv * cv; - nng_pipe mng_pipe; - nng_socket mng_sock; - int bind_count; - adapter_bind_info_t *bind_adapters; + nng_mtx * mtx; + nng_cv * cv; + nng_pipe mng_pipe; + nng_socket mng_sock; + int bind_count; } manager_bind_info_t; typedef enum manager_state { @@ -66,58 +65,138 @@ struct neu_manager { nng_thread * thrd; uint32_t new_adapter_id; manager_bind_info_t bind_info; + plugin_manager_t * plugin_manager; + vector_t reg_adapters; }; static const char *const manager_url = "inproc://neu_manager"; #define NEU_HAS_SAMPLE_ADAPTER 1 -#ifdef NEU_HAS_SAMPLE_ADAPTER + #if defined(__APPLE__) -#define SAMPLE_PLUGIN_LIB_NAME "libsample-plugin.dylib" + +#ifdef NEU_HAS_SAMPLE_ADAPTER +#define SAMPLE_PLUGIN_LIB_NAME "libplugin-sample.dylib" +#endif +#define WEBSERVER_PLUGIN_LIB_NAME "libplugin-webserver-proxy.dylib" +#define MQTT_PLUGIN_LIB_NAME "libplugin-mqtt.dylib" +#define MODBUS_PLUGIN_LIB_NAME "libplugin-modbus.dylib" + #else -#define SAMPLE_PLUGIN_LIB_NAME "libsample-plugin.so" + +#ifdef NEU_HAS_SAMPLE_ADAPTER +#define SAMPLE_PLUGIN_LIB_NAME "libplugin-sample.so" +#endif +#define WEBSERVER_PLUGIN_LIB_NAME "libplugin-webserver-proxy.so" +#define MQTT_PLUGIN_LIB_NAME "libplugin-mqtt.so" +#define MODBUS_PLUGIN_LIB_NAME "libplugin-modbus.so" + #endif -static neu_adapter_info_t sample_adapter_info = { .type = ADAPTER_TYPE_DRIVER, - .name = "sample-adapter", - .plugin_lib_name = SAMPLE_PLUGIN_LIB_NAME }; - -neu_adapter_t *sample_adapter; +#ifdef NEU_HAS_SAMPLE_ADAPTER +#define SAMPLE_PLUGIN_NAME "sample-plugin" #endif +#define WEBSERVER_PLUGIN_NAME "webserver-plugin-proxy" +#define MQTT_PLUGIN_NAME "mqtt-plugin" +#define MODBUS_PLUGIN_NAME "modbus-plugin" #define ADAPTER_NAME_MAX_LEN 50 #define PLUGIN_LIB_NAME_MAX_LEN 50 -typedef struct adapter_add_cmd { - adapter_type_e type; - char name[ADAPTER_NAME_MAX_LEN + 1]; - char plugin_lib_name[PLUGIN_LIB_NAME_MAX_LEN + 1]; -} adapter_add_cmd_t; +typedef struct adapter_reg_param { + adapter_type_e adapter_type; + const char * adapter_name; + const char * plugin_name; + // optional value, it is nothing when set to 0 + plugin_id_t plugin_id; +} adapter_reg_param_t; -static const adapter_add_cmd_t builtin_adapter_add_cmds[] = { +static const adapter_reg_param_t default_adapter_reg_params[] = { #ifdef NEU_HAS_SAMPLE_ADAPTER - { .type = ADAPTER_TYPE_DRIVER, - .name = "sample-adapter", - .plugin_lib_name = SAMPLE_PLUGIN_LIB_NAME }, + { + .adapter_type = ADAPTER_TYPE_DRIVER, + .adapter_name = "sample-adapter", + .plugin_name = SAMPLE_PLUGIN_NAME, + .plugin_id = { 0 } // The plugin_id is nothing + }, #endif /* - { - .type = ADAPTER_TYPE_WEBSERVER, - .name = "webserver-adapter", - .plugin_lib_name = WEBSERVER_PLUGIN_LIB_NAME - }, - { - .type = ADAPTER_TYPE_MQTT, - .name = "mqtt-adapter", - .plugin_lib_name = MQTT_PLUGIN_LIB_NAME - }, - { - .type = ADAPTER_TYPE_DRIVER, - .name = "modbus-adapter", - .plugin_lib_name = MODBUS_PLUGIN_LIB_NAME - }, + { + .adapter_type = ADAPTER_TYPE_WEBSERVER, + .adapter_name = "builtin-webserver", + .plugin_name = NULL, // The builtin webserver has'nt plugin + .plugin_id = { 0 } // The plugin_id is nothing + }, + { // for remote customize webserver by user, the plugin is a proxy + .adapter_type = ADAPTER_TYPE_WEBSERVER, + .adapter_name = "webserver-adapter", + .plugin_name = WEBSERVER_PLUGIN_NAME, + .plugin_id = { 0 } // The plugin_id is nothing + }, + { + .adapter_type = ADAPTER_TYPE_MQTT, + .adapter_name = "mqtt-adapter", + .plugin_name = MQTT_PLUGIN_NAME, + .plugin_id = { 0 } // The plugin_id is nothing + }, + { + .adapter_type = ADAPTER_TYPE_DRIVER, + .adapter_name = "modbus-adapter", + .plugin_name = MODBUS_PLUGIN_NAME, + .plugin_id = { 0 } // The plugin_id is nothing + }, */ }; +#define DEFAULT_ADAPTER_ADD_INFO_SIZE \ + (sizeof(default_adapter_reg_params) / sizeof(default_adapter_reg_params[0])) + +static const char *default_plugin_lib_names[] = { +#ifdef NEU_HAS_SAMPLE_ADAPTER + SAMPLE_PLUGIN_LIB_NAME, +#endif + /* + WEBSERVER_PLUGIN_LIB_NAME, + MQTT_PLUGIN_LIB_NAME, + MODBUS_PLUGIN_LIB_NAME, + */ +}; +#define DEFAULT_PLUGIN_COUNT \ + (sizeof(default_plugin_lib_names) / sizeof(default_plugin_lib_names[0])) + +// clang-format off +static const plugin_reg_param_t system_plugin_infos[] = { +#ifdef NEU_HAS_SAMPLE_ADAPTER + { + .plugin_kind = PLUGIN_KIND_SYSTEM, + .adapter_type = ADAPTER_TYPE_DRIVER, + .plugin_name = SAMPLE_PLUGIN_NAME, + .plugin_lib_name = SAMPLE_PLUGIN_LIB_NAME + }, +#endif + /* + { + .plugin_kind = PLUGIN_KIND_SYSTEM, + .adapter_type = ADAPTER_TYPE_WEBSERVER, + .plugin_name = WEBSERVER_PLUGIN_NAME, + .plugin_lib_name = WEBSERVER_PLUGIN_LIB_NAME + }, + { + .plugin_kind = PLUGIN_KIND_SYSTEM, + .adapter_type = ADAPTER_TYPE_MQTT, + .plugin_name = MQTT_PLUGIN_NAME, + .plugin_lib_name = MQTT_PLUGIN_LIB_NAME + }, + { + .plugin_kind = PLUGIN_KIND_SYSTEM, + .adapter_type = ADAPTER_TYPE_DRIVER, + .plugin_name = MODBUS_PLUGIN_NAME, + .plugin_lib_name = MODBUS_PLUGIN_LIB_NAME + }, + */ +}; +// clang-format on +#define SYSTEM_PLUGIN_INFO_SIZE \ + (sizeof(system_plugin_infos) / sizeof(system_plugin_infos[0])) static int init_bind_info(manager_bind_info_t *mng_bind_info) { @@ -133,9 +212,7 @@ static int init_bind_info(manager_bind_info_t *mng_bind_info) neu_panic("Failed to initialize mutex and cv in manager_bind_info"); } - mng_bind_info->bind_count = 0; - mng_bind_info->bind_adapters = &g_bind_adapters[0]; - + mng_bind_info->bind_count = 0; return 0; } @@ -160,7 +237,7 @@ static int manager_add_config(neu_manager_t *manager) static uint32_t manager_get_adapter_id(neu_manager_t *manager) { - uint32_t adapter_id; + adapter_id_t adapter_id; adapter_id = manager->new_adapter_id++; return adapter_id; @@ -168,7 +245,6 @@ static uint32_t manager_get_adapter_id(neu_manager_t *manager) static void manager_bind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg) { - int rv = 0; neu_adapter_t *adapter; adapter = (neu_adapter_t *) arg; @@ -180,7 +256,6 @@ static void manager_bind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg) static void manager_unbind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg) { - int rv = 0; neu_adapter_t *adapter; adapter = (neu_adapter_t *) arg; @@ -190,18 +265,84 @@ static void manager_unbind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg) return; } -static int manager_add_adapter(neu_manager_t *manager, adapter_add_cmd_t *cmd) +// Return SIZE_MAX if can't find a adapter +static size_t find_adapter_by_id(vector_t *adapters, adapter_id_t id) { - int rv = 0; + size_t index = SIZE_MAX; + adapter_reg_entity_t *reg_entity; - return rv; + VECTOR_FOR_EACH(adapters, iter) + { + reg_entity = (adapter_reg_entity_t *) iterator_get(&iter); + if (reg_entity->adapter_id == id) { + index = iterator_index(adapters, &iter); + break; + } + } + + return index; } -static int manager_remove_adapter( - neu_manager_t *manager, neu_adapter_t *adapter) +// The output parameter p_adapter hold a new adapter +static adapter_id_t manager_reg_adapter(neu_manager_t *manager, + adapter_reg_param_t *reg_param, neu_adapter_t **p_adapter) { - int rv = 0; + adapter_id_t adapter_id; + neu_adapter_t * adapter; + neu_adapter_info_t adapter_info; + plugin_reg_info_t plugin_reg_info; + if (reg_param->plugin_id.id_val != 0) { + plugin_manager_get_reg_info( + manager->plugin_manager, reg_param->plugin_id, &plugin_reg_info); + } else { + plugin_manager_get_reg_info_by_name( + manager->plugin_manager, reg_param->plugin_name, &plugin_reg_info); + } + + adapter_id = 0; + adapter_info.id = manager_get_adapter_id(manager); + adapter_info.type = reg_param->adapter_type; + adapter_info.name = reg_param->adapter_name; + adapter_info.plugin_id = plugin_reg_info.plugin_id; + adapter_info.plugin_lib_name = plugin_reg_info.plugin_lib_name; + adapter = neu_adapter_create(&adapter_info); + if (adapter != NULL) { + adapter_reg_entity_t reg_entity; + + adapter_id = adapter_info.id; + reg_entity.adapter_id = adapter_info.id; + reg_entity.adapter = adapter; + reg_entity.bind_count = 0; + vector_push_back(&manager->reg_adapters, ®_entity); + if (p_adapter != NULL) { + *p_adapter = adapter; + } + } + return adapter_id; +} + +static int manager_unreg_adapter(neu_manager_t *manager, adapter_id_t id) +{ + int rv = 0; + size_t index; + vector_t * reg_adapters; + neu_adapter_t *adapter; + + adapter = NULL; + reg_adapters = &manager->reg_adapters; + index = find_adapter_by_id(reg_adapters, id); + if (index != SIZE_MAX) { + adapter_reg_entity_t *reg_entity; + + reg_entity = (adapter_reg_entity_t *) vector_get(reg_adapters, index); + adapter = reg_entity->adapter; + vector_erase(reg_adapters, index); + } + + if (adapter != NULL) { + neu_adapter_destroy(adapter); + } return rv; } @@ -242,6 +383,81 @@ static int manager_stop_adapter(neu_manager_t *manager, neu_adapter_t *adapter) return rv; } +// Call this function before start manager loop, so it don't need lock +static int register_default_plugins(neu_manager_t *manager) +{ + int rv = 0; + int i, j; + plugin_id_t plugin_id; + const plugin_reg_param_t *reg_param; + + for (i = 0; i < DEFAULT_PLUGIN_COUNT; i++) { + plugin_id.id_val = 0; + for (j = 0; j < SYSTEM_PLUGIN_INFO_SIZE; j++) { + if (strcmp(system_plugin_infos[j].plugin_lib_name, + default_plugin_lib_names[i]) == 0) { + break; + } + } + + if (j < SYSTEM_PLUGIN_INFO_SIZE) { + reg_param = &system_plugin_infos[j]; + plugin_id = + plugin_manager_reg_plugin(manager->plugin_manager, reg_param); + if (plugin_id.id_val == 0) { + log_warn( + "Failed to register plugin: %s", reg_param->plugin_name); + } + } + } + + return rv; +} + +// Call this function after quit manager loop, so it don't need lock +static void unregister_all_reg_plugins(neu_manager_t *manager) +{ + plugin_manager_unreg_all_plugins(manager->plugin_manager); + return; +} + +// Call this function before start manager loop, so it don't need lock +static void add_and_start_default_adapters(neu_manager_t *manager) +{ + int i; + adapter_id_t id; + neu_adapter_t *p_adapter; + + for (i = 0; i < DEFAULT_ADAPTER_ADD_INFO_SIZE; i++) { + id = manager_reg_adapter(manager, + (adapter_reg_param_t *) &default_adapter_reg_params[i], &p_adapter); + if (id != 0 && p_adapter != NULL) { + manager_start_adapter(manager, p_adapter); + } + } + return; +} + +// Call this function after quit manager loop, so it don't need lock +static void stop_and_destroy_bind_adapters(neu_manager_t *manager) +{ + vector_t * adapters; + adapter_reg_entity_t *reg_entity; + + adapters = &manager->reg_adapters; + VECTOR_FOR_EACH(adapters, iter) + { + neu_adapter_t *adapter; + + reg_entity = (adapter_reg_entity_t *) iterator_get(&iter); + adapter = reg_entity->adapter; + if (adapter != NULL) { + manager_stop_adapter(manager, adapter); + manager_unreg_adapter(manager, reg_entity->adapter_id); + } + } +} + static void manager_loop(void *arg) { int rv; @@ -263,12 +479,9 @@ static void manager_loop(void *arg) manager->state = MANAGER_STATE_RUNNING; nng_mtx_unlock(manager->mtx); -#ifdef NEU_HAS_SAMPLE_ADAPTER - sample_adapter_info.id = manager_get_adapter_id(manager); - sample_adapter = neu_adapter_create(&sample_adapter_info); - manager_start_adapter(manager, sample_adapter); -#endif - + register_default_plugins(manager); + add_and_start_default_adapters(manager); + log_info("Start message loop of neu_manager"); while (1) { nng_msg *msg; @@ -364,10 +577,9 @@ static void manager_loop(void *arg) nng_msg_free(msg); } -#ifdef NEU_HAS_SAMPLE_ADAPTER - manager_stop_adapter(manager, sample_adapter); - neu_adapter_destroy(sample_adapter); -#endif + log_info("End message loop of neu_manager"); + stop_and_destroy_bind_adapters(manager); + unregister_all_reg_plugins(manager); nng_close(manager_bind->mng_sock); return; } @@ -393,7 +605,18 @@ neu_manager_t *neu_manager_create() return NULL; } + rv = vector_setup(&manager->reg_adapters, DEFAULT_ADAPTER_REG_COUNT, + sizeof(adapter_reg_entity_t)); + if (rv != 0) { + neu_panic("Failed to create vector of registered adapters"); + } + init_bind_info(&manager->bind_info); + manager->plugin_manager = plugin_manager_create(); + if (manager->plugin_manager == NULL) { + neu_panic("Failed to create plugin manager"); + } + nng_thread_create(&manager->thrd, manager_loop, manager); return manager; } @@ -401,8 +624,10 @@ neu_manager_t *neu_manager_create() void neu_manager_destroy(neu_manager_t *manager) { nng_thread_destroy(manager->thrd); - nng_mtx_free(manager->mtx); + plugin_manager_destroy(manager->plugin_manager); uninit_bind_info(&manager->bind_info); + vector_destroy(&manager->reg_adapters); + nng_mtx_free(manager->mtx); free(manager); return; } diff --git a/src/core/plugin_info.h b/src/core/plugin_info.h new file mode 100644 index 00000000..6fad39aa --- /dev/null +++ b/src/core/plugin_info.h @@ -0,0 +1,27 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef _PLUGIN_INFO_H +#define _PLUGIN_INFO_H + +typedef struct plugin_id { + uint32_t id_val; +} plugin_id_t; + +#endif diff --git a/src/core/plugin_manager.c b/src/core/plugin_manager.c index 968b5060..ee9f9d97 100644 --- a/src/core/plugin_manager.c +++ b/src/core/plugin_manager.c @@ -22,26 +22,186 @@ #include #include +#include +#include + +#include "adapter/adapter_internal.h" +#include "neu_panic.h" #include "neu_vector.h" #include "plugin_manager.h" -typedef struct plugin_reg_info { -} plugin_reg_info_t; +typedef struct plugin_reg_entity { + plugin_id_t plugin_id; + plugin_kind_e plugin_kind; + adapter_type_e adapter_type; + const char * plugin_name; + const char * plugin_lib_name; +} plugin_reg_entity_t; struct plugin_manager { - const plugin_reg_info_t *builtin_plugins; - vector_t reg_plugins; + nng_mtx * mtx; + plugin_id_t new_plugin_id; + // for cache availabale id + vector_t available_ids; + vector_t system_plugins; + vector_t custom_plugins; }; -// Add builtin plugins -// +#define SYSTEM_PLUGIN_COUNT_DEFAULT 8 +static int init_system_plugins(vector_t *plugins_vec) +{ + int rv = 0; + + rv = vector_setup( + plugins_vec, SYSTEM_PLUGIN_COUNT_DEFAULT, sizeof(plugin_reg_entity_t)); + return rv; +} + +static void uninit_system_plugins(vector_t *plugins_vec) +{ + vector_destroy(plugins_vec); +} + +#define CUSTOM_PLUGIN_COUNT_DEFAULT 16 +static int init_custom_plugins(vector_t *plugins_vec) +{ + int rv = 0; + + rv = vector_setup( + plugins_vec, CUSTOM_PLUGIN_COUNT_DEFAULT, sizeof(plugin_reg_entity_t)); + return rv; +} + +static void uninit_custom_plugins(vector_t *plugins_vec) +{ + plugin_reg_entity_t *reg_entity; + + VECTOR_FOR_EACH(plugins_vec, iter) + { + reg_entity = (plugin_reg_entity_t *) iterator_get(&iter); + free((void *) reg_entity->plugin_name); + free((void *) reg_entity->plugin_lib_name); + } + vector_destroy(plugins_vec); +} + +// Return SIZE_MAX if can't find a plugin +static size_t find_plugin_by_id(vector_t *plugins, plugin_id_t id) +{ + size_t index = SIZE_MAX; + plugin_reg_entity_t *reg_entity; + + VECTOR_FOR_EACH(plugins, iter) + { + reg_entity = (plugin_reg_entity_t *) iterator_get(&iter); + if (reg_entity->plugin_id.id_val == id.id_val) { + index = iterator_index(plugins, &iter); + break; + } + } + + return index; +} + +// Return SIZE_MAX if can't find a plugin +static size_t find_plugin_by_name(vector_t *plugins, const char *plugin_name) +{ + size_t index = SIZE_MAX; + plugin_reg_entity_t *reg_entity; + + VECTOR_FOR_EACH(plugins, iter) + { + reg_entity = (plugin_reg_entity_t *) iterator_get(&iter); + if (strcmp(reg_entity->plugin_name, plugin_name) == 0) { + index = iterator_index(plugins, &iter); + break; + } + } + + return index; +} + +static plugin_id_t plugin_manager_get_plugin_id( + plugin_manager_t *plugin_mng, const plugin_reg_param_t *param) +{ + size_t index; + plugin_id_t plugin_id; + vector_t * plugins; + + nng_mtx_lock(plugin_mng->mtx); + plugin_id = plugin_mng->new_plugin_id; + plugin_mng->new_plugin_id.id_val++; + nng_mtx_unlock(plugin_mng->mtx); + + plugins = &plugin_mng->system_plugins; + if (param->plugin_kind == PLUGIN_KIND_CUSTOM) { + plugins = &plugin_mng->custom_plugins; + } + + // Check if there is a plugin with same name + nng_mtx_lock(plugin_mng->mtx); + index = find_plugin_by_name(plugins, param->plugin_name); + if (index != SIZE_MAX) { + plugin_id.id_val = 0; + return plugin_id; + } + nng_mtx_unlock(plugin_mng->mtx); + + // TODO: get unique id for plugin + return plugin_id; +} + +static void reg_entity_init(plugin_reg_entity_t *reg_entity, + plugin_id_t plugin_id, const plugin_reg_param_t *param) +{ + reg_entity->plugin_id = plugin_id; + reg_entity->plugin_kind = param->plugin_kind; + reg_entity->adapter_type = param->adapter_type; + reg_entity->plugin_name = param->plugin_name; + reg_entity->plugin_lib_name = param->plugin_lib_name; + return; +} + +static void reg_entity_uninit(plugin_reg_entity_t *reg_entity) +{ + reg_entity->plugin_id.id_val = 0; + reg_entity->adapter_type = 0; + if (reg_entity->plugin_kind == PLUGIN_KIND_CUSTOM) { + free((void *) reg_entity->plugin_name); + free((void *) reg_entity->plugin_lib_name); + } + return; +} plugin_manager_t *plugin_manager_create() { + int rv, rv1; plugin_manager_t *plugin_mng; plugin_mng = (plugin_manager_t *) malloc(sizeof(plugin_manager_t)); if (plugin_mng == NULL) { + log_error("Out of memory for create plugin manager"); + return NULL; + } + + rv = nng_mtx_alloc(&plugin_mng->mtx); + if (rv != 0) { + log_error("Failed to initialize mutex in plugin manager"); + return NULL; + } + + plugin_mng->new_plugin_id.id_val = 1; + rv = init_system_plugins(&plugin_mng->system_plugins); + rv1 = init_custom_plugins(&plugin_mng->custom_plugins); + if (rv != 0 || rv1 != 0) { + log_error("Failed to initialize register table of plugins"); + return NULL; + } + + rv = vector_setup(&plugin_mng->available_ids, CUSTOM_PLUGIN_COUNT_DEFAULT, + sizeof(plugin_id_t)); + if (rv != 0) { + log_error("Failed to initialize vector of available ids"); return NULL; } @@ -54,21 +214,216 @@ void plugin_manager_destroy(plugin_manager_t *plugin_mng) return; } + vector_destroy(&plugin_mng->available_ids); + uninit_custom_plugins(&plugin_mng->custom_plugins); + uninit_system_plugins(&plugin_mng->system_plugins); + nng_mtx_free(plugin_mng->mtx); free(plugin_mng); } +/* If there already has same plugin registered, then return 0 + */ plugin_id_t plugin_manager_reg_plugin( - plugin_manager_t *plugin_mng, plugin_reg_info_t *info) + plugin_manager_t *plugin_mng, const plugin_reg_param_t *param) { - plugin_id_t plugin_id; + int rv; + int retry_count; + vector_t * plugins; + plugin_id_t plugin_id; + plugin_reg_entity_t reg_entity; + plugins = &plugin_mng->system_plugins; + if (param->plugin_kind == PLUGIN_KIND_CUSTOM) { + plugins = &plugin_mng->custom_plugins; + } + + plugin_id = plugin_manager_get_plugin_id(plugin_mng, param); + if (plugin_id.id_val == 0) { + // There has same registered plugin + log_warn("A plugin with same name has already been registered"); + return plugin_id; + } + + reg_entity_init(®_entity, plugin_id, param); + log_info("Register the plugin: %s", reg_entity.plugin_name); + + rv = -1; + retry_count = 3; + useconds_t sleep_time = 300; + while (retry_count-- > 0 && rv != 0) { + nng_mtx_lock(plugin_mng->mtx); + rv = vector_push_back(plugins, ®_entity); + nng_mtx_unlock(plugin_mng->mtx); + if (rv != 0) { + usleep(sleep_time); + sleep_time *= 2; + } + } + if (rv != 0) { + neu_panic("Failed to register plugin"); + } return plugin_id; } int plugin_manager_unreg_plugin( - plugin_manager_t *plugin_mng, plugin_id_t plugin_id) + plugin_manager_t *plugin_mng, const plugin_reg_param_t *param) { - int rv = 0; + int rv = 0; + size_t index; + vector_t * plugins; + plugin_reg_entity_t *reg_entity; + plugins = &plugin_mng->system_plugins; + if (param->plugin_kind == PLUGIN_KIND_CUSTOM) { + plugins = &plugin_mng->custom_plugins; + } + + nng_mtx_lock(plugin_mng->mtx); + index = find_plugin_by_name(plugins, (const char *) param->plugin_name); + if (index == SIZE_MAX) { + reg_entity = (plugin_reg_entity_t *) vector_get(plugins, index); + reg_entity_uninit(reg_entity); + vector_erase(plugins, index); + } + nng_mtx_unlock(plugin_mng->mtx); + if (index == SIZE_MAX) { + log_info("Unregister the plugin: %s", reg_entity->plugin_name); + } return rv; } + +void plugin_manager_unreg_all_plugins(plugin_manager_t *plugin_mng) +{ + vector_t * plugins; + plugin_reg_entity_t *reg_entity; + + plugins = &plugin_mng->system_plugins; + VECTOR_FOR_EACH(plugins, iter) + { + reg_entity = (plugin_reg_entity_t *) iterator_get(&iter); + reg_entity_uninit(reg_entity); + break; + } + vector_clear(plugins); + + plugins = &plugin_mng->custom_plugins; + VECTOR_FOR_EACH(plugins, iter) + { + reg_entity = (plugin_reg_entity_t *) iterator_get(&iter); + reg_entity_uninit(reg_entity); + break; + } + vector_clear(plugins); + return; +} + +int plugin_manager_update_plugin( + plugin_manager_t *plugin_mng, const plugin_reg_param_t *param) +{ + int rv = 0; + size_t index; + vector_t *plugins; + + plugins = &plugin_mng->system_plugins; + if (param->plugin_kind == PLUGIN_KIND_CUSTOM) { + plugins = &plugin_mng->custom_plugins; + } + + nng_mtx_lock(plugin_mng->mtx); + index = find_plugin_by_name(plugins, (const char *) param->plugin_name); + if (index == SIZE_MAX) { + plugin_id_t plugin_id; + plugin_reg_entity_t *old_reg_entity; + plugin_reg_entity_t reg_entity; + + old_reg_entity = (plugin_reg_entity_t *) vector_get(plugins, index); + plugin_id = old_reg_entity->plugin_id; + reg_entity_uninit(old_reg_entity); + + reg_entity_init(®_entity, plugin_id, param); + vector_assign(plugins, index, ®_entity); + reg_entity_uninit(®_entity); + } else { + // Nothing to update + rv = -1; + } + nng_mtx_unlock(plugin_mng->mtx); + return rv; +} + +void reg_info_from_reg_entity( + plugin_reg_info_t *reg_info, plugin_reg_entity_t *reg_entity) +{ + reg_info->plugin_id = reg_entity->plugin_id; + reg_info->plugin_kind = reg_entity->plugin_kind; + reg_info->adapter_type = reg_entity->adapter_type; + reg_info->plugin_name = reg_entity->plugin_name; + reg_info->plugin_lib_name = reg_entity->plugin_lib_name; + return; +} + +int plugin_manager_get_reg_info(plugin_manager_t *plugin_mng, + plugin_id_t plugin_id, plugin_reg_info_t *reg_info) +{ + size_t index; + vector_t * plugins; + plugin_reg_entity_t *reg_entity; + + if (plugin_mng == NULL || reg_info == NULL) { + return -1; + } + + plugins = &plugin_mng->system_plugins; + index = find_plugin_by_id(plugins, plugin_id); + if (index != SIZE_MAX) { + reg_entity = (plugin_reg_entity_t *) vector_get(plugins, index); + reg_info_from_reg_entity(reg_info, reg_entity); + return 0; + } + + plugins = &plugin_mng->custom_plugins; + index = find_plugin_by_id(plugins, plugin_id); + if (index != SIZE_MAX) { + reg_entity = (plugin_reg_entity_t *) vector_get(plugins, index); + reg_info_from_reg_entity(reg_info, reg_entity); + return 0; + } + + return -1; +} + +int plugin_manager_get_reg_info_by_name(plugin_manager_t *plugin_mng, + const char *plugin_name, plugin_reg_info_t *reg_info) +{ + size_t index; + vector_t * plugins; + plugin_reg_entity_t *reg_entity; + + if (plugin_mng == NULL || plugin_name == NULL || reg_info == NULL) { + return -1; + } + + plugins = &plugin_mng->system_plugins; + index = find_plugin_by_name(plugins, plugin_name); + if (index != SIZE_MAX) { + reg_entity = (plugin_reg_entity_t *) vector_get(plugins, index); + reg_info_from_reg_entity(reg_info, reg_entity); + return 0; + } + + plugins = &plugin_mng->custom_plugins; + index = find_plugin_by_name(plugins, plugin_name); + if (index != SIZE_MAX) { + reg_entity = (plugin_reg_entity_t *) vector_get(plugins, index); + reg_info_from_reg_entity(reg_info, reg_entity); + return 0; + } + + return -1; +} + +void plugin_manager_dump(plugin_manager_t *plugin_mng) +{ + // TODO: dump system plugins and custom plugins + return; +} diff --git a/src/core/plugin_manager.h b/src/core/plugin_manager.h index 130e1d7c..c6676cc0 100644 --- a/src/core/plugin_manager.h +++ b/src/core/plugin_manager.h @@ -22,10 +22,50 @@ #include -typedef struct plugin_id { - uint32_t id; -} plugin_id_t; +#include "adapter/adapter_internal.h" +#include "core/plugin_info.h" + +typedef enum plugin_kind { + PLUGIN_KIND_SYSTEM, + PLUGIN_KIND_CUSTOM +} plugin_kind_e; + +typedef struct plugin_reg_param { + plugin_kind_e plugin_kind; + adapter_type_e adapter_type; + // The buffer should be move to plugin manager, don't free it + const char *plugin_name; + // The buffer should be move to plugin manager, don't free it + const char *plugin_lib_name; +} plugin_reg_param_t; + +typedef struct plugin_reg_info { + plugin_id_t plugin_id; + plugin_kind_e plugin_kind; + adapter_type_e adapter_type; + // The buffer is reference from plugin entity of register table, + // don't free it + const char *plugin_name; + // The buffer is reference from plugin entity of register table, + // don't free it + const char *plugin_lib_name; +} plugin_reg_info_t; typedef struct plugin_manager plugin_manager_t; +plugin_manager_t *plugin_manager_create(); +void plugin_manager_destroy(plugin_manager_t *plugin_mng); +plugin_id_t plugin_manager_reg_plugin( + plugin_manager_t *plugin_mng, const plugin_reg_param_t *param); +int plugin_manager_unreg_plugin( + plugin_manager_t *plugin_mng, const plugin_reg_param_t *param); +int plugin_manager_update_plugin( + plugin_manager_t *plugin_mng, const plugin_reg_param_t *param); +int plugin_manager_get_reg_info(plugin_manager_t *plugin_mng, + plugin_id_t plugin_id, plugin_reg_info_t *reg_info); +int plugin_manager_get_reg_info_by_name(plugin_manager_t *plugin_mng, + const char *plugin_name, plugin_reg_info_t *reg_info); + +void plugin_manager_unreg_all_plugins(plugin_manager_t *plugin_mng); +void plugin_manager_dump(plugin_manager_t *plugin_mng); #endif diff --git a/src/main.c b/src/main.c index f581df59..907aff49 100644 --- a/src/main.c +++ b/src/main.c @@ -50,7 +50,7 @@ static void init() nng_mtx_alloc(&log_mtx); log_set_lock(log_lock, log_mtx); log_set_level(LOG_DEBUG); - FILE *g_logfile = fopen("neuron.log", "w"); + FILE *g_logfile = fopen("rest-server.log", "a"); if (g_logfile == NULL) { fprintf(stderr, "Failed to open logfile when"