diff options
| -rw-r--r-- | include/stellar/module.h | 8 | ||||
| -rw-r--r-- | infra/module_manager/module_manager.c | 48 | ||||
| -rw-r--r-- | infra/module_manager/module_manager_interna.h | 11 | ||||
| -rw-r--r-- | infra/module_manager/test/gtest_module_manager_main.cpp | 6 | ||||
| -rw-r--r-- | infra/monitor/monitor_rpc.c | 2 | ||||
| -rw-r--r-- | infra/packet_manager/packet_manager.c | 2 | ||||
| -rw-r--r-- | infra/session_manager/session_manager.c | 2 |
7 files changed, 33 insertions, 46 deletions
diff --git a/include/stellar/module.h b/include/stellar/module.h index 0f43a4c..db15bf0 100644 --- a/include/stellar/module.h +++ b/include/stellar/module.h @@ -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 } diff --git a/infra/module_manager/module_manager.c b/infra/module_manager/module_manager.c index a1f6abf..1f8b7be 100644 --- a/infra/module_manager/module_manager.c +++ b/infra/module_manager/module_manager.c @@ -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) -{ - 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); -} - -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) +int module_manager_register_polling_node(struct module_manager *mod_mgr, on_polling_callback *on_polling, void *polling_arg) { - if(mod_mgr == NULL)return; - mq_runtime_publish_message(local_mq_rt, mod_mgr->topic_polling_id, NULL); + 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; } - 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; }
\ No newline at end of file diff --git a/infra/module_manager/module_manager_interna.h b/infra/module_manager/module_manager_interna.h index 08b285a..1b27bd7 100644 --- a/infra/module_manager/module_manager_interna.h +++ b/infra/module_manager/module_manager_interna.h @@ -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 } diff --git a/infra/module_manager/test/gtest_module_manager_main.cpp b/infra/module_manager/test/gtest_module_manager_main.cpp index 4087611..32d5f0c 100644 --- a/infra/module_manager/test/gtest_module_manager_main.cpp +++ b/infra/module_manager/test/gtest_module_manager_main.cpp @@ -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); diff --git a/infra/monitor/monitor_rpc.c b/infra/monitor/monitor_rpc.c index 8e2af67..45d11ee 100644 --- a/infra/monitor/monitor_rpc.c +++ b/infra/monitor/monitor_rpc.c @@ -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(); } diff --git a/infra/packet_manager/packet_manager.c b/infra/packet_manager/packet_manager.c index b823a61..57f5f0c 100644 --- a/infra/packet_manager/packet_manager.c +++ b/infra/packet_manager/packet_manager.c @@ -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) diff --git a/infra/session_manager/session_manager.c b/infra/session_manager/session_manager.c index 8444f3b..02da2a6 100644 --- a/infra/session_manager/session_manager.c +++ b/infra/session_manager/session_manager.c @@ -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) |