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

Optimize packet I/O and timeouts

Browse Source 
- Introduce per-thread I/O statistics for packet I/O to reduce performance overhead.
- Implement packet_io_yield() for better thread management during I/O operations.
- Refactor time wheel management:
  - Replace timeouts-based cron tasks with (now_ts - last_ts > timeout) for scheduled tasks.
  - Update the time wheel every 5 ms for improved time management.
This commit is contained in:
luwenpeng
2024-04-18 14:20:28 +08:00
parent 892842c61b
commit 5508454a1b
32 changed files with 377 additions and 540 deletions
Download Patch File Download Diff File Expand all files Collapse all files

252
src/stellar/stellar.cpp
View File

@@ -10,7 +10,6 @@
#include "logo.h"
#include "stat.h"
#include "cron.h"
#include "stellar.h"
#include "config.h"
#include "packet_private.h"
@@ -25,12 +24,13 @@
#define STELLAR_LOG_ERROR(format, ...) LOG_ERROR("stellar", format, ##__VA_ARGS__)
#define STELLAR_LOG_DEBUG(format, ...) LOG_DEBUG("stellar", format, ##__VA_ARGS__)
struct thread_ctx
struct stellar_thread
{
pthread_t tid;
uint16_t idx;
uint64_t is_runing;
struct thread_cron *cron;
uint64_t stat_last_merge_ts;
uint64_t timing_wheel_last_update_ts;
struct ip_reassembly *ip_mgr;
struct session_manager *sess_mgr;
};
@@ -38,25 +38,21 @@ struct thread_ctx
struct stellar_runtime
{
uint64_t need_exit;
struct thread_cron *cron;
uint64_t stat_last_output_ts;
struct stellar_stat *stat;
struct packet_io *packet_io;
struct plugin_manager *plug_mgr;
struct thread_ctx threads[MAX_THREAD_NUM];
struct stellar_thread threads[MAX_THREAD_NUM];
};
struct stellar_runtime __runtime;
struct stellar_runtime __runtime = {0};
struct stellar_runtime *runtime = &__runtime;
struct stellar_config __config;
struct stellar_config __config = {0};
struct stellar_config *config = &__config;
static const char *log_config_file = "./conf/log.toml";
static const char *stellar_config_file = "./conf/stellar.toml";
/******************************************************************************
* util
******************************************************************************/
static void signal_handler(int signo)
{
if (signo == SIGINT)
@@ -84,16 +80,15 @@ static void signal_handler(int signo)
}
}
static void execute_packet_action(struct packet_io *packet_io, struct session *sess, struct packet *pkt, uint16_t thr_idx)
static void update_session_stat(struct session *sess, struct packet *pkt)
{
int is_ctrl = packet_is_ctrl(pkt);
int need_drop = packet_need_drop(pkt);
if (sess != NULL)
if (sess)
{
enum session_stat stat_pkt;
enum session_stat stat_byte;
if (need_drop)
enum session_direction dir = session_get_current_direction(sess);
int is_ctrl = packet_is_ctrl(pkt);
if (packet_need_drop(pkt))
{
stat_pkt = is_ctrl ? STAT_CTRL_PKTS_DROP : STAT_RAW_PKTS_DROP;
stat_byte = is_ctrl ? STAT_CTRL_BYTES_DROP : STAT_RAW_BYTES_DROP;
@@ -104,41 +99,61 @@ static void execute_packet_action(struct packet_io *packet_io, struct session *s
stat_byte = is_ctrl ? STAT_CTRL_BYTES_TX : STAT_RAW_BYTES_TX;
}
session_inc_stat(sess, session_get_current_direction(sess), stat_pkt, 1);
session_inc_stat(sess, session_get_current_direction(sess), stat_byte, packet_get_len(pkt));
session_inc_stat(sess, dir, stat_pkt, 1);
session_inc_stat(sess, dir, stat_byte, packet_get_len(pkt));
session_set_current_packet(sess, NULL);
session_set_current_direction(sess, SESSION_DIRECTION_NONE);
}
}
if (need_drop)
static inline void free_evicted_sessions(struct session_manager *sess_mgr, uint64_t max_free)
{
void *plugin_ctx = NULL;
struct session *sess = NULL;
for (uint64_t i = 0; i < max_free; i++)
{
packet_io_drop(packet_io, thr_idx, pkt, 1);
}
else
{
packet_io_egress(packet_io, thr_idx, pkt, 1);
sess = session_manager_get_evicted_session(sess_mgr);
if (sess)
{
plugin_ctx = session_get_user_data(sess);
plugin_manager_free_ctx(plugin_ctx);
session_manager_free_session(sess_mgr, sess);
}
else
{
break;
}
}
}
/******************************************************************************
* thread
******************************************************************************/
static inline void thread_set_name(const char *thd_symbol, uint16_t thd_idx)
static inline void free_expired_sessions(struct session_manager *sess_mgr, uint64_t max_free, uint64_t now)
{
char thd_name[16];
snprintf(thd_name, sizeof(thd_name), "%s:%d", thd_symbol, thd_idx);
prctl(PR_SET_NAME, (unsigned long long)thd_name, NULL, NULL, NULL);
void *plugin_ctx = NULL;
struct session *sess = NULL;
for (uint64_t i = 0; i < max_free; i++)
{
sess = session_manager_get_expired_session(sess_mgr, now);
if (sess)
{
plugin_ctx = session_get_user_data(sess);
plugin_manager_free_ctx(plugin_ctx);
session_manager_free_session(sess_mgr, sess);
}
else
{
break;
}
}
}
static inline void thread_stat_merge_cron(void *ctx)
static inline void merge_thread_stat(struct stellar_thread *thread, uint64_t now)
{
struct thread_ctx *thr_ctx = (struct thread_ctx *)ctx;
struct thread_stat thr_stat = {
ip_reassembly_get_stat(thr_ctx->ip_mgr),
session_manager_get_stat(thr_ctx->sess_mgr),
packet_io_stat(runtime->packet_io, thread->idx),
ip_reassembly_stat(thread->ip_mgr),
session_manager_stat(thread->sess_mgr),
};
stellar_peek_thr_stat(runtime->stat, &thr_stat, thr_ctx->idx);
stellar_stat_merge(runtime->stat, &thr_stat, thread->idx);
}
static void *work_thread(void *arg)
@@ -146,26 +161,20 @@ static void *work_thread(void *arg)
int nr_recv;
uint64_t now = 0;
uint16_t thr_idx = 0;
void *plugin_ctx;
struct packet *pkt;
char thd_name[16] = {0};
void *plugin_ctx = NULL;
struct packet *pkt = NULL;
struct packet packets[RX_BURST_MAX];
struct session *sess;
struct session *evicted_sess;
struct session *expired_sess;
struct session *sess = NULL;
struct packet_io *packet_io = runtime->packet_io;
struct plugin_manager *plug_mgr = runtime->plug_mgr;
struct thread_ctx *thr_ctx = (struct thread_ctx *)arg;
struct thread_cron *cron = thr_ctx->cron;
struct ip_reassembly *ip_reass = thr_ctx->ip_mgr;
struct session_manager *sess_mgr = thr_ctx->sess_mgr;
thr_idx = thr_ctx->idx;
struct stellar_thread *thread = (struct stellar_thread *)arg;
struct ip_reassembly *ip_reass = thread->ip_mgr;
struct session_manager *sess_mgr = thread->sess_mgr;
thr_idx = thread->idx;
struct cron_task stat_task = {
.callback = thread_stat_merge_cron,
.data = thr_ctx,
.cycle = 2000, // ms
};
thread_cron_add_task(cron, &stat_task);
snprintf(thd_name, sizeof(thd_name), "stellar:%d", thr_idx);
prctl(PR_SET_NAME, (unsigned long long)thd_name, NULL, NULL, NULL);
if (packet_io_init(packet_io, thr_idx) != 0)
{
@@ -173,8 +182,7 @@ static void *work_thread(void *arg)
return NULL;
}
ATOMIC_SET(&thr_ctx->is_runing, 1);
thread_set_name("stellar", thr_idx);
ATOMIC_SET(&thread->is_runing, 1);
STELLAR_LOG_STATE("worker thread %d runing", thr_idx);
while (ATOMIC_READ(&runtime->need_exit) == 0)
@@ -227,39 +235,47 @@ static void *work_thread(void *arg)
plugin_manager_dispatch_session(plug_mgr, sess, pkt);
fast_path:
execute_packet_action(packet_io, sess, pkt, thr_idx);
}
idle_tasks:
// nr_recv packet atmost trigger nr_recv session evict
for (int i = 0; i < nr_recv; i++)
{
evicted_sess = session_manager_get_evicted_session(sess_mgr);
if (evicted_sess)
update_session_stat(sess, pkt);
if (packet_need_drop(pkt))
{
plugin_ctx = session_get_user_data(evicted_sess);
plugin_manager_free_ctx(plugin_ctx);
session_manager_free_session(sess_mgr, evicted_sess);
packet_io_drop(packet_io, thr_idx, pkt, 1);
}
else
{
packet_io_egress(packet_io, thr_idx, pkt, 1);
}
}
while ((expired_sess = session_manager_get_expired_session(sess_mgr, now)))
{
plugin_ctx = session_get_user_data(expired_sess);
plugin_manager_free_ctx(plugin_ctx);
session_manager_free_session(sess_mgr, expired_sess);
}
ip_reassembly_expire(ip_reass, now);
idle_tasks:
// nr_recv packet atmost trigger nr_recv session evicted
free_evicted_sessions(sess_mgr, nr_recv);
thread_cron_run(cron, now);
// per 5 ms, atmost free 8 expired session
if (now - thread->timing_wheel_last_update_ts > 5)
{
free_expired_sessions(sess_mgr, 8, now);
thread->timing_wheel_last_update_ts = now;
}
if (now - thread->stat_last_merge_ts > 1000)
{
merge_thread_stat(thread, now);
thread->stat_last_merge_ts = now;
}
ip_reassembly_expire(ip_reass, now);
// TODO
// plugin_manager_cron();
// poll_non_packet_events();
// packet_io_yield();
if (nr_recv == 0)
{
packet_io_yield(packet_io, thr_idx, 10);
}
}
ATOMIC_SET(&thr_ctx->is_runing, 0);
ATOMIC_SET(&thread->is_runing, 0);
STELLAR_LOG_STATE("worker thread %d exit !!!", thr_idx);
return NULL;
@@ -270,23 +286,19 @@ static int stellar_thread_init(struct stellar_runtime *ctx, uint8_t nr_threads)
uint64_t now = timestamp_get_msec();
for (uint8_t i = 0; i < nr_threads; i++)
{
struct thread_ctx *thr_ctx = &ctx->threads[i];
thr_ctx->idx = i;
thr_ctx->is_runing = 0;
thr_ctx->cron = thread_cron_new(now);
if (thr_ctx->cron == NULL)
{
STELLAR_LOG_ERROR("unable to create thread cron");
return -1;
}
thr_ctx->sess_mgr = session_manager_new(&config->sess_mgr_opts, now);
if (thr_ctx->sess_mgr == NULL)
struct stellar_thread *thread = &ctx->threads[i];
thread->idx = i;
thread->is_runing = 0;
thread->stat_last_merge_ts = now;
thread->timing_wheel_last_update_ts = now;
thread->sess_mgr = session_manager_new(&config->sess_mgr_opts, now);
if (thread->sess_mgr == NULL)
{
STELLAR_LOG_ERROR("unable to create session manager");
return -1;
}
thr_ctx->ip_mgr = ip_reassembly_new(&config->ip_opts);
if (thr_ctx->ip_mgr == NULL)
thread->ip_mgr = ip_reassembly_new(&config->ip_opts);
if (thread->ip_mgr == NULL)
{
STELLAR_LOG_ERROR("unable to create ip reassemble manager");
return -1;
@@ -300,13 +312,12 @@ static void stellar_thread_clean(struct stellar_runtime *ctx, uint8_t nr_threads
{
for (uint8_t i = 0; i < nr_threads; i++)
{
struct thread_ctx *thr_ctx = &ctx->threads[i];
if (ATOMIC_READ(&thr_ctx->is_runing) == 0)
struct stellar_thread *thread = &ctx->threads[i];
if (ATOMIC_READ(&thread->is_runing) == 0)
{
STELLAR_LOG_STATE("wait worker thread %d free context", i);
session_manager_free(thr_ctx->sess_mgr);
ip_reassembly_free(thr_ctx->ip_mgr);
thread_cron_free(thr_ctx->cron);
session_manager_free(thread->sess_mgr);
ip_reassembly_free(thread->ip_mgr);
}
}
}
@@ -315,8 +326,8 @@ static int stellar_thread_run(struct stellar_runtime *ctx, uint8_t nr_threads)
{
for (uint8_t i = 0; i < nr_threads; i++)
{
struct thread_ctx *thr_ctx = &ctx->threads[i];
if (pthread_create(&thr_ctx->tid, NULL, work_thread, (void *)thr_ctx) < 0)
struct stellar_thread *thread = &ctx->threads[i];
if (pthread_create(&thread->tid, NULL, work_thread, (void *)thread) < 0)
{
STELLAR_LOG_ERROR("unable to create worker thread, error %d: %s", errno, strerror(errno));
return -1;
@@ -330,8 +341,8 @@ static void stellar_thread_join(struct stellar_runtime *ctx, uint8_t nr_threads)
{
for (uint8_t i = 0; i < nr_threads; i++)
{
struct thread_ctx *thr_ctx = &ctx->threads[i];
while (ATOMIC_READ(&thr_ctx->is_runing) == 1)
struct stellar_thread *thread = &ctx->threads[i];
while (ATOMIC_READ(&thread->is_runing) == 1)
{
STELLAR_LOG_STATE("wait worker thread %d stop", i);
sleep(1);
@@ -339,28 +350,9 @@ static void stellar_thread_join(struct stellar_runtime *ctx, uint8_t nr_threads)
}
}
/******************************************************************************
* main
******************************************************************************/
static inline void stellar_stat_output_cron(void *ctx)
{
struct stellar_runtime *runtime = (struct stellar_runtime *)ctx;
stellar_peek_io_stat(runtime->stat, packet_io_get_stat(runtime->packet_io));
stellar_stat_output(runtime->stat);
}
int main(int argc, char **argv)
{
uint8_t nr_threads = 0;
struct cron_task stat_task =
{
.callback = stellar_stat_output_cron,
.data = runtime,
.cycle = 2000, // ms
};
memset(runtime, 0, sizeof(struct stellar_runtime));
memset(config, 0, sizeof(struct stellar_config));
timestamp_update();
signal(SIGINT, signal_handler);
@@ -375,12 +367,12 @@ int main(int argc, char **argv)
}
STELLAR_LOG_STATE("start stellar (version: %s)\n %s", __stellar_version, logo_str);
if (stellar_config_load(stellar_config_file, config) != 0)
if (stellar_load_config(stellar_config_file, config) != 0)
{
STELLAR_LOG_ERROR("unable to load config file");
goto error_out;
}
stellar_config_print(config);
stellar_print_config(config);
STELLAR_LOG_DEBUG("sizeof(struct session) = %lu bytes", sizeof(struct session));
nr_threads = config->io_opts.nr_threads;
@@ -390,13 +382,6 @@ int main(int argc, char **argv)
goto error_out;
}
runtime->cron = thread_cron_new(timestamp_get_msec());
if (runtime->cron == NULL)
{
STELLAR_LOG_ERROR("unable to create runtime cron");
goto error_out;
}
runtime->stat = stellar_stat_new(nr_threads);
if (runtime->stat == NULL)
{
@@ -430,12 +415,16 @@ int main(int argc, char **argv)
goto error_out;
}
thread_cron_add_task(runtime->cron, &stat_task);
runtime->stat_last_output_ts = timestamp_get_msec();
while (!ATOMIC_READ(&runtime->need_exit))
{
timestamp_update();
thread_cron_run(runtime->cron, timestamp_get_msec());
usleep(5 * 1000);
if (timestamp_get_msec() - runtime->stat_last_output_ts > 1000)
{
runtime->stat_last_output_ts = timestamp_get_msec();
stellar_stat_output(runtime->stat);
}
usleep(1000); // 1ms
}
error_out:
@@ -444,7 +433,6 @@ error_out:
packet_io_free(runtime->packet_io);
plugin_manager_free(runtime->plug_mgr);
stellar_stat_free(runtime->stat);
thread_cron_free(runtime->cron);
STELLAR_LOG_STATE("stellar exit !!!\n");
log_free();