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

✨ feat(module manager): Define and implement the polling API

Browse Source
This commit is contained in:
yangwei
2024-09-25 17:44:27 +08:00
parent c550acef84
commit 7291db5969
6 changed files with 102 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
include/stellar/module_manager.h
View File

@@ -37,6 +37,11 @@ int stellar_module_manager_get_max_thread_num(struct stellar_module_manager* mod
const char *stellar_module_manager_get_toml_path(struct stellar_module_manager *mod_mgr);
struct mq_schema *stellar_module_manager_get_mq_schema(struct stellar_module_manager *mod_mgr);
typedef void module_on_polling_func(struct stellar_module_manager* mod_mgr, void *polling_arg);
//return 0 if success, otherwise return -1.
int stellar_module_manager_polling_subscribe(struct stellar_module_manager* mod_mgr, module_on_polling_func on_polling, void *polling_arg);
void stellar_module_manager_polling_dispatch(struct stellar_module_manager *mod_mgr);
void stellar_module_manager_polling_active(struct stellar_module_manager *mod_mgr);
#ifdef __cplusplus
}

5
include/stellar/stellar.h
View File

@@ -18,11 +18,6 @@ typedef void plugin_on_packet_func(struct packet *pkt, void *on_packet_cb_arg);
//return 0 if success, otherwise return -1.
int stellar_raw_packet_subscribe(struct stellar *st, plugin_on_packet_func *on_packet_cb, void *on_packet_cb_arg);
//return on_polling state, 0: idle, 1: working
typedef int plugin_on_polling_func(void *polling_arg);
//return 0 if success, otherwise return -1.
int stellar_polling_subscribe(struct stellar *st, plugin_on_polling_func on_polling, void *polling_arg);
void stellar_emit_datapath_telemetry(struct packet *pkt, const char * module, const char *str);
// only send user build packet, can't send packet which come from network

2
infra/module_manager/CMakeLists.txt
View File

@@ -3,6 +3,6 @@ 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})
target_link_libraries(module_manager PUBLIC toml mq ${CMAKE_DL_LIBS})
add_subdirectory(test)

39
infra/module_manager/module_manager.c
View File

@@ -237,3 +237,42 @@ void stellar_module_set_name(struct stellar_module* mod, const char *name)
memcpy(mod->name, name, MIN(NAME_MAX, strlen(name)));
return;
}
#define TOPIC_POLLING "POLLING"
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wcast-function-type"
static void stellar_module_manager_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 stellar_module_manager *mod_mgr=(struct stellar_module_manager *)dispatch_arg;
module_on_polling_func *polling = (module_on_polling_func *)on_msg_cb;
polling(mod_mgr, on_msg_cb_arg);
}
int stellar_module_manager_polling_subscribe(struct stellar_module_manager* mod_mgr, module_on_polling_func on_polling, void *polling_arg)
{
mod_mgr->schema.polling_topic_id=mq_schema_get_topic_id(mod_mgr->schema.mq_schema, TOPIC_POLLING);
if(mod_mgr->schema.polling_topic_id<0)
{
mod_mgr->schema.polling_topic_id=mq_schema_create_topic(mod_mgr->schema.mq_schema, TOPIC_POLLING, stellar_module_manager_on_polling_dispatch, mod_mgr, NULL, NULL);
}
return mq_schema_subscribe(mod_mgr->schema.mq_schema, mod_mgr->schema.polling_topic_id, (on_msg_cb_func *)on_polling, polling_arg);
}
#pragma GCC diagnostic pop
void stellar_module_manager_polling_active(struct stellar_module_manager *mod_mgr)
{
mq_runtime_publish_message(local_mq_rt, mod_mgr->schema.polling_topic_id, NULL);
}
void stellar_module_manager_polling_dispatch(struct stellar_module_manager *mod_mgr)
{
stellar_module_manager_polling_active(mod_mgr);
mq_runtime_dispatch(local_mq_rt);
mq_runtime_clean(local_mq_rt);
return;
}

1
infra/module_manager/module_manager_interna.h
View File

@@ -40,6 +40,7 @@ struct stellar_module_manager
int load_module_num;
int max_thread_num;
struct mq_schema *mq_schema;
int polling_topic_id;
}schema;
}__attribute__((aligned(sizeof(void*))));

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

@@ -200,6 +200,61 @@ TEST(module_manager, basic_module) {
}
/***********************************
* TEST MODULE MANAGER POLLING API *
***********************************/
struct test_module_polling_env
{
int N_round;
int polling_count;
int polling_active_count;
};
void test_module_on_polling(struct stellar_module_manager* mod_mgr, void *polling_arg)
{
struct test_module_polling_env *env = (struct test_module_polling_env*)polling_arg;
env->polling_count++;
if(env->polling_count%2==0)
{
stellar_module_manager_polling_active(mod_mgr);
env->polling_active_count++;
}
}
TEST(module_manager_polling, basic_polling_module) {
struct mq_schema *mq_schema=mq_schema_new();
struct stellar_module_manager *mod_mgr=stellar_module_manager_new(NULL, 10, mq_schema);
EXPECT_TRUE(mod_mgr!=NULL);
EXPECT_EQ(stellar_module_manager_get_max_thread_num(mod_mgr), 10);
EXPECT_EQ(stellar_module_manager_get_mq_schema(mod_mgr), mq_schema);
struct test_module_polling_env env={};
env.N_round=10;
stellar_module_manager_polling_subscribe(mod_mgr, test_module_on_polling, &env);
struct mq_runtime *mq_rt = mq_runtime_new(mq_schema);
stellar_module_manager_register_thread(mod_mgr, 1, mq_rt);
EXPECT_EQ((long)stellar_module_manager_get_thread_id(mod_mgr), 1);
EXPECT_EQ(stellar_module_manager_get_mq_runtime(mod_mgr), mq_rt);
for(int i=0; i<env.N_round; i++)
{
stellar_module_manager_polling_dispatch(mod_mgr);
}
stellar_module_manager_free(mod_mgr);
EXPECT_EQ(env.polling_count, env.N_round+env.polling_active_count);
}
/**********************************************
* GTEST MAIN *
**********************************************/