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

🦄 refactor(module manager): rename polling API

Browse Source
This commit is contained in:
yangwei
2024-11-25 19:23:01 +08:00
parent ef5a65155b
commit 73a8402a09
7 changed files with 33 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
include/stellar/module.h
View File

@@ -44,6 +44,8 @@ struct module_hooks
};
struct module_manager *module_manager_new(struct module_hooks mod_specs[], size_t n_mod, int max_thread_num, const char *toml_path, struct mq_schema *mq_schema, struct logger *logger);
struct module_manager *module_manager_new_with_toml(const char *toml_path, int max_thread_num, struct mq_schema *mq_schema, struct logger *logger);
void module_manager_free(struct module_manager *mod_mgr);
void module_manager_register_thread(struct module_manager *mod_mgr, int thread_id, struct mq_runtime *mq_rt);
@@ -64,9 +66,9 @@ struct logger *module_manager_get_logger(struct module_manager *mod_mgr);
* polling API *
*******************************************/
typedef void module_on_polling_func(struct module_manager *mod_mgr, void *polling_arg);
int module_manager_polling_subscribe(struct module_manager *mod_mgr, module_on_polling_func on_polling, void *polling_arg);
void module_manager_polling_active(struct module_manager *mod_mgr);
typedef void on_polling_callback(struct module_manager *mod_mgr, void *polling_arg);
int module_manager_register_polling_node(struct module_manager *mod_mgr, on_polling_callback *on_polling, void *polling_arg);
void module_manager_polling_dispatch(struct module_manager *mod_mgr);
#ifdef __cplusplus
}

48
infra/module_manager/module_manager.c
View File

@@ -7,6 +7,7 @@
#include <string.h>
#include "toml/toml.h"
#include "uthash/utlist.h"
/*******************************************
* module manager API *
@@ -285,45 +286,24 @@ void module_set_name(struct module* mod, const char *name)
* polling API *
*******************************************/
#define TOPIC_POLLING "polling"
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wcast-function-type"
static void on_polling_dispatch(int topic_id __unused,
void *msg __unused,
on_msg_cb_func* on_msg_cb,
void *on_msg_cb_arg,
void *dispatch_arg)
int module_manager_register_polling_node(struct module_manager *mod_mgr, on_polling_callback *on_polling, void *polling_arg)
{
struct module_manager *mod_mgr=(struct module_manager *)dispatch_arg;
module_on_polling_func *polling = (module_on_polling_func *)on_msg_cb;
polling(mod_mgr, on_msg_cb_arg);
if(mod_mgr == NULL|| on_polling == NULL)return -1;
struct polling_node *node = CALLOC(struct polling_node, 1);
node->on_polling = on_polling;
node->polling_arg = polling_arg;
LL_APPEND(mod_mgr->node_list, node);
return 0;
}
int module_manager_polling_subscribe(struct module_manager *mod_mgr, module_on_polling_func on_polling, void *polling_arg)
{
if(mod_mgr == NULL)return -1;
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->config.mq_schema, TOPIC_POLLING, on_polling_dispatch, mod_mgr, NULL, NULL);
}
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
void module_manager_polling_active(struct module_manager *mod_mgr)
{
if(mod_mgr == NULL)return;
mq_runtime_publish_message(local_mq_rt, mod_mgr->topic_polling_id, NULL);
}
void module_manager_polling_dispatch(struct module_manager *mod_mgr)
{
if(mod_mgr==NULL)return;
module_manager_polling_active(mod_mgr);
mq_runtime_dispatch(local_mq_rt);
struct polling_node *node;
LL_FOREACH(mod_mgr->node_list, node) {
if (node->on_polling) {
node->on_polling(mod_mgr, node->polling_arg);
}
}
return;
}

11
infra/module_manager/module_manager_interna.h
View File

@@ -27,11 +27,18 @@ struct module_descriptor
}__attribute__((aligned(sizeof(void*))));
struct polling_node
{
on_polling_callback *on_polling;
void *polling_arg;
struct polling_node *next;
};
struct module_manager
{
struct module_descriptor *descriptors;
int n_descriptor;
int topic_polling_id;
struct polling_node *node_list;
struct
{
char *toml_path;
@@ -42,9 +49,7 @@ struct module_manager
}__attribute__((aligned(sizeof(void*))));
struct module_manager *module_manager_new_with_toml(const char *toml_path, int max_thread_num, struct mq_schema *mq_schema, struct logger *logger);
void module_manager_polling_dispatch(struct module_manager *mod_mgr);
#ifdef __cplusplus
}

6
infra/module_manager/test/gtest_module_manager_main.cpp
View File

@@ -222,8 +222,8 @@ struct test_module_polling_env
env->polling_count++;
if(env->polling_count%2==0)
{
module_manager_polling_active(mod_mgr);
env->polling_active_count++;
//module_manager_polling_active(mod_mgr);
//env->polling_active_count++;
}
}
@@ -242,7 +242,7 @@ TEST(module_manager, basic_polling_module) {
struct test_module_polling_env env={};
env.N_round=10;
module_manager_polling_subscribe(mod_mgr, test_module_on_polling, &env);
module_manager_register_polling_node(mod_mgr, test_module_on_polling, &env);
struct mq_runtime *mq_rt = mq_runtime_new(mq_schema);
module_manager_register_thread(mod_mgr, 1, mq_rt);

2
infra/monitor/monitor_rpc.c
View File

@@ -105,7 +105,7 @@ void stm_rpc_free(struct monitor_rpc *rpc_ins)
struct monitor_rpc *monitor_rpc_new(struct stellar_monitor *stm, struct module_manager *mod_mgr)
{
module_manager_polling_subscribe(mod_mgr, module_rpc_worker_thread_polling_cb, (void *)stm);
module_manager_register_polling_node(mod_mgr, module_rpc_worker_thread_polling_cb, (void *)stm);
return stm_rpc_new();
}

2
infra/packet_manager/packet_manager.c
View File

@@ -530,7 +530,7 @@ struct module *packet_manager_on_init(struct module_manager *mod_mgr)
{
return NULL;
}
module_manager_polling_subscribe(mod_mgr, on_polling, pkt_mgr);
module_manager_register_polling_node(mod_mgr, on_polling, pkt_mgr);
struct module *pkt_mgr_mod = module_new(PACKET_MANAGER_MODULE_NAME, NULL);
if (pkt_mgr_mod == NULL)

2
infra/session_manager/session_manager.c
View File

@@ -497,7 +497,7 @@ struct module *session_manager_on_init(struct module_manager *mod_mgr)
{
return NULL;
}
module_manager_polling_subscribe(mod_mgr, on_polling, sess_mgr);
module_manager_register_polling_node(mod_mgr, on_polling, sess_mgr);
struct module *sess_mgr_mod = module_new(SESSION_MANAGER_MODULE_NAME, NULL);
if (sess_mgr_mod == NULL)