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

🦄 refactor(module_manager ): new with hooks

Browse Source
This commit is contained in:
yangwei
2024-11-25 19:23:01 +08:00
parent e277641440
commit ef5a65155b
6 changed files with 141 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

217
infra/module_manager/module_manager.c
View File

@@ -6,32 +6,71 @@
#include <assert.h>
#include <string.h>
#include "toml/toml.h"
/*******************************************
* module manager API *
*******************************************/
#include "toml/toml.h"
struct module_manager *module_manager_new(struct module_hooks mod_hooks[], size_t n_mod, int max_thread_num, const char *toml_path, struct mq_schema *mq_schema, struct logger *logger)
{
struct module_manager *mod_mgr = CALLOC(struct module_manager, 1);
mod_mgr->config.max_thread_num=max_thread_num;
mod_mgr->config.mq_schema=mq_schema;
mod_mgr->config.logger=logger;
if(toml_path)mod_mgr->config.toml_path=strdup(toml_path);
struct module_manager *module_manager_new(struct module_specification mod_specs[], size_t n_mod, int max_thread_num, struct mq_schema *mq_schema, struct logger *logger)
if(mod_hooks==NULL || n_mod==0)return mod_mgr;
mod_mgr->descriptors = CALLOC(struct module_descriptor, n_mod);
mod_mgr->n_descriptor = n_mod;
for (size_t i = 0; i < n_mod; i++)
{
mod_mgr->descriptors[i].hooks = mod_hooks[i];
if (mod_hooks[i].on_instance_init_cb)
{
mod_mgr->descriptors[i].mod = mod_hooks[i].on_instance_init_cb(mod_mgr);
if (module_manager_get_module(mod_mgr, mod_mgr->descriptors[i].mod->name) == NULL)
{
mod_mgr->descriptors[i].initialized = true;
}
else
{
fprintf(stderr, "Module %s already exists\n", mod_mgr->descriptors[i].mod->name);
if (mod_hooks[i].on_instance_exit_cb)
mod_hooks[i].on_instance_exit_cb(mod_mgr, mod_mgr->descriptors[i].mod);
}
}
}
return mod_mgr;
}
struct module_manager *module_manager_new_with_toml(const char *toml_path, int max_thread_num, struct mq_schema *mq_schema, struct logger *logger)
{
struct module_manager *mod_mgr = CALLOC(struct module_manager, 1);
mod_mgr->schema.max_thread_num=max_thread_num;
mod_mgr->schema.mq_schema=mq_schema;
mod_mgr->schema.logger=logger;
FILE *fp=fopen(toml_path, "r");
if(!fp)return module_manager_new(NULL, 0, max_thread_num, toml_path, mq_schema, logger);
int mod_num = 0;
toml_table_t *conf = toml_parse_file(fp, NULL, 0);
fclose(fp);
if(conf==NULL)return module_manager_new(NULL, 0, max_thread_num, toml_path, mq_schema, logger);
toml_array_t* mod_array = toml_array_in(conf, "module");
if(mod_array==NULL)
{
toml_free(conf);
return module_manager_new(NULL, 0, max_thread_num, toml_path, mq_schema, logger);
}
if(mod_array==NULL)goto MODULE_SPEC_LOAD_END;
int mod_num = toml_array_nelem(mod_array);
mod_num = toml_array_nelem(mod_array);
mod_mgr->module_specs = CALLOC(struct module_spec_load, mod_num);
struct module_hooks mod_hooks[mod_num];
memset(mod_hooks, 0, sizeof(mod_hooks));
// TODO: store module specific in hash
for (int i = 0; i < mod_num; i++) {
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");
@@ -40,103 +79,71 @@ struct module_manager *module_manager_new(struct module_specification mod_specs[
const char *thread_init_func_name_raw = toml_raw_in(toml_mod, "thread_init");
const char *thread_exit_func_name_raw = toml_raw_in(toml_mod, "thread_exit");
toml_rtos(path_raw, &mod_mgr->module_specs[i].path);
toml_rtos(init_func_name_raw, &mod_mgr->module_specs[i].instance_init_cb_name);
toml_rtos(exit_func_name_raw, &mod_mgr->module_specs[i].instance_exit_cb_name);
toml_rtos(thread_init_func_name_raw, &mod_mgr->module_specs[i].thread_init_cb_name);
toml_rtos(thread_exit_func_name_raw, &mod_mgr->module_specs[i].thread_exit_cb_name);
char *path = NULL;
char *instance_init_cb_name = NULL, *instance_exit_cb_name = NULL, *thread_init_cb_name = NULL,
*thread_exit_cb_name = NULL;
void* handle = dlopen(mod_mgr->module_specs[i].path, RTLD_NOW|RTLD_LAZY|RTLD_GLOBAL);
toml_rtos(path_raw, &path);
toml_rtos(init_func_name_raw, &instance_init_cb_name);
toml_rtos(exit_func_name_raw, &instance_exit_cb_name);
toml_rtos(thread_init_func_name_raw, &thread_init_cb_name);
toml_rtos(thread_exit_func_name_raw, &thread_exit_cb_name);
void* handle = dlopen(path, RTLD_NOW|RTLD_LAZY|RTLD_GLOBAL);
if (!handle) {
fprintf(stderr, "Error loading module %s: %s\n", mod_mgr->module_specs[i].path, dlerror());
fprintf(stderr, "Error loading module %s: %s\n", path, dlerror());
break;
}
if (mod_mgr->module_specs[i].instance_init_cb_name)
if (instance_init_cb_name)
{
mod_mgr->module_specs[i].on_instance_init_cb =
(module_on_instance_init_func *)dlsym(handle, mod_mgr->module_specs[i].instance_init_cb_name);
if (mod_mgr->module_specs[i].on_instance_init_cb)
{
mod_mgr->module_specs[i].mod = mod_mgr->module_specs[i].on_instance_init_cb(mod_mgr);
if (module_manager_get_module(mod_mgr, mod_mgr->module_specs[i].mod->name) == NULL)
{
mod_mgr->module_specs[i].init_succ = true;
}
else
{
fprintf(stderr, "Module %s already exists\n", mod_mgr->module_specs[i].mod->name);
if (mod_mgr->module_specs[i].on_instance_exit_cb)
mod_mgr->module_specs[i].on_instance_exit_cb(mod_mgr, mod_mgr->module_specs[i].mod);
}
}
else
{
fprintf(stderr, "Could not load init function %s: %s\n", mod_mgr->module_specs[i].instance_init_cb_name, dlerror());
}
mod_hooks[i].on_instance_init_cb =
(module_on_instance_init_func *)dlsym(handle, instance_init_cb_name);
//TODO: logger error
FREE(instance_init_cb_name);
}
if (mod_mgr->module_specs[i].instance_exit_cb_name)
if (instance_exit_cb_name)
{
mod_mgr->module_specs[i].on_instance_exit_cb =
(module_on_instance_exit_func *)dlsym(handle, mod_mgr->module_specs[i].instance_exit_cb_name);
if (!mod_mgr->module_specs[i].on_instance_exit_cb)
{
fprintf(stderr, "Could not load exit function %s: %s\n", mod_mgr->module_specs[i].instance_exit_cb_name, dlerror());
}
mod_hooks[i].on_instance_exit_cb =
(module_on_instance_exit_func *)dlsym(handle, instance_exit_cb_name);
FREE(instance_exit_cb_name);
}
if (mod_mgr->module_specs[i].thread_init_cb_name)
if (thread_init_cb_name)
{
mod_mgr->module_specs[i].on_thread_init_cb =
(module_on_thread_init_func *)dlsym(handle, mod_mgr->module_specs[i].thread_init_cb_name);
if (!mod_mgr->module_specs[i].on_thread_init_cb)
{
fprintf(stderr, "Could not load init function %s: %s\n", mod_mgr->module_specs[i].thread_init_cb_name, dlerror());
}
mod_hooks[i].on_thread_init_cb =
(module_on_thread_init_func *)dlsym(handle, thread_init_cb_name);
FREE(thread_init_cb_name);
}
if (mod_mgr->module_specs[i].thread_exit_cb_name)
if (thread_exit_cb_name)
{
mod_mgr->module_specs[i].on_thread_exit_cb =
(module_on_thread_exit_func *)dlsym(handle, mod_mgr->module_specs[i].thread_exit_cb_name);
if (!mod_mgr->module_specs[i].on_thread_exit_cb)
{
fprintf(stderr, "Could not load exit function %s: %s\n", mod_mgr->module_specs[i].thread_exit_cb_name, dlerror());
}
mod_hooks[i].on_thread_exit_cb =
(module_on_thread_exit_func *)dlsym(handle, thread_exit_cb_name);
FREE(thread_exit_cb_name);
}
mod_mgr->load_module_num+=1;
}
MODULE_SPEC_LOAD_END:
assert(mod_mgr->load_module_num==mod_num);
if(conf )toml_free(conf);
if(fp)fclose(fp);
return mod_mgr;
toml_free(conf);
return module_manager_new(mod_hooks, mod_num, max_thread_num, toml_path, mq_schema, logger);
}
void module_manager_free(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return;
if(mod_mgr->module_spec_toml_path)FREE(mod_mgr->module_spec_toml_path);
if (mod_mgr->module_specs)
if(mod_mgr->config.toml_path)FREE(mod_mgr->config.toml_path);
if (mod_mgr->descriptors)
{
for (int i = 0; i < mod_mgr->load_module_num; i++)
for (int i = 0; i < mod_mgr->n_descriptor; i++)
{
if (mod_mgr->module_specs[i].on_instance_exit_cb != NULL &&
mod_mgr->module_specs[i].init_succ)
if (mod_mgr->descriptors[i].hooks.on_instance_exit_cb != NULL &&
mod_mgr->descriptors[i].initialized)
{
mod_mgr->module_specs[i].on_instance_exit_cb(mod_mgr, mod_mgr->module_specs[i].mod);
mod_mgr->descriptors[i].hooks.on_instance_exit_cb(mod_mgr, mod_mgr->descriptors[i].mod);
}
if(mod_mgr->module_specs[i].path)FREE(mod_mgr->module_specs[i].path);
if(mod_mgr->module_specs[i].instance_init_cb_name)FREE(mod_mgr->module_specs[i].instance_init_cb_name);
if(mod_mgr->module_specs[i].instance_exit_cb_name)FREE(mod_mgr->module_specs[i].instance_exit_cb_name);
if(mod_mgr->module_specs[i].thread_init_cb_name)FREE(mod_mgr->module_specs[i].thread_init_cb_name);
if(mod_mgr->module_specs[i].thread_exit_cb_name)FREE(mod_mgr->module_specs[i].thread_exit_cb_name);
}
FREE(mod_mgr->module_specs);
FREE(mod_mgr->descriptors);
}
FREE(mod_mgr);
return;
@@ -145,25 +152,25 @@ void module_manager_free(struct module_manager *mod_mgr)
int module_manager_get_max_thread_num(struct module_manager*mod_mgr)
{
if(mod_mgr==NULL)return -1;
return mod_mgr->schema.max_thread_num;
return mod_mgr->config.max_thread_num;
}
struct mq_schema *module_manager_get_mq_schema(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return NULL;
return mod_mgr->schema.mq_schema;
return mod_mgr->config.mq_schema;
}
struct logger *module_manager_get_logger(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return NULL;
return mod_mgr->schema.logger;
return mod_mgr->config.logger;
}
const char *module_manager_get_toml_path(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return NULL;
return mod_mgr->module_spec_toml_path;
return mod_mgr->config.toml_path;
}
__thread int local_thread_id=-1;
@@ -184,12 +191,12 @@ void module_manager_register_thread(struct module_manager* mod_mgr, int thread_i
local_thread_id=thread_id;
local_mq_rt=mq_rt;
for(int i=0; i<mod_mgr->load_module_num; i++)
for(int i=0; i<mod_mgr->n_descriptor; i++)
{
if(mod_mgr->module_specs[i].mod == NULL)break;
if(mod_mgr->module_specs[i].on_thread_init_cb && mod_mgr->module_specs[i].init_succ)
if(mod_mgr->descriptors[i].mod == NULL)break;
if(mod_mgr->descriptors[i].hooks.on_thread_init_cb && mod_mgr->descriptors[i].initialized)
{
mod_mgr->module_specs[i].on_thread_init_cb(mod_mgr, thread_id, mod_mgr->module_specs[i].mod);
mod_mgr->descriptors[i].hooks.on_thread_init_cb(mod_mgr, thread_id, mod_mgr->descriptors[i].mod);
}
}
return;
@@ -198,12 +205,12 @@ void module_manager_register_thread(struct module_manager* mod_mgr, int thread_i
void module_manager_unregister_thread(struct module_manager *mod_mgr, int thread_id)
{
assert(local_thread_id==thread_id);
for(int i=0; i<mod_mgr->load_module_num; i++)
for(int i=0; i<mod_mgr->n_descriptor; i++)
{
if(mod_mgr->module_specs[i].mod == NULL)break;
if(mod_mgr->module_specs[i].on_thread_exit_cb && mod_mgr->module_specs[i].init_succ)
if(mod_mgr->descriptors[i].mod == NULL)break;
if(mod_mgr->descriptors[i].hooks.on_thread_exit_cb && mod_mgr->descriptors[i].initialized)
{
mod_mgr->module_specs[i].on_thread_exit_cb(mod_mgr, thread_id, mod_mgr->module_specs[i].mod);
mod_mgr->descriptors[i].hooks.on_thread_exit_cb(mod_mgr, thread_id, mod_mgr->descriptors[i].mod);
}
}
local_thread_id=-1;
@@ -214,14 +221,14 @@ void module_manager_unregister_thread(struct module_manager *mod_mgr, int thread
struct module *module_manager_get_module(struct module_manager *mod_mgr, const char *module_name)
{
if(mod_mgr==NULL || module_name == NULL)return NULL;
if (mod_mgr->module_specs)
if (mod_mgr->descriptors)
{
for(int i=0; i<mod_mgr->load_module_num; i++)
for(int i=0; i<mod_mgr->n_descriptor; i++)
{
if(mod_mgr->module_specs[i].mod == NULL)break;
if(strcmp(mod_mgr->module_specs[i].mod->name, module_name)==0 && mod_mgr->module_specs[i].init_succ)
if(mod_mgr->descriptors[i].mod == NULL)break;
if(strcmp(mod_mgr->descriptors[i].mod->name, module_name)==0 && mod_mgr->descriptors[i].initialized)
{
return mod_mgr->module_specs[i].mod;
return mod_mgr->descriptors[i].mod;
}
}
}
@@ -237,7 +244,7 @@ struct module *module_new(const char *name, void *ctx)
{
struct module *mod = CALLOC(struct module, 1);
memcpy(mod->name, name, MIN(NAME_MAX, strlen(name)));
mod->module_ctx=ctx;
mod->ctx=ctx;
return mod;
}
@@ -251,13 +258,13 @@ void module_free(struct module *mod)
void * module_get_ctx(struct module *mod)
{
if(mod==NULL)return NULL;
return mod->module_ctx;
return mod->ctx;
}
void module_set_ctx(struct module *mod, void *ctx)
{
if(mod==NULL)return;
mod->module_ctx=ctx;
mod->ctx=ctx;
return;
}
@@ -299,9 +306,9 @@ int module_manager_polling_subscribe(struct module_manager *mod_mgr, module_on_
mod_mgr->topic_polling_id=mq_schema_get_topic_id(module_manager_get_mq_schema(mod_mgr), TOPIC_POLLING);
if(mod_mgr->topic_polling_id<0)
{
mod_mgr->topic_polling_id=mq_schema_create_topic(mod_mgr->schema.mq_schema, TOPIC_POLLING, on_polling_dispatch, mod_mgr, NULL, NULL);
mod_mgr->topic_polling_id=mq_schema_create_topic(mod_mgr->config.mq_schema, TOPIC_POLLING, on_polling_dispatch, mod_mgr, NULL, NULL);
}
return mq_schema_subscribe(mod_mgr->schema.mq_schema, mod_mgr->topic_polling_id, (on_msg_cb_func *)on_polling, polling_arg);
return mq_schema_subscribe(mod_mgr->config.mq_schema, mod_mgr->topic_polling_id, (on_msg_cb_func *)on_polling, polling_arg);
}
#pragma GCC diagnostic pop
@@ -313,7 +320,7 @@ void module_manager_polling_active(struct module_manager *mod_mgr)
}
void stellar_polling_dispatch(struct module_manager *mod_mgr)
void module_manager_polling_dispatch(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return;
module_manager_polling_active(mod_mgr);