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

packet manager support output fieldstat

Browse Source
This commit is contained in:
luwenpeng
2024-10-31 18:26:21 +08:00
parent 89971d4b0a
commit 159ad20ea5
8 changed files with 244 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

295
infra/packet_manager/packet_manager.c
View File

@@ -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)
{
if (pkt_mgr_rt)
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)
{
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;
struct packet *pkt = (struct packet *)msg;
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;
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;