gfwleak/stellar-stellar
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ feat(module manager): from plugin manager to module manager

Browse Source
This commit is contained in:
yangwei
2024-09-14 12:18:26 +08:00
parent 0b142cd0bb
commit 1f55a6f240
17 changed files with 1987 additions and 608 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
include/stellar/module_manager.h Normal file
View File

@@ -0,0 +1,40 @@
#pragma once
#ifdef __cplusplus
extern "C"
{
#endif
#include "stellar/mq.h"
struct stellar_module;
struct stellar_module *stellar_module_new(const char *name);
void stellar_module_free(struct stellar_module *mod);
void * stellar_module_get_ctx(struct stellar_module *mod);
void stellar_module_set_ctx(struct stellar_module *mod, void *ctx);
const char *stellar_module_get_name(struct stellar_module* mod);
struct stellar_module_manager;
typedef struct stellar_module *module_on_init_func(struct stellar_module_manager *mod_mgr);
typedef void module_on_exit_func(struct stellar_module_manager *mod_mgr, struct stellar_module *mod);
struct stellar_module_manager *stellar_module_manager_new(const char *module_spec_toml_path, int max_thread_num, struct mq_schema *mq_schema);
void stellar_module_manager_free(struct stellar_module_manager *mod_mgr);
struct stellar_module *stellar_module_manager_get_module(struct stellar_module_manager *mod_mgr, const char *module_name);
void stellar_module_manager_register_thread(struct stellar_module_manager* mod_mgr, int thread_id, struct mq_runtime *mq_rt);
// return -1 on error
int stellar_module_manager_get_thread_id(struct stellar_module_manager* mod_mgr);
int stellar_module_manager_get_max_thread_num(struct stellar_module_manager* mod_mgr);
struct mq_schema *stellar_module_get_mq_schema(struct stellar_module_manager *mod_mgr);
struct mq_runtime *stellar_module_get_mq_runtime(struct stellar_module_manager *mod_mgr);
#ifdef __cplusplus
}
#endif

10
include/stellar/stellar.h
View File

@@ -7,15 +7,12 @@ extern "C"
#include <stdint.h>
#include "stellar/mq.h"
#include "stellar/log.h"
#include "stellar/packet.h"
struct stellar;
//return plugin_env
typedef void *plugin_on_load_func(struct stellar *st);
typedef void plugin_on_unload_func(void *plugin_env);
typedef void plugin_on_packet_func(struct packet *pkt, void *on_packet_cb_arg);
//return 0 if success, otherwise return -1.
@@ -28,8 +25,6 @@ int stellar_polling_subscribe(struct stellar *st, plugin_on_polling_func on_pol
void stellar_emit_datapath_telemetry(struct packet *pkt, const char * module, const char *str);
int stellar_get_worker_thread_num(struct stellar *st);
uint16_t stellar_get_current_thread_index();
// only send user build packet, can't send packet which come from network
void stellar_send_build_packet(struct stellar *st, struct packet *pkt);
@@ -40,9 +35,6 @@ void stellar_loopbreak(struct stellar *st);
void stellar_reload_log_level(struct stellar *st);
struct logger *stellar_get_logger(struct stellar *st);
struct mq_schema *stellar_get_mq_schema(struct stellar *st);
struct mq_runtime *stellar_get_mq_runtime(struct stellar *st);
#ifdef __cplusplus
}
#endif

2
infra/CMakeLists.txt
View File

@@ -1,4 +1,4 @@
set(INFRA exdata mq tuple packet_manager packet_io ip_reassembly tcp_reassembly session_manager plugin_manager)
set(INFRA exdata mq tuple packet_manager packet_io ip_reassembly tcp_reassembly session_manager module_manager)
set(DEPS bitmap dablooms interval_tree logger nmx_pool rbtree timeout toml)
#set(DECODERS http lpi)
set(WHOLE_ARCHIVE ${DEPS} ${INFRA} ${DECODERS})

1
infra/exdata/CMakeLists.txt
View File

@@ -1,3 +1,4 @@
add_library(exdata exdata.c)
# //TODO: Add test
#add_subdirectory(test)

11
infra/plugin_manager/test/CMakeLists.txt → infra/exdata/test/CMakeLists.txt
View File

@@ -1,13 +1,14 @@
add_executable(gtest_plugin_manager
plugin_manager_gtest_main.cpp
add_executable(gtest_module_manager
module_manager_gtest_main.cpp
)
include_directories(${CMAKE_SOURCE_DIR}/infra/plugin_manager/)
include_directories(${CMAKE_SOURCE_DIR}/infra/tuple/)
target_link_libraries(
gtest_plugin_manager
plugin_manager
gtest_module_manager
module_manager
dl
"-rdynamic"
gtest
@@ -15,4 +16,4 @@ target_link_libraries(
)
include(GoogleTest)
gtest_discover_tests(gtest_plugin_manager)
gtest_discover_tests(gtest_module_manager)

40
infra/plugin_manager/test/plugin_manager_gtest_main.cpp → infra/exdata/test/gtest_module_manager_main.cpp
View File

@@ -2,46 +2,8 @@
#include <gtest/gtest.h>
#include "plugin_manager.h"
#include "plugin_manager_gtest_mock.h"
#define STELLAR_INTRINSIC_TOPIC_NUM 0
#define TOPIC_NAME_MAX 512
#if 0
void whitebox_test_plugin_manager_intrisic_metadata(struct stellar *st, struct plugin_manager_schema *plug_mgr)
{
SCOPED_TRACE("whitebox test intrisic metadata");
EXPECT_TRUE(plug_mgr!=NULL);
EXPECT_EQ(plug_mgr->st, st);
//load spec null
EXPECT_TRUE(plug_mgr->plugin_load_specs_array==NULL);
//session exdata schema null
EXPECT_TRUE(plug_mgr->exdata_schema!=NULL);
//stellar mq schema null
EXPECT_TRUE(plug_mgr->stellar_mq_schema_array==NULL);
//registered plugin array null
EXPECT_TRUE(plug_mgr->registered_polling_plugin_array==NULL);
EXPECT_TRUE(plug_mgr->registered_packet_plugin_array==NULL);
EXPECT_TRUE(plug_mgr->per_thread_data!=NULL);
int thread_num=stellar_get_worker_thread_num(st);
for(int i=0; i<thread_num; i++)
{
EXPECT_TRUE(plug_mgr->per_thread_data[i].exdata_array==NULL);
EXPECT_TRUE(plug_mgr->per_thread_data[i].dealth_letter_queue==NULL);
for(int j=0; j<STELLAR_MQ_PRIORITY_MAX; j++)
EXPECT_TRUE(plug_mgr->per_thread_data[i].priority_mq[j]==NULL);
}
}
#endif
#include "module_manager/module_manager_interna.h"
/***********************************
* TEST PLUGIN MANAGER INIT & EXIT *

8
infra/module_manager/CMakeLists.txt Normal file
View File

@@ -0,0 +1,8 @@
add_library(module_manager module_manager.c)
target_include_directories(module_manager PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_include_directories(module_manager PUBLIC ${CMAKE_SOURCE_DIR}/include/)
target_include_directories(module_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra/)
target_include_directories(module_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/)
target_link_libraries(module_manager PUBLIC toml ${CMAKE_DL_LIBS})
#add_subdirectory(test)

234
infra/module_manager/module_manager.c Normal file
View File

@@ -0,0 +1,234 @@
#include "module_manager_interna.h"
#include "stellar/module_manager.h"
#include "stellar/utils.h"
#include "toml/toml.h"
#include <dlfcn.h>
#include <stdbool.h>
UT_icd module_specs_icd = {sizeof(struct module_specific), NULL, NULL, NULL};
static struct module_specific *module_specs_load(const char *toml_conf_path, int *mod_num)
{
*mod_num = 0;
FILE* fp = fopen(toml_conf_path, "r");
if(fp==NULL)return NULL;
char errbuf[256];
toml_table_t* conf = toml_parse_file(fp, errbuf, sizeof(errbuf));
fclose(fp);
if (!conf) {
fprintf(stderr, "Error parsing toml: %s\n", errbuf);
return NULL;
}
struct module_specific* mod_spec=NULL;
toml_array_t* mod_array = toml_array_in(conf, "module");
if(mod_array==NULL)goto MODULE_SPEC_LOAD_ERROR;
*mod_num = toml_array_nelem(mod_array);
mod_spec = CALLOC(struct module_specific, *mod_num);
for (int i = 0; i < *mod_num; i++) {
toml_table_t* toml_mod = toml_table_at(mod_array, i);
const char *path_raw = toml_raw_in(toml_mod, "path");
const char *init_func_name_raw = toml_raw_in(toml_mod, "init");
const char *exit_func_name_raw = toml_raw_in(toml_mod, "exit");
char *path = NULL;
char *init_func_name = NULL;
char *exit_func_name = NULL;
if (toml_rtos(path_raw, &path) || toml_rtos(init_func_name_raw, &init_func_name) ||
toml_rtos(exit_func_name_raw, &exit_func_name))
{
goto MODULE_SPEC_LOAD_ERROR;
}
void* handle = dlopen(path, RTLD_NOW|RTLD_LAZY|RTLD_GLOBAL);
if (!handle) {
fprintf(stderr, "Error loading plugin %s: %s\n", path, dlerror());
goto MODULE_SPEC_LOAD_ERROR;
}
mod_spec[i].load_cb = (module_on_init_func *) dlsym(handle, init_func_name);
if (!mod_spec[i].load_cb) {
fprintf(stderr, "Could not load init function %s: %s\n", init_func_name, dlerror());
}
mod_spec[i].unload_cb = (module_on_exit_func *) dlsym(handle, exit_func_name);
if (!mod_spec[i].unload_cb) {
fprintf(stderr, "Could not load exit function %s: %s\n", exit_func_name, dlerror());
}
FREE(path);
FREE(init_func_name);
FREE(exit_func_name);
}
toml_free(conf);
return mod_spec;
MODULE_SPEC_LOAD_ERROR:
toml_free(conf);
if(mod_spec)FREE(mod_spec);
*mod_num=0;
return NULL;
}
/*******************************************
* stellar module manager API *
*******************************************/
struct stellar_module_manager *stellar_module_manager_new(const char *module_spec_toml_path, int max_thread_num, struct mq_schema *mq_schema)
{
int spec_num;
struct module_specific *specs = module_specs_load(module_spec_toml_path, &spec_num);
if(spec_num < 0)
{
return NULL;
}
struct stellar_module_manager *mod_mgr = CALLOC(struct stellar_module_manager, 1);
if(spec_num > 0)
{
utarray_new(mod_mgr->schema.module_specs_array,&module_specs_icd);
utarray_reserve(mod_mgr->schema.module_specs_array, spec_num);
}
mod_mgr->schema.max_thread_num=max_thread_num;
mod_mgr->schema.mq_schema=mq_schema;
// TODO: store module specific data in hash
for(int i = 0; i < spec_num; i++)
{
if (specs[i].load_cb != NULL)
{
//TODO: duplicate check mod_name
specs[i].mod=specs[i].load_cb(mod_mgr);
utarray_push_back(mod_mgr->schema.module_specs_array, &specs[i]);
}
}
FREE(specs);
return mod_mgr;
}
void stellar_module_manager_free(struct stellar_module_manager *mod_mgr)
{
if(mod_mgr==NULL)return;
struct module_specific *p=NULL;
if (mod_mgr->schema.module_specs_array)
{
while ((p = (struct module_specific *)utarray_next(mod_mgr->schema.module_specs_array, p)))
{
if (p->unload_cb)
p->unload_cb(mod_mgr, p->mod);
}
utarray_free(mod_mgr->schema.module_specs_array);
}
#if 0
if(plug_mgr->stellar_mq_schema_array)
{
for(unsigned int i = 0; i < utarray_len(plug_mgr->stellar_mq_schema_array); i++)
{
stellar_mq_destroy_topic( plug_mgr->st, i);
}
utarray_free(plug_mgr->stellar_mq_schema_array);
}
//if(plug_mgr->stellar_exdata_schema_array)utarray_free(plug_mgr->stellar_exdata_schema_array);
if(plug_mgr->registered_polling_plugin_array)utarray_free(plug_mgr->registered_polling_plugin_array);
if(plug_mgr->registered_packet_plugin_array)
{
struct registered_plugin_schema *s = NULL;
while ((s = (struct registered_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, s)))
{
if(s->registed_mq_subscriber_info)utarray_free(s->registed_mq_subscriber_info);
}
utarray_free(plug_mgr->registered_packet_plugin_array);
}
#endif
FREE(mod_mgr);
return;
}
int stellar_module_manager_get_max_thread_num(struct stellar_module_manager*mod_mgr)
{
if(mod_mgr==NULL)return -1;
return mod_mgr->schema.max_thread_num;
}
struct mq_schema *stellar_module_get_mq_schema(struct stellar_module_manager *mod_mgr)
{
if(mod_mgr==NULL)return NULL;
return mod_mgr->schema.mq_schema;
}
__thread int local_thread_id=-1;
__thread struct mq_runtime *local_mq_rt=NULL;
int stellar_module_manager_get_thread_id(struct stellar_module_manager* mod_mgr __unused)
{
return local_thread_id;
}
struct mq_runtime *stellar_module_get_mq_runtime(struct stellar_module_manager *mod_mgr __unused)
{
return local_mq_rt;
}
void stellar_module_manager_register_thread(struct stellar_module_manager* mod_mgr __unused, int thread_id, struct mq_runtime *mq_rt)
{
local_thread_id=thread_id;
local_mq_rt=mq_rt;
return;
}
struct stellar_module *stellar_module_manager_get_module(struct stellar_module_manager *mod_mgr, const char *module_name)
{
if(mod_mgr==NULL)return NULL;
struct module_specific *p=NULL;
if (mod_mgr->schema.module_specs_array)
{
while ((p = (struct module_specific *)utarray_next(mod_mgr->schema.module_specs_array, p)))
{
if(strcmp(p->mod->name, module_name)==0)
{
return p->mod;
}
}
}
return NULL;
}
/*******************************************
* stellar module API *
*******************************************/
struct stellar_module *stellar_module_new(const char *name)
{
struct stellar_module *mod = CALLOC(struct stellar_module, 1);
strncpy(mod->name, name, NAME_MAX);
return mod;
}
void stellar_module_free(struct stellar_module *mod)
{
if(mod==NULL)return;
FREE(mod);
return;
}
void * stellar_module_get_ctx(struct stellar_module *mod)
{
if(mod==NULL)return NULL;
return mod->module_ctx;
}
void stellar_module_set_ctx(struct stellar_module *mod, void *ctx)
{
if(mod==NULL)return;
mod->module_ctx=ctx;
return;
}
const char *stellar_module_get_name(struct stellar_module* mod)
{
if(mod==NULL)return NULL;
return mod->name;
}

41
infra/module_manager/module_manager_interna.h Normal file
View File

@@ -0,0 +1,41 @@
#pragma once
#ifdef __cplusplus
extern "C"
{
#endif
#include "stellar/module_manager.h"
#include "uthash/utarray.h"
#include "stellar/mq.h"
#include <limits.h>
struct stellar_module
{
char name[NAME_MAX];
void *module_ctx;
};
struct stellar_module_manager
{
struct
{
UT_array *module_specs_array;
int max_thread_num;
struct mq_schema *mq_schema;
}schema;
}__attribute__((aligned(sizeof(void*))));
struct module_specific
{
struct stellar_module *mod;
module_on_init_func *load_cb;
module_on_exit_func *unload_cb;
}__attribute__((aligned(sizeof(void*))));
#ifdef __cplusplus
}
#endif

19
infra/module_manager/test/CMakeLists.txt Normal file
View File

@@ -0,0 +1,19 @@
add_executable(gtest_module_manager
module_manager_gtest_main.cpp
)
include_directories(${CMAKE_SOURCE_DIR}/infra/plugin_manager/)
include_directories(${CMAKE_SOURCE_DIR}/infra/tuple/)
target_link_libraries(
gtest_module_manager
module_manager
dl
"-rdynamic"
gtest
gmock
)
include(GoogleTest)
gtest_discover_tests(gtest_module_manager)

1617
infra/module_manager/test/gtest_module_manager_main.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

11
infra/plugin_manager/CMakeLists.txt
View File

@@ -1,11 +0,0 @@
add_library(plugin_manager plugin_manager.c)
target_include_directories(plugin_manager PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/include/)
target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra/)
target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra/packet_manager)
target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra/session_manager)
target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra/tuple)
target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/)
target_link_libraries(plugin_manager PUBLIC session_manager bitmap toml exdata ${CMAKE_DL_LIBS})
add_subdirectory(test)

361
infra/plugin_manager/plugin_manager.c
View File

@@ -1,361 +0,0 @@
#include "plugin_manager_interna.h"
#include "stellar/utils.h"
#include "toml/toml.h"
#include <dlfcn.h>
#include <stdbool.h>
#include "stellar_core.h"
#if 0
void stellar_per_stage_message_counter_incby(struct plugin_manager_schema *plug_mgr, int tid, long long increment)
{
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt+=increment;
}
void stellar_per_stage_message_counter_set(struct plugin_manager_schema *plug_mgr, int tid, long long increment)
{
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt=increment;
}
bool stellar_per_stage_message_counter_overlimt(struct plugin_manager_schema *plug_mgr, int tid)
{
if(plug_mgr->per_thread_data[tid].pub_packet_msg_cnt >= plug_mgr->max_message_dispatch)return true;
return false;
}
#endif
UT_icd plugin_specs_icd = {sizeof(struct plugin_specific), NULL, NULL, NULL};
static struct plugin_specific *plugin_specs_load(const char *toml_conf_path, int *spec_num)
{
*spec_num = 0;
FILE* fp = fopen(toml_conf_path, "r");
if(fp==NULL)return NULL;
char errbuf[256];
toml_table_t* conf = toml_parse_file(fp, errbuf, sizeof(errbuf));
fclose(fp);
if (!conf) {
fprintf(stderr, "Error parsing toml: %s\n", errbuf);
return NULL;
}
struct plugin_specific* plugins=NULL;
toml_array_t* plugin_array = toml_array_in(conf, "plugin");
if(plugin_array==NULL)goto PLUGIN_SPEC_LOAD_ERROR;
*spec_num = toml_array_nelem(plugin_array);
plugins = CALLOC(struct plugin_specific, *spec_num);
for (int i = 0; i < *spec_num; i++) {
toml_table_t* plugin = toml_table_at(plugin_array, i);
const char *path_raw = toml_raw_in(plugin, "path");
const char *init_func_name_raw = toml_raw_in(plugin, "init");
const char *exit_func_name_raw = toml_raw_in(plugin, "exit");
char *path = NULL;
char *init_func_name = NULL;
char *exit_func_name = NULL;
if (toml_rtos(path_raw, &path) || toml_rtos(init_func_name_raw, &init_func_name) ||
toml_rtos(exit_func_name_raw, &exit_func_name))
{
goto PLUGIN_SPEC_LOAD_ERROR;
}
void* handle = dlopen(path, RTLD_NOW|RTLD_LAZY|RTLD_GLOBAL);
if (!handle) {
fprintf(stderr, "Error loading plugin %s: %s\n", path, dlerror());
goto PLUGIN_SPEC_LOAD_ERROR;
}
plugins[i].load_cb = (plugin_on_load_func *) dlsym(handle, init_func_name);
if (!plugins[i].load_cb) {
fprintf(stderr, "Could not load init function %s: %s\n", init_func_name, dlerror());
}
plugins[i].unload_cb = (plugin_on_unload_func *) dlsym(handle, exit_func_name);
if (!plugins[i].unload_cb) {
fprintf(stderr, "Could not load exit function %s: %s\n", exit_func_name, dlerror());
}
FREE(path);
FREE(init_func_name);
FREE(exit_func_name);
}
toml_free(conf);
return plugins;
PLUGIN_SPEC_LOAD_ERROR:
toml_free(conf);
if(plugins)FREE(plugins);
return NULL;
}
#if 0
static struct plugin_manager_per_thread_data *plugin_manager_per_thread_data_new(struct stellar *st)
{
if(st == NULL)return NULL;
int thread_num=stellar_get_worker_thread_num(st);
struct plugin_manager_per_thread_data *per_thread_data = CALLOC(struct plugin_manager_per_thread_data, thread_num);
return per_thread_data;
}
static void plugin_manager_per_thread_data_free(struct plugin_manager_per_thread_data *per_thread_data, struct stellar *st)
{
if(per_thread_data == NULL || st == NULL)return;
int thread_num=stellar_get_worker_thread_num(st);
struct plugin_manager_per_thread_data *p_data;
for (int i = 0; i < thread_num; i++)
{
p_data=per_thread_data+i;
exdata_handle_free(p_data->exdata_array);
}
FREE(per_thread_data);
return;
}
#endif
struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char *plugin_spec_file_path)
{
int spec_num;
struct plugin_specific *specs = plugin_specs_load(plugin_spec_file_path, &spec_num);
if(spec_num < 0)
{
return NULL;
}
struct plugin_manager_schema *plug_mgr = CALLOC(struct plugin_manager_schema, 1);
//plug_mgr->max_message_dispatch=max_msg_per_stage;
if(spec_num > 0)
{
utarray_new(plug_mgr->plugin_load_specs_array,&plugin_specs_icd);
utarray_reserve(plug_mgr->plugin_load_specs_array, spec_num);
}
plug_mgr->st = st;
stellar_set_plugin_manger(st, plug_mgr);
//plug_mgr->exdata_schema=exdata_schema_new();
for(int i = 0; i < spec_num; i++)
{
if (specs[i].load_cb != NULL)
{
specs[i].plugin_ctx=specs[i].load_cb(st);
utarray_push_back(plug_mgr->plugin_load_specs_array, &specs[i]);
}
}
FREE(specs);
//plug_mgr->per_thread_data = plugin_manager_per_thread_data_new(st);
return plug_mgr;
}
void plugin_manager_exit(struct plugin_manager_schema *plug_mgr)
{
if(plug_mgr==NULL)return;
struct plugin_specific *p=NULL;
if (plug_mgr->plugin_load_specs_array)
{
while ((p = (struct plugin_specific *)utarray_next(plug_mgr->plugin_load_specs_array, p)))
{
if (p->unload_cb)
p->unload_cb(p->plugin_ctx);
}
utarray_free(plug_mgr->plugin_load_specs_array);
}
#if 0
if(plug_mgr->stellar_mq_schema_array)
{
for(unsigned int i = 0; i < utarray_len(plug_mgr->stellar_mq_schema_array); i++)
{
stellar_mq_destroy_topic( plug_mgr->st, i);
}
utarray_free(plug_mgr->stellar_mq_schema_array);
}
//if(plug_mgr->stellar_exdata_schema_array)utarray_free(plug_mgr->stellar_exdata_schema_array);
if(plug_mgr->registered_polling_plugin_array)utarray_free(plug_mgr->registered_polling_plugin_array);
if(plug_mgr->registered_packet_plugin_array)
{
struct registered_plugin_schema *s = NULL;
while ((s = (struct registered_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, s)))
{
if(s->registed_mq_subscriber_info)utarray_free(s->registed_mq_subscriber_info);
}
utarray_free(plug_mgr->registered_packet_plugin_array);
}
#endif
//plugin_manager_per_thread_data_free(plug_mgr->per_thread_data, plug_mgr->st);
//exdata_schema_free(plug_mgr->exdata_schema);
FREE(plug_mgr);
return;
}
/*******************************
* STELLAR EXDATA *
*******************************/
#if 0
int stellar_exdata_new_index(struct stellar *st, const char *name, stellar_exdata_free *free_func,void *free_arg)
{
if(st==NULL || name==NULL)return -1;
struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
if(plug_mgr->exdata_schema==NULL)return -1;
return exdata_new_index(plug_mgr->exdata_schema, name, free_func, free_arg);
}
/*******************************
* PACKET EXDATA *
*******************************/
static struct exdata_runtime *per_thread_packet_exdata_arrary_get(struct plugin_manager_schema *plug_mgr)
{
if(plug_mgr==NULL || plug_mgr->exdata_schema == NULL)return NULL;
int tid=stellar_get_current_thread_index();
if(plug_mgr->per_thread_data[tid].exdata_array == NULL)
{
plug_mgr->per_thread_data[tid].exdata_array = exdata_handle_new(plug_mgr->exdata_schema);
}
return plug_mgr->per_thread_data[tid].exdata_array;
}
static void per_thread_packet_exdata_arrary_clean(struct plugin_manager_schema *plug_mgr)
{
if(plug_mgr==NULL || plug_mgr->exdata_schema == NULL)return;
struct exdata_runtime *per_thread_exdata_handle = per_thread_packet_exdata_arrary_get(plug_mgr);
return exdata_handle_reset(per_thread_exdata_handle);
}
int packet_exdata_set(struct packet *pkt, int idx, void *ex_ptr)
{
if(pkt == NULL)return -1;
struct plugin_manager_schema *plug_mgr = (struct plugin_manager_schema *)packet_get_user_data(pkt);
return exdata_set(per_thread_packet_exdata_arrary_get(plug_mgr), idx, ex_ptr);
}
void *packet_exdata_get(struct packet *pkt, int idx)
{
if(pkt == NULL)return NULL;
struct plugin_manager_schema *plug_mgr = (struct plugin_manager_schema *)packet_get_user_data(pkt);
return exdata_get( per_thread_packet_exdata_arrary_get(plug_mgr), idx);
}
/*******************************
* SESSION EXDATA *
*******************************/
int session_exdata_set(struct session *sess, int idx, void *ex_ptr)
{
struct exdata_runtime *sess_exdata = (struct exdata_runtime *)session_get_user_data(sess);
if(sess_exdata == NULL)return -1;
return exdata_set(sess_exdata,idx, ex_ptr);
}
void *session_exdata_get(struct session *sess, int idx)
{
struct exdata_runtime *sess_exdata = (struct exdata_runtime *)session_get_user_data(sess);
if(sess_exdata == NULL)return NULL;
return exdata_get(sess_exdata, idx);
}
#endif
#if 0
/*******************************
* PLUGIN MANAGER SESSION RUNTIME *
*******************************/
struct exdata_runtime *session_exdata_runtime_new(struct stellar *st)
{
struct plugin_manager_schema *plug_mgr=stellar_get_plugin_manager(st);
return exdata_handle_new(plug_mgr->exdata_schema);
}
void session_exdata_runtime_free(struct exdata_runtime *exdata_h)
{
return exdata_handle_free(exdata_h);
}
/*********************************************
* PLUGIN MANAGER PLUGIN *
*********************************************/
UT_icd registered_plugin_array_icd = {sizeof(struct registered_plugin_schema), NULL, NULL, NULL};
int stellar_plugin_register(struct stellar *st, plugin_on_packet_func on_packet_input, plugin_on_packet_func on_packet_output, void *plugin_env)
{
struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
if(plug_mgr->registered_packet_plugin_array == NULL)
{
utarray_new(plug_mgr->registered_packet_plugin_array, &registered_plugin_array_icd);
}
struct registered_plugin_schema packet_plugin_schema;
memset(&packet_plugin_schema, 0, sizeof(packet_plugin_schema));
packet_plugin_schema.on_packet[PACKET_STAGE_INPUT] = on_packet_input;
packet_plugin_schema.on_packet[PACKET_STAGE_OUTPUT] = on_packet_output;
packet_plugin_schema.plugin_env = plugin_env;
utarray_push_back(plug_mgr->registered_packet_plugin_array, &packet_plugin_schema);
return (utarray_len(plug_mgr->registered_packet_plugin_array)-1);// return packet plugin_id, equals to packet plugin arrary index
}
static void plugin_manager_on_packet(struct plugin_manager_schema *plug_mgr, struct packet *pkt, enum packet_stage in_out)
{
if(plug_mgr==NULL || plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
struct registered_plugin_schema *p=NULL;
//int tid=stellar_get_current_thread_index();
//stellar_per_stage_message_counter_set(plug_mgr, tid, 0);
while ((p = (struct registered_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, p)))
{
if(p->on_packet[in_out])
{
p->on_packet[in_out](pkt, p->plugin_env);
}
}
//stellar_mq_dispatch(plug_mgr->per_thread_data[tid].priority_mq, &plug_mgr->per_thread_data[tid].dealth_letter_queue);
return;
}
void plugin_manager_on_packet_input(struct plugin_manager_schema *plug_mgr, struct packet *pkt)
{
plugin_manager_on_packet(plug_mgr, pkt, PACKET_STAGE_INPUT);
}
void plugin_manager_on_packet_output(struct plugin_manager_schema *plug_mgr, struct packet *pkt)
{
if(plug_mgr == NULL || plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
plugin_manager_on_packet(plug_mgr, pkt, PACKET_STAGE_OUTPUT);
//int tid=stellar_get_current_thread_index();
//stellar_per_stage_message_counter_set(plug_mgr, tid, -1);
//stellar_mq_free(&plug_mgr->per_thread_data[tid].dealth_letter_queue,
// plug_mgr->stellar_mq_schema_array);
//per_thread_packet_exdata_arrary_clean(plug_mgr);
}
/*********************************************
* PLUGIN MANAGER POLLING PLUGIN *
*********************************************/
UT_icd registered_polling_plugin_array_icd = {sizeof(struct registered_polling_plugin_schema), NULL, NULL, NULL};
int stellar_on_polling_register(struct stellar *st, plugin_on_polling_func on_polling, void *plugin_env)
{
struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
if(plug_mgr->registered_polling_plugin_array == NULL)
{
utarray_new(plug_mgr->registered_polling_plugin_array, &registered_polling_plugin_array_icd);
}
struct registered_polling_plugin_schema polling_plugin_schema;
memset(&polling_plugin_schema, 0, sizeof(polling_plugin_schema));
polling_plugin_schema.on_polling = on_polling;
polling_plugin_schema.plugin_env = plugin_env;
utarray_push_back(plug_mgr->registered_polling_plugin_array, &polling_plugin_schema);
return (utarray_len(plug_mgr->registered_polling_plugin_array)-1);// return polling plugin_id, equals to polling plugin arrary index + POLLING_PULGIN_ID_BASE
}
int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr)
{
if(plug_mgr==NULL || plug_mgr->registered_polling_plugin_array == NULL)return 0;
struct registered_polling_plugin_schema *p=NULL;
int polling_state=0;
while ((p = (struct registered_polling_plugin_schema *)utarray_next(plug_mgr->registered_polling_plugin_array, p)))
{
if(p->on_polling)
{
if(p->on_polling(p->plugin_env)==1)
{
polling_state=1;
}
}
}
return polling_state;
}
#endif

26
infra/plugin_manager/plugin_manager.h
View File

@@ -1,26 +0,0 @@
#pragma once
#include "stellar/stellar.h"
#ifdef __cplusplus
extern "C"
{
#endif
struct plugin_manager_schema;
struct plugin_manager_runtime;
struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char *plugin_spec_file_path);
void plugin_manager_exit(struct plugin_manager_schema *plug_mgr);
//TODO
void *plugin_manager_get_plugin_env(const char *plugin_name);
//void plugin_manager_on_packet_input(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
//void plugin_manager_on_packet_output(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
//return polling work state, 0: idle, 1: working
//int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr);
#ifdef __cplusplus
}
#endif

33
infra/plugin_manager/plugin_manager_interna.h
View File

@@ -1,33 +0,0 @@
#pragma once
#ifdef __cplusplus
extern "C"
{
#endif
#include "stellar/stellar.h"
#include "uthash/utarray.h"
/*******************************
* PLUGIN MANAGER INIT & EXIT *
*******************************/
struct plugin_manager_schema
{
struct stellar *st;
UT_array *plugin_load_specs_array;
}__attribute__((aligned(sizeof(void*))));
struct plugin_specific
{
char plugin_name[256];
plugin_on_load_func *load_cb;
plugin_on_unload_func *unload_cb;
void *plugin_ctx;
}__attribute__((aligned(sizeof(void*))));
#ifdef __cplusplus
}
#endif

106
infra/plugin_manager/test/plugin_manager_gtest_mock.h
View File

@@ -1,106 +0,0 @@
#pragma once
#ifdef __cplusplus
extern "C"
{
#endif
#include "plugin_manager/plugin_manager_interna.h"
#include "stellar/session.h"
#include "tuple.h"
//mock stellar
struct stellar
{
struct plugin_manager_schema *plug_mgr;
};
enum packet_type
{
UNKNOWN,
IPv4,
IPv6,
UDP,
TCP,
TCP_STREAM,
CONTROL,
};
struct packet
{
struct stellar *st;
enum packet_type type;
unsigned char ip_proto;
};
struct session
{
struct exdata_handle *session_exdat_rt;
enum session_type type;
enum session_state state;
int sess_pkt_cnt;
};
struct plugin_manager_schema * stellar_get_plugin_manager(struct stellar *st)
{
return st->plug_mgr;
}
int stellar_set_plugin_manger(struct stellar *st, struct plugin_manager_schema *pm)
{
st->plug_mgr=pm;
return 0;
}
int stellar_get_worker_thread_num(struct stellar *st __attribute__((unused)))
{
return 16;
}
uint16_t stellar_get_current_thread_index()
{
return 0;
}
unsigned char packet_get_ip_protocol(struct packet *pkt)
{
return pkt->ip_proto;
}
enum session_type session_get_type(const struct session *sess)
{
return sess->type;
}
void session_set_user_data(struct session *sess, void *user_data)
{
sess->session_exdat_rt = (struct exdata_handle *)user_data;
}
void *session_get_user_data(const struct session *sess)
{
return sess->session_exdat_rt;
}
void *packet_get_user_data(const struct packet *pkt)
{
return pkt->st->plug_mgr;
}
int packet_get_innermost_tuple6(const struct packet *pkt, struct tuple6 *tuple)
{
tuple->ip_proto = pkt->ip_proto;
return 0;
}
uint8_t packet_is_ctrl(const struct packet *pkt __attribute__((unused)))
{
return 0;
}
#ifdef __cplusplus
}
#endif

35
infra/stellar_core.c
View File

@@ -8,13 +8,14 @@
#include <pthread.h>
#include <sys/prctl.h>
#include "stellar/module_manager.h"
#include "utils.h"
#include "packet_io.h"
#include "log_private.h"
#include "stellar_stat.h"
#include "stellar_core.h"
#include "packet_private.h"
#include "plugin_manager.h"
#include "session_private.h"
#include "session_manager.h"
@@ -51,7 +52,8 @@ struct stellar_runtime
struct logger *logger;
struct stellar_stat *stat;
struct packet_io *packet_io;
struct plugin_manager_schema *plug_mgr;
struct mq_schema *mq_schema;
struct stellar_module_manager *mod_mgr;
struct stellar_thread threads[MAX_THREAD_NUM];
};
@@ -130,7 +132,11 @@ static void *worker_thread(void *arg)
struct stellar *st = thread->st;
struct stellar_runtime *runtime = &st->runtime;
struct packet_io *packet_io = runtime->packet_io;
struct plugin_manager_schema *plug_mgr = runtime->plug_mgr;
struct stellar_module_manager *mod_mgr = runtime->mod_mgr;
struct mq_runtime *mq_rt = mq_runtime_new(runtime->mq_schema);
stellar_module_manager_register_thread(mod_mgr, thread->tid, mq_rt);
struct thread_stat thr_stat = {
.pkt_io = packet_io_stat(packet_io, thread->idx),
.ip_reass = ip_reassembly_stat(ip_reass),
@@ -145,7 +151,7 @@ static void *worker_thread(void *arg)
for (int i = 0; i < RX_BURST_MAX; i++)
{
packet_set_user_data(&packets[i], (void *)plug_mgr);
packet_set_user_data(&packets[i], (void *)mod_mgr);
}
snprintf(thd_name, sizeof(thd_name), "stellar:%d", thr_idx);
@@ -292,6 +298,8 @@ static void *worker_thread(void *arg)
stellar_stat_merge(runtime->stat, &thr_stat, thread->idx, UINT64_MAX);
stellar_stat_print(runtime->stat, &thr_stat, thread->idx);
mq_runtime_free(mq_rt);
ATOMIC_SET(&thread->is_runing, 0);
CORE_LOG_FATAL("worker thread %d exit", thr_idx);
@@ -448,8 +456,9 @@ struct stellar *stellar_new(const char *stellar_cfg_file, const char *plugin_cfg
CORE_LOG_ERROR("unable to create stellar stat");
goto error_out;
}
runtime->plug_mgr = plugin_manager_init(st, plugin_cfg_file);
if (runtime->plug_mgr == NULL)
runtime->mq_schema=mq_schema_new();
runtime->mod_mgr = stellar_module_manager_new(plugin_cfg_file, config->pkt_io_cfg->nr_worker_thread, runtime->mq_schema);
if (runtime->mod_mgr == NULL)
{
CORE_LOG_ERROR("unable to create plugin manager");
goto error_out;
@@ -527,7 +536,8 @@ void stellar_free(struct stellar *st)
struct stellar_config *config = &st->config;
packet_io_free(runtime->packet_io);
plugin_manager_exit(runtime->plug_mgr);
stellar_module_manager_free(runtime->mod_mgr);
mq_schema_free(runtime->mq_schema);
stellar_stat_free(runtime->stat);
session_manager_config_free(config->sess_mgr_cfg);
@@ -564,20 +574,11 @@ void stellar_reload_log_level(struct stellar *st)
* Stellar Utility Function
******************************************************************************/
struct plugin_manager_schema *stellar_get_plugin_manager(const struct stellar *st)
{
return st->runtime.plug_mgr;
}
void stellar_set_plugin_manger(struct stellar *st, struct plugin_manager_schema *plug_mgr)
{
st->runtime.plug_mgr = plug_mgr;
}
// only send user build packet, can't send packet which come from network
void stellar_send_build_packet(struct stellar *st, struct packet *pkt)
{
uint16_t thr_idx = stellar_get_current_thread_index();
uint16_t thr_idx = stellar_module_manager_get_thread_id(st->runtime.mod_mgr);
struct packet_io *packet_io = st->runtime.packet_io;
struct session_manager *sess_mgr = st->runtime.threads[thr_idx].sess_mgr;
session_manager_record_duplicated_packet(sess_mgr, pkt);