diff options
| -rw-r--r-- | infra/packet_io/packet_io.c | 2 | ||||
| -rw-r--r-- | infra/packet_manager/CMakeLists.txt | 2 | ||||
| -rw-r--r-- | infra/packet_manager/packet_manager.c | 293 | ||||
| -rw-r--r-- | infra/packet_manager/packet_manager.h | 56 | ||||
| -rw-r--r-- | infra/packet_manager/test/gtest_packet_manager.cpp | 16 | ||||
| -rw-r--r-- | infra/session_manager/session_manager.c | 19 | ||||
| -rw-r--r-- | infra/stellar_core.c | 3 | ||||
| -rw-r--r-- | infra/utils_internal.h | 1 |
8 files changed, 243 insertions, 149 deletions
diff --git a/infra/packet_io/packet_io.c b/infra/packet_io/packet_io.c index 902cea5..e0a7e1c 100644 --- a/infra/packet_io/packet_io.c +++ b/infra/packet_io/packet_io.c @@ -5,8 +5,6 @@ #include "packet_internal.h" #include "fieldstat/fieldstat_easy.h" -#define SYNC_STAT_INTERVAL_MS 1000 - struct packet_io_cfg { char mode[64]; diff --git a/infra/packet_manager/CMakeLists.txt b/infra/packet_manager/CMakeLists.txt index a665944..de5b0bb 100644 --- a/infra/packet_manager/CMakeLists.txt +++ b/infra/packet_manager/CMakeLists.txt @@ -12,6 +12,6 @@ target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/uthash target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/logger) target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/include) target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra) -target_link_libraries(packet_manager tuple logger dablooms mq exdata module_manager) +target_link_libraries(packet_manager tuple logger dablooms mq exdata module_manager fieldstat4) add_subdirectory(test)
\ No newline at end of file diff --git a/infra/packet_manager/packet_manager.c b/infra/packet_manager/packet_manager.c index be36103..f132536 100644 --- a/infra/packet_manager/packet_manager.c +++ b/infra/packet_manager/packet_manager.c @@ -3,6 +3,7 @@ #include "utils_internal.h" #include "packet_internal.h" #include "packet_manager.h" +#include "fieldstat/fieldstat_easy.h" #define PACKET_MANAGER_MODULE_NAME "packet_manager_module" @@ -10,7 +11,7 @@ #define PACKET_MANAGER_LOG_FATAL(format, ...) STELLAR_LOG_FATAL(__thread_local_logger, "packet manager", format, ##__VA_ARGS__) #define PACKET_MANAGER_LOG_INFO(format, ...) STELLAR_LOG_INFO(__thread_local_logger, "packet manager", format, ##__VA_ARGS__) -struct packet_manager_runtime +struct packet_manager_rte { enum packet_stage curr_stage; struct packet_queue queue[PACKET_QUEUE_MAX]; @@ -18,26 +19,29 @@ struct packet_manager_runtime void *claim_arg; on_packet_claimed_callback *claim_cb; - struct mq_runtime *mq; + struct mq_runtime *mq_rte; struct packet_manager_stat stat; }; -struct packet_manager_schema +struct packet_manager_sche { - struct exdata_schema *exdata; - struct mq_schema *mq; - int topic_id[PACKET_STAGE_MAX]; + struct exdata_schema *ex_sche; + struct mq_schema *mq_sche; + int pkt_msg_id[PACKET_STAGE_MAX]; }; struct packet_manager { uint16_t thread_num; - struct packet_manager_schema *schema; - struct packet_manager_runtime *runtime[MAX_THREAD_NUM]; + struct packet_manager_sche *sche; + struct packet_manager_rte *rte[MAX_THREAD_NUM]; + + struct fieldstat_easy *fs; + int pkt_mgr_fs_idx[PKT_MGR_STAT_MAX]; }; /****************************************************************************** - * packet manager runtime + * utils ******************************************************************************/ const char *packet_stage_to_str(enum packet_stage stage) @@ -59,58 +63,76 @@ const char *packet_stage_to_str(enum packet_stage stage) } } -void packet_manager_runtime_free(struct packet_manager_runtime *pkt_mgr_rt) +uint64_t packet_manager_stat_get(struct packet_manager_stat *stat, enum pkt_mgr_stat_type type) +{ + switch (type) + { +#define XX(_type, _name, _val) case _type: return stat->_val; + PKT_MGR_STAT_MAP(XX) +#undef XX + default: + return 0; + } +} + +/****************************************************************************** + * packet manager rte + ******************************************************************************/ + +static void packet_manager_rte_free(struct packet_manager_rte *pkt_mgr_rte) { - if (pkt_mgr_rt) + struct packet *pkt = NULL; + + if (pkt_mgr_rte) { for (int i = 0; i < PACKET_QUEUE_MAX; i++) { - struct packet *pkt = NULL; - while ((pkt = TAILQ_FIRST(&pkt_mgr_rt->queue[i]))) + while ((pkt = TAILQ_FIRST(&pkt_mgr_rte->queue[i]))) { - TAILQ_REMOVE(&pkt_mgr_rt->queue[i], pkt, stage_tqe); + TAILQ_REMOVE(&pkt_mgr_rte->queue[i], pkt, stage_tqe); packet_free(pkt); } } + + free(pkt_mgr_rte); + pkt_mgr_rte = NULL; } - free(pkt_mgr_rt); - pkt_mgr_rt = NULL; } -struct packet_manager_runtime *packet_manager_runtime_new(struct mq_runtime *mq_rt) +static struct packet_manager_rte *packet_manager_rte_new(struct mq_runtime *mq_rte) { - struct packet_manager_runtime *pkt_mgr_rt = calloc(1, sizeof(struct packet_manager_runtime)); - if (pkt_mgr_rt == NULL) + struct packet_manager_rte *pkt_mgr_rte = calloc(1, sizeof(struct packet_manager_rte)); + if (pkt_mgr_rte == NULL) { - PACKET_MANAGER_LOG_ERROR("failed to allocate memory for packet_manager_runtime"); + PACKET_MANAGER_LOG_ERROR("failed to allocate memory for packet_manager_rte"); return NULL; } for (int i = 0; i < PACKET_QUEUE_MAX; i++) { - TAILQ_INIT(&pkt_mgr_rt->queue[i]); + TAILQ_INIT(&pkt_mgr_rte->queue[i]); } - pkt_mgr_rt->mq = mq_rt; + pkt_mgr_rte->mq_rte = mq_rte; - return pkt_mgr_rt; + return pkt_mgr_rte; } /****************************************************************************** - * packet manager schema + * packet manager sche ******************************************************************************/ -static void on_packet_stage_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_arg, void *dispatch_arg) +static void on_packet_stage_dispatch(int pkt_msg_id, void *msg, on_msg_cb_func *cb, void *cb_arg, void *dispatch_arg) { assert(msg); assert(dispatch_arg); - struct packet_manager_schema *schema = (struct packet_manager_schema *)dispatch_arg; + enum packet_stage stage = PACKET_STAGE_MAX; struct packet *pkt = (struct packet *)msg; + struct packet_manager_sche *pkt_mgr_sche = (struct packet_manager_sche *)dispatch_arg; - enum packet_stage stage = PACKET_STAGE_MAX; for (int i = 0; i < PACKET_STAGE_MAX; i++) { - if (schema->topic_id[i] == topic_id) + if (pkt_mgr_sche->pkt_msg_id[i] == pkt_msg_id) { stage = i; break; @@ -120,62 +142,62 @@ static void on_packet_stage_dispatch(int topic_id, void *msg, on_msg_cb_func *cb ((on_packet_stage_callback *)(void *)cb)(pkt, stage, cb_arg); } -static void packet_schema_free(struct packet_manager_schema *pkt_mgr_schema) +static void packet_manager_sche_free(struct packet_manager_sche *pkt_mgr_sche) { - if (pkt_mgr_schema) + if (pkt_mgr_sche) { - if (pkt_mgr_schema->mq) + if (pkt_mgr_sche->mq_sche) { for (int i = 0; i < PACKET_STAGE_MAX; i++) { - if (pkt_mgr_schema->topic_id[i] >= 0) + if (pkt_mgr_sche->pkt_msg_id[i] >= 0) { - mq_schema_destroy_topic(pkt_mgr_schema->mq, pkt_mgr_schema->topic_id[i]); + mq_schema_destroy_topic(pkt_mgr_sche->mq_sche, pkt_mgr_sche->pkt_msg_id[i]); } } } - if (pkt_mgr_schema->exdata) + if (pkt_mgr_sche->ex_sche) { - exdata_schema_free(pkt_mgr_schema->exdata); + exdata_schema_free(pkt_mgr_sche->ex_sche); } - free(pkt_mgr_schema); - pkt_mgr_schema = NULL; + free(pkt_mgr_sche); + pkt_mgr_sche = NULL; } } -static struct packet_manager_schema *packet_schema_new(struct mq_schema *mq) +static struct packet_manager_sche *packet_manager_sche_new(struct mq_schema *mq_sche) { - struct packet_manager_schema *pkt_mgr_schema = calloc(1, sizeof(struct packet_manager_schema)); - if (pkt_mgr_schema == NULL) + struct packet_manager_sche *pkt_mgr_sche = calloc(1, sizeof(struct packet_manager_sche)); + if (pkt_mgr_sche == NULL) { PACKET_MANAGER_LOG_ERROR("failed to allocate memory for packet_schema"); return NULL; } - pkt_mgr_schema->exdata = exdata_schema_new(); - if (pkt_mgr_schema->exdata == NULL) + pkt_mgr_sche->mq_sche = mq_sche; + pkt_mgr_sche->ex_sche = exdata_schema_new(); + if (pkt_mgr_sche->ex_sche == NULL) { PACKET_MANAGER_LOG_ERROR("failed to create exdata_schema"); goto error_out; } - pkt_mgr_schema->mq = mq; for (int i = 0; i < PACKET_STAGE_MAX; i++) { - pkt_mgr_schema->topic_id[i] = mq_schema_create_topic(pkt_mgr_schema->mq, packet_stage_to_str(i), &on_packet_stage_dispatch, pkt_mgr_schema, NULL, NULL); - if (pkt_mgr_schema->topic_id[i] < 0) + pkt_mgr_sche->pkt_msg_id[i] = mq_schema_create_topic(pkt_mgr_sche->mq_sche, packet_stage_to_str(i), &on_packet_stage_dispatch, pkt_mgr_sche, NULL, NULL); + if (pkt_mgr_sche->pkt_msg_id[i] < 0) { PACKET_MANAGER_LOG_ERROR("failed to create topic %s", packet_stage_to_str(i)); goto error_out; } } - return pkt_mgr_schema; + return pkt_mgr_sche; error_out: - packet_schema_free(pkt_mgr_schema); + packet_manager_sche_free(pkt_mgr_sche); return NULL; } @@ -183,7 +205,7 @@ error_out: * packet manager ******************************************************************************/ -struct packet_manager *packet_manager_new(struct mq_schema *mq_schema, uint16_t thread_num) +struct packet_manager *packet_manager_new(struct mq_schema *mq_sche, uint16_t thread_num) { struct packet_manager *pkt_mgr = calloc(1, sizeof(struct packet_manager)); if (pkt_mgr == NULL) @@ -193,13 +215,29 @@ struct packet_manager *packet_manager_new(struct mq_schema *mq_schema, uint16_t } pkt_mgr->thread_num = thread_num; - pkt_mgr->schema = packet_schema_new(mq_schema); - if (pkt_mgr->schema == NULL) + pkt_mgr->sche = packet_manager_sche_new(mq_sche); + if (pkt_mgr->sche == NULL) { PACKET_MANAGER_LOG_ERROR("failed to create packet_schema"); goto error_out; } + pkt_mgr->fs = fieldstat_easy_new(pkt_mgr->thread_num, "packet_manager", NULL, 0); + if (pkt_mgr->fs == NULL) + { + PACKET_MANAGER_LOG_ERROR("failed to create fieldstat_easy"); + goto error_out; + } + if (fieldstat_easy_enable_auto_output(pkt_mgr->fs, "packet_manager.fs4", 2) != 0) + { + PACKET_MANAGER_LOG_ERROR("failed to enable auto output for fieldstat_easy"); + goto error_out; + } + for (int i = 0; i < PKT_MGR_STAT_MAX; i++) + { + pkt_mgr->pkt_mgr_fs_idx[i] = fieldstat_easy_register_counter(pkt_mgr->fs, pkt_mgr_stat_str[i]); + } + return pkt_mgr; error_out: @@ -211,7 +249,12 @@ void packet_manager_free(struct packet_manager *pkt_mgr) { if (pkt_mgr) { - packet_schema_free(pkt_mgr->schema); + if (pkt_mgr->fs) + { + fieldstat_easy_free(pkt_mgr->fs); + } + + packet_manager_sche_free(pkt_mgr->sche); free(pkt_mgr); pkt_mgr = NULL; @@ -221,30 +264,29 @@ void packet_manager_free(struct packet_manager *pkt_mgr) int packet_manager_new_packet_exdata_index(struct packet_manager *pkt_mgr, const char *name, exdata_free *func, void *arg) { assert(pkt_mgr); - return exdata_schema_new_index(pkt_mgr->schema->exdata, name, func, arg); + return exdata_schema_new_index(pkt_mgr->sche->ex_sche, name, func, arg); } int packet_manager_subscribe(struct packet_manager *pkt_mgr, enum packet_stage stage, on_packet_stage_callback *cb, void *arg) { assert(pkt_mgr); - return mq_schema_subscribe(pkt_mgr->schema->mq, pkt_mgr->schema->topic_id[stage], (on_msg_cb_func *)(void *)cb, arg); + return mq_schema_subscribe(pkt_mgr->sche->mq_sche, pkt_mgr->sche->pkt_msg_id[stage], (on_msg_cb_func *)(void *)cb, arg); } -int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rt) +int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rte) { assert(pkt_mgr); assert(thread_id < pkt_mgr->thread_num); - assert(mq_rt); + assert(mq_rte); - struct packet_manager_runtime *pkt_mgr_rt = packet_manager_runtime_new(mq_rt); - if (pkt_mgr_rt == NULL) + pkt_mgr->rte[thread_id] = packet_manager_rte_new(mq_rte); + if (pkt_mgr->rte[thread_id] == NULL) { - PACKET_MANAGER_LOG_ERROR("failed to create packet_manager_runtime"); + PACKET_MANAGER_LOG_ERROR("failed to create packet_manager_rte"); return -1; } else { - pkt_mgr->runtime[thread_id] = pkt_mgr_rt; return 0; } } @@ -254,38 +296,35 @@ void packet_manager_clean(struct packet_manager *pkt_mgr, uint16_t thread_id) assert(pkt_mgr); assert(thread_id < pkt_mgr->thread_num); - struct packet_manager_runtime *pkt_mgr_rt = pkt_mgr->runtime[thread_id]; - PACKET_MANAGER_LOG_INFO("runtime: %p, idx: %d, will be cleaned", pkt_mgr_rt, thread_id); - packet_manager_print_stat(pkt_mgr, thread_id); - packet_manager_runtime_free(pkt_mgr_rt); - pkt_mgr_rt = NULL; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; + packet_manager_rte_free(pkt_mgr_rte); + pkt_mgr_rte = NULL; } void packet_manager_ingress(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt) { - struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id]; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; + struct exdata_runtime *ex_rte = exdata_runtime_new(pkt_mgr->sche->ex_sche); + packet_set_user_data(pkt, ex_rte); - struct exdata_runtime *exdata_rt = exdata_runtime_new(pkt_mgr->schema->exdata); - packet_set_user_data(pkt, exdata_rt); - - runtime->stat.total.pkts_ingress++; - runtime->stat.queue[PACKET_STAGE_PREROUTING].pkts_in++; - TAILQ_INSERT_TAIL(&runtime->queue[PACKET_STAGE_PREROUTING], pkt, stage_tqe); + pkt_mgr_rte->stat.pkts_ingress++; + pkt_mgr_rte->stat.queue[PACKET_STAGE_PREROUTING].pkts_in++; + TAILQ_INSERT_TAIL(&pkt_mgr_rte->queue[PACKET_STAGE_PREROUTING], pkt, stage_tqe); } struct packet *packet_manager_egress(struct packet_manager *pkt_mgr, uint16_t thread_id) { - struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id]; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; + struct packet *pkt = TAILQ_FIRST(&pkt_mgr_rte->queue[PACKET_STAGE_MAX]); - struct packet *pkt = TAILQ_FIRST(&runtime->queue[PACKET_STAGE_MAX]); if (pkt) { - runtime->stat.total.pkts_egress++; - runtime->stat.queue[PACKET_STAGE_MAX].pkts_out++; - TAILQ_REMOVE(&runtime->queue[PACKET_STAGE_MAX], pkt, stage_tqe); + pkt_mgr_rte->stat.pkts_egress++; + pkt_mgr_rte->stat.queue[PACKET_STAGE_MAX].pkts_out++; + TAILQ_REMOVE(&pkt_mgr_rte->queue[PACKET_STAGE_MAX], pkt, stage_tqe); - struct exdata_runtime *exdata_rt = packet_get_user_data(pkt); - exdata_runtime_free(exdata_rt); + struct exdata_runtime *ex_rte = packet_get_user_data(pkt); + exdata_runtime_free(ex_rte); return pkt; } else @@ -296,45 +335,45 @@ struct packet *packet_manager_egress(struct packet_manager *pkt_mgr, uint16_t th void packet_manager_dispatch(struct packet_manager *pkt_mgr, uint16_t thread_id) { - struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id]; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; for (int i = 0; i < PACKET_STAGE_MAX; i++) { - runtime->curr_stage = i; + pkt_mgr_rte->curr_stage = i; struct packet *pkt = NULL; - while ((pkt = TAILQ_FIRST(&runtime->queue[runtime->curr_stage]))) + while ((pkt = TAILQ_FIRST(&pkt_mgr_rte->queue[pkt_mgr_rte->curr_stage]))) { packet_set_claim(pkt, false); - runtime->claim_cb = NULL; - runtime->claim_arg = NULL; + pkt_mgr_rte->claim_cb = NULL; + pkt_mgr_rte->claim_arg = NULL; - TAILQ_REMOVE(&runtime->queue[runtime->curr_stage], pkt, stage_tqe); - runtime->stat.queue[runtime->curr_stage].pkts_out++; + TAILQ_REMOVE(&pkt_mgr_rte->queue[pkt_mgr_rte->curr_stage], pkt, stage_tqe); + pkt_mgr_rte->stat.queue[pkt_mgr_rte->curr_stage].pkts_out++; - mq_runtime_publish_message(runtime->mq, runtime->curr_stage, pkt); - mq_runtime_dispatch(runtime->mq); + mq_runtime_publish_message(pkt_mgr_rte->mq_rte, pkt_mgr_rte->curr_stage, pkt); + mq_runtime_dispatch(pkt_mgr_rte->mq_rte); if (packet_is_claim(pkt)) { - if (runtime->claim_cb) + if (pkt_mgr_rte->claim_cb) { - runtime->claim_cb(pkt, runtime->claim_arg); + pkt_mgr_rte->claim_cb(pkt, pkt_mgr_rte->claim_arg); } continue; } - TAILQ_INSERT_TAIL(&runtime->queue[runtime->curr_stage + 1], pkt, stage_tqe); - runtime->stat.queue[runtime->curr_stage + 1].pkts_in++; + TAILQ_INSERT_TAIL(&pkt_mgr_rte->queue[pkt_mgr_rte->curr_stage + 1], pkt, stage_tqe); + pkt_mgr_rte->stat.queue[pkt_mgr_rte->curr_stage + 1].pkts_in++; } } - runtime->curr_stage = -1; + pkt_mgr_rte->curr_stage = -1; } int packet_manager_claim_packet(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt, on_packet_claimed_callback cb, void *arg) { assert(pkt_mgr); - struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id]; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; if (packet_is_claim(pkt)) { @@ -343,10 +382,10 @@ int packet_manager_claim_packet(struct packet_manager *pkt_mgr, uint16_t thread_ } else { - runtime->claim_cb = cb; - runtime->claim_arg = arg; + pkt_mgr_rte->claim_cb = cb; + pkt_mgr_rte->claim_arg = arg; packet_set_claim(pkt, true); - runtime->stat.queue[runtime->curr_stage].pkts_claim++; + pkt_mgr_rte->stat.queue[pkt_mgr_rte->curr_stage].pkts_claim++; return 0; } } @@ -354,7 +393,7 @@ int packet_manager_claim_packet(struct packet_manager *pkt_mgr, uint16_t thread_ void packet_manager_schedule_packet(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt, enum packet_stage stage) { assert(pkt_mgr); - struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id]; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; if (stage >= PACKET_STAGE_MAX) { @@ -363,33 +402,33 @@ void packet_manager_schedule_packet(struct packet_manager *pkt_mgr, uint16_t thr return; } - runtime->stat.queue[stage].pkts_schedule++; - runtime->stat.queue[stage].pkts_in++; - TAILQ_INSERT_TAIL(&runtime->queue[stage], pkt, stage_tqe); + pkt_mgr_rte->stat.queue[stage].pkts_schedule++; + pkt_mgr_rte->stat.queue[stage].pkts_in++; + TAILQ_INSERT_TAIL(&pkt_mgr_rte->queue[stage], pkt, stage_tqe); } struct packet_manager_stat *packet_manager_get_stat(struct packet_manager *pkt_mgr, uint16_t thread_id) { - struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id]; - return &runtime->stat; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; + return &pkt_mgr_rte->stat; } void packet_manager_print_stat(struct packet_manager *pkt_mgr, uint16_t thread_id) { - struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id]; + struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id]; PACKET_MANAGER_LOG_INFO("runtime: %p, pkts_ingress: %lu, pkts_egress: %lu", - runtime, runtime->stat.total.pkts_ingress, - runtime->stat.total.pkts_egress); + pkt_mgr_rte, pkt_mgr_rte->stat.pkts_ingress, + pkt_mgr_rte->stat.pkts_egress); for (int i = 0; i < PACKET_QUEUE_MAX; i++) { PACKET_MANAGER_LOG_INFO("runtime: %p, %-24s stat => pkts_in: %lu, pkts_out: %lu, pkts_claim: %lu, pkts_schedule: %lu", - runtime, + pkt_mgr_rte, packet_stage_to_str(i), - runtime->stat.queue[i].pkts_in, - runtime->stat.queue[i].pkts_out, - runtime->stat.queue[i].pkts_claim, - runtime->stat.queue[i].pkts_schedule); + pkt_mgr_rte->stat.queue[i].pkts_in, + pkt_mgr_rte->stat.queue[i].pkts_out, + pkt_mgr_rte->stat.queue[i].pkts_claim, + pkt_mgr_rte->stat.queue[i].pkts_schedule); } } @@ -397,6 +436,27 @@ void packet_manager_print_stat(struct packet_manager *pkt_mgr, uint16_t thread_i * packet manager module ******************************************************************************/ +static void on_polling(struct stellar_module_manager *mod_mgr, void *args) +{ + uint64_t now_ms = clock_get_real_time_ms(); + int thread_id = stellar_module_manager_get_thread_id(mod_mgr); + struct packet_manager *pkt_mgr = (struct packet_manager *)args; + + static __thread uint64_t last_sync_stat_ms = 0; + static __thread struct packet_manager_stat pkt_mgr_last_stat = {0}; + if (now_ms - last_sync_stat_ms >= SYNC_STAT_INTERVAL_MS) + { + struct packet_manager_stat *pkt_mgr_curr_stat = &pkt_mgr->rte[thread_id]->stat; + for (int i = 0; i < PKT_MGR_STAT_MAX; i++) + { + uint64_t val = packet_manager_stat_get(pkt_mgr_curr_stat, i) - packet_manager_stat_get(&pkt_mgr_last_stat, i); + fieldstat_easy_counter_incrby(pkt_mgr->fs, thread_id, pkt_mgr->pkt_mgr_fs_idx[i], NULL, 0, val); + } + pkt_mgr_last_stat = *pkt_mgr_curr_stat; + last_sync_stat_ms = now_ms; + } +} + struct packet_manager *stellar_module_get_packet_manager(struct stellar_module_manager *mod_mgr) { assert(mod_mgr); @@ -411,15 +471,16 @@ struct packet_manager *stellar_module_get_packet_manager(struct stellar_module_m struct stellar_module *packet_manager_on_init(struct stellar_module_manager *mod_mgr) { assert(mod_mgr); - struct mq_schema *mq_schema = stellar_module_manager_get_mq_schema(mod_mgr); - assert(mq_schema); + struct mq_schema *mq_sche = stellar_module_manager_get_mq_schema(mod_mgr); + assert(mq_sche); uint16_t thread_num = stellar_module_manager_get_max_thread_num(mod_mgr); - struct packet_manager *pkt_mgr = packet_manager_new(mq_schema, thread_num); + struct packet_manager *pkt_mgr = packet_manager_new(mq_sche, thread_num); if (pkt_mgr == NULL) { return NULL; } + stellar_module_manager_polling_subscribe(mod_mgr, on_polling, pkt_mgr); struct stellar_module *pkt_mgr_mod = stellar_module_new(PACKET_MANAGER_MODULE_NAME, NULL); if (pkt_mgr_mod == NULL) @@ -450,11 +511,11 @@ struct stellar_module *packet_manager_on_thread_init(struct stellar_module_manag { struct packet_manager *pkt_mgr = stellar_module_get_ctx(mod); assert(pkt_mgr); - struct mq_runtime *mq_rt = stellar_module_manager_get_mq_runtime(mod_mgr); - assert(mq_rt); + struct mq_runtime *mq_rte = stellar_module_manager_get_mq_runtime(mod_mgr); + assert(mq_rte); assert(thread_id < pkt_mgr->thread_num); - if (packet_manager_init(pkt_mgr, thread_id, mq_rt) != 0) + if (packet_manager_init(pkt_mgr, thread_id, mq_rte) != 0) { PACKET_MANAGER_LOG_ERROR("failed to init packet_manager_init"); return NULL; diff --git a/infra/packet_manager/packet_manager.h b/infra/packet_manager/packet_manager.h index 7821758..61d5f04 100644 --- a/infra/packet_manager/packet_manager.h +++ b/infra/packet_manager/packet_manager.h @@ -12,11 +12,8 @@ extern "C" struct packet_manager_stat { - struct - { - uint64_t pkts_ingress; - uint64_t pkts_egress; - } total; + uint64_t pkts_ingress; + uint64_t pkts_egress; struct { uint64_t pkts_in; // include the packets that are scheduled @@ -26,10 +23,55 @@ struct packet_manager_stat } queue[PACKET_QUEUE_MAX]; // the last queue is for sending packets }; +// XX(type, name, val) +#define PKT_MGR_STAT_MAP(XX) \ + XX(PKT_MGR_STAT_PKTS_INGRESS, pkts_ingress, pkts_ingress) \ + XX(PKT_MGR_STAT_PKTS_EGRESS, pkts_egress, pkts_egress) \ + /* PREROUTING */ \ + XX(PKT_MGR_STAT_PKTS_IN_ON_PREROUTING, pkts_in_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_in) \ + XX(PKT_MGR_STAT_PKTS_OUT_ON_PREROUTING, pkts_out_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_out) \ + XX(PKT_MGR_STAT_PKTS_CLAIM_ON_PREROUTING, pkts_claim_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_claim) \ + XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_PREROUTING, pkts_schedule_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_schedule) \ + /* INPUT */ \ + XX(PKT_MGR_STAT_PKTS_IN_ON_INPUT, pkts_in_on_input, queue[PACKET_STAGE_INPUT].pkts_in) \ + XX(PKT_MGR_STAT_PKTS_OUT_ON_INPUT, pkts_out_on_input, queue[PACKET_STAGE_INPUT].pkts_out) \ + XX(PKT_MGR_STAT_PKTS_CLAIM_ON_INPUT, pkts_claim_on_input, queue[PACKET_STAGE_INPUT].pkts_claim) \ + XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_INPUT, pkts_schedule_on_input, queue[PACKET_STAGE_INPUT].pkts_schedule) \ + /* FORWARD */ \ + XX(PKT_MGR_STAT_PKTS_IN_ON_FORWARD, pkts_in_on_forward, queue[PACKET_STAGE_FORWARD].pkts_in) \ + XX(PKT_MGR_STAT_PKTS_OUT_ON_FORWARD, pkts_out_on_forward, queue[PACKET_STAGE_FORWARD].pkts_out) \ + XX(PKT_MGR_STAT_PKTS_CLAIM_ON_FORWARD, pkts_claim_on_forward, queue[PACKET_STAGE_FORWARD].pkts_claim) \ + XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_FORWARD, pkts_schedule_on_forward, queue[PACKET_STAGE_FORWARD].pkts_schedule) \ + /* OUTPUT */ \ + XX(PKT_MGR_STAT_PKTS_IN_ON_OUTPUT, pkts_in_on_output, queue[PACKET_STAGE_OUTPUT].pkts_in) \ + XX(PKT_MGR_STAT_PKTS_OUT_ON_OUTPUT, pkts_out_on_output, queue[PACKET_STAGE_OUTPUT].pkts_out) \ + XX(PKT_MGR_STAT_PKTS_CLAIM_ON_OUTPUT, pkts_claim_on_output, queue[PACKET_STAGE_OUTPUT].pkts_claim) \ + XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_OUTPUT, pkts_schedule_on_output, queue[PACKET_STAGE_OUTPUT].pkts_schedule) \ + /* POSTROUTING */ \ + XX(PKT_MGR_STAT_PKTS_IN_ON_POSTROUTING, pkts_in_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_in) \ + XX(PKT_MGR_STAT_PKTS_OUT_ON_POSTROUTING, pkts_out_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_out) \ + XX(PKT_MGR_STAT_PKTS_CLAIM_ON_POSTROUTING, pkts_claim_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_claim) \ + XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_POSTROUTING, pkts_schedule_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_schedule) + +enum pkt_mgr_stat_type +{ +#define XX(type, name, val) type, + PKT_MGR_STAT_MAP(XX) +#undef XX + PKT_MGR_STAT_MAX +}; + +__attribute__((unused)) static const char pkt_mgr_stat_str[PKT_MGR_STAT_MAX][64] = +{ +#define XX(type, name, val) #name, + PKT_MGR_STAT_MAP(XX) +#undef XX +}; + struct packet_manager *packet_manager_new(struct mq_schema *mq_schema, uint16_t thread_num); void packet_manager_free(struct packet_manager *pkt_mgr); -int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rt); +int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rte); void packet_manager_clean(struct packet_manager *pkt_mgr, uint16_t thread_id); void packet_manager_ingress(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt); struct packet *packet_manager_egress(struct packet_manager *pkt_mgr, uint16_t thread_id); @@ -39,6 +81,8 @@ void packet_manager_print_stat(struct packet_manager *pkt_mgr, uint16_t thread_i const char *packet_stage_to_str(enum packet_stage stage); +uint64_t packet_manager_stat_get(struct packet_manager_stat *stat, enum pkt_mgr_stat_type type); + #ifdef __cplusplus } #endif diff --git a/infra/packet_manager/test/gtest_packet_manager.cpp b/infra/packet_manager/test/gtest_packet_manager.cpp index 902b7c7..30abcdb 100644 --- a/infra/packet_manager/test/gtest_packet_manager.cpp +++ b/infra/packet_manager/test/gtest_packet_manager.cpp @@ -71,8 +71,8 @@ static struct packet_manager_stat init_stat = {}; static void check_stat(struct packet_manager_stat *curr_stat, struct packet_manager_stat *expect_stat) { - EXPECT_TRUE(curr_stat->total.pkts_ingress == expect_stat->total.pkts_ingress); - EXPECT_TRUE(curr_stat->total.pkts_egress == expect_stat->total.pkts_egress); + EXPECT_TRUE(curr_stat->pkts_ingress == expect_stat->pkts_ingress); + EXPECT_TRUE(curr_stat->pkts_egress == expect_stat->pkts_egress); for (int i = 0; i < PACKET_QUEUE_MAX; i++) { @@ -143,7 +143,8 @@ TEST(PACKET_MANAGER, SUBSCRIBER_PACKET_STAGE) EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == &pkt); EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == NULL); struct packet_manager_stat expect_stat = { - .total = {.pkts_ingress = 1, .pkts_egress = 1}, + .pkts_ingress = 1, + .pkts_egress = 1, .queue = { [PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 0, .pkts_schedule = 0}, [PACKET_STAGE_INPUT] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 0, .pkts_schedule = 0}, @@ -246,7 +247,8 @@ TEST(PACKET_MANAGER, CLAIM_PACKET) packet_manager_dispatch(pkt_mgr, thread_id); EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == NULL); struct packet_manager_stat expect_stat = { - .total = {.pkts_ingress = 1, .pkts_egress = 0}, + .pkts_ingress = 1, + .pkts_egress = 0, .queue = { [PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 1, .pkts_schedule = 0}, [PACKET_STAGE_INPUT] = {.pkts_in = 0, .pkts_out = 0, .pkts_claim = 0, .pkts_schedule = 0}, @@ -329,7 +331,8 @@ TEST(PACKET_MANAGER, SCHEDULE_PACKET) EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == &pkt); EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == NULL); struct packet_manager_stat expect_stat = { - .total = {.pkts_ingress = 1, .pkts_egress = 5}, + .pkts_ingress = 1, + .pkts_egress = 5, .queue = { [PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 0, .pkts_schedule = 0}, [PACKET_STAGE_INPUT] = {.pkts_in = 2, .pkts_out = 2, .pkts_claim = 0, .pkts_schedule = 1}, @@ -423,7 +426,8 @@ TEST(PACKET_MANAGER, SCHEDULE_CLAIMED_PACKET) packet_manager_dispatch(pkt_mgr, thread_id); EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == &pkt); struct packet_manager_stat expect_stat = { - .total = {.pkts_ingress = 1, .pkts_egress = 1}, + .pkts_ingress = 1, + .pkts_egress = 1, .queue = { [PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 1, .pkts_schedule = 0}, [PACKET_STAGE_INPUT] = {.pkts_in = 0, .pkts_out = 0, .pkts_claim = 0, .pkts_schedule = 0}, diff --git a/infra/session_manager/session_manager.c b/infra/session_manager/session_manager.c index a8a3879..bd602f4 100644 --- a/infra/session_manager/session_manager.c +++ b/infra/session_manager/session_manager.c @@ -8,7 +8,6 @@ #include "session_manager_stat.h" #include "fieldstat/fieldstat_easy.h" -#define SYNC_STAT_INTERVAL_MS 1000 #define CLEAN_SESSION_BURST 1024 #define SESSION_MANAGER_MODULE_NAME "session_manager_module" @@ -516,16 +515,6 @@ void session_manager_clean(struct session_manager *sess_mgr, uint16_t thread_id) mq_runtime_dispatch(mq_rte); } -<<<<<<< HEAD -======= - // flush stat before free - for (int i = 0; i < SESS_MGR_STAT_MAX; i++) - { - uint64_t val = session_manager_stat_get(stat, i); - fieldstat_easy_counter_set(sess_mgr->fs, thread_id, sess_mgr->stat_idx[i], NULL, 0, val); - } - ->>>>>>> 29ffae8 (packet IO support output fieldstat) session_manager_rte_free(sess_mgr->rte[thread_id]); sess_mgr->rte[thread_id] = NULL; } @@ -550,16 +539,12 @@ struct stellar_module *session_manager_on_init(struct stellar_module_manager *mo assert(mod_mgr); struct packet_manager *pkt_mgr = stellar_module_get_packet_manager(mod_mgr); assert(pkt_mgr); - struct mq_schema *mq_schema = stellar_module_manager_get_mq_schema(mod_mgr); - assert(mq_schema); + struct mq_schema *mq_sche = stellar_module_manager_get_mq_schema(mod_mgr); + assert(mq_sche); const char *toml_file = stellar_module_manager_get_toml_path(mod_mgr); assert(toml_file); -<<<<<<< HEAD - struct session_manager *sess_mgr = session_manager_new(mod_mgr, pkt_mgr, mq_schema, toml_file); -======= struct session_manager *sess_mgr = session_manager_new(pkt_mgr, mq_sche, toml_file); ->>>>>>> 6e4d813 (bugfix sess) if (sess_mgr == NULL) { return NULL; diff --git a/infra/stellar_core.c b/infra/stellar_core.c index 16e4f4f..1761021 100644 --- a/infra/stellar_core.c +++ b/infra/stellar_core.c @@ -84,9 +84,10 @@ static void *worker_thread(void *arg) packet_io_send(pkt_io, thread_id, &pkt, 1); } } + packet_io_clean(pkt_io, thread_id); stellar_polling_dispatch(mod_mgr); - packet_io_clean(pkt_io, thread_id); + if (nr_recv == 0) { packet_io_yield(pkt_io, thread_id); diff --git a/infra/utils_internal.h b/infra/utils_internal.h index fba80f8..20ff0a2 100644 --- a/infra/utils_internal.h +++ b/infra/utils_internal.h @@ -17,6 +17,7 @@ extern "C" #define RX_BURST_MAX 32 #define MAX_THREAD_NUM 256 // limit by snowflake +#define SYNC_STAT_INTERVAL_MS 1000 #define ATOMIC_INC(x) __atomic_fetch_add(x, 1, __ATOMIC_RELAXED) #define ATOMIC_DEC(x) __atomic_fetch_sub(x, 1, __ATOMIC_RELAXED) |