GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-28 05:56:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

Implement the plugin manager and adapter manager

Browse Source
This commit is contained in:
Parker Liu
2021-08-02 15:05:16 +08:00
parent f9493475e2
commit bd65bc7c3a
8 changed files with 784 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+5 -5
View File
@@ -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})
src/adapter/adapter.c
+44 -24
View File
@@ -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;
}
src/adapter/adapter_internal.h
+6 -3
View File
@@ -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);
src/core/neu_manager.c
+293 -68
View File
@@ -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, &reg_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;
}
src/core/plugin_info.h
+27
View File
@@ -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
src/core/plugin_manager.c
+365 -10
View File
@@ -22,26 +22,186 @@
#include <string.h>
#include <unistd.h>
#include <nng/nng.h>
#include <nng/supplemental/util/platform.h>
#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(&reg_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, &reg_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(&reg_entity, plugin_id, param);
vector_assign(plugins, index, &reg_entity);
reg_entity_uninit(&reg_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;
}
src/core/plugin_manager.h
+43 -3
View File
@@ -22,10 +22,50 @@
#include <stdint.h>
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
src/main.c
+1 -1
View File
@@ -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"