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

add session id generator

Browse Source
This commit is contained in:
luwenpeng
2024-01-31 14:45:50 +08:00
parent 2766b3cfeb
commit 97ef872d9a
21 changed files with 385 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
src/stellar/stellar.cpp
View File

@@ -10,17 +10,13 @@
#include <netinet/ether.h>
#include "logo.h"
#include "stellar.h"
#include "marsio.h"
#include "config.h"
#include "packet.h"
#include "timestamp.h"
#include "session_manager.h"
#ifdef STELLAR_GIT_VERSION
static __attribute__((__used__)) const char *__stellar_version = STELLAR_GIT_VERSION;
#else
static __attribute__((__used__)) const char *__stellar_version = "Unknown";
#endif
#include "id_generator.h"
#define STELLAR_LOG_STATE(format, ...) LOG_STATE("stellar", format, ##__VA_ARGS__)
#define STELLAR_LOG_ERROR(format, ...) LOG_ERROR("stellar", format, ##__VA_ARGS__)
@@ -29,8 +25,6 @@ static __attribute__((__used__)) const char *__stellar_version = "Unknown";
#define ATOMIC_SET(x, y) __atomic_store_n(x, y, __ATOMIC_RELAXED)
#define ATOMIC_READ(x) __atomic_load_n(x, __ATOMIC_RELAXED)
#define RX_BURST_MAX 128
struct packet_io
{
struct mr_instance *mr_ins;
@@ -73,9 +67,8 @@ static void __packet_plugin_dispatch_example(const struct packet *pkt)
return;
}
printf("\n");
printf("=> packet dispatch: %p\n", pkt);
printf("<= packet dispatch\n\n");
printf("<= packet dispatch\n");
}
static void __session_plugin_dispatch_example(struct session *sess)
@@ -85,10 +78,9 @@ static void __session_plugin_dispatch_example(struct session *sess)
return;
}
printf("\n");
printf("=> session dispatch: %p\n", sess);
session_dump(sess);
printf("<= session dispatch\n\n");
printf("<= session dispatch\n");
// after session dispatch, we should reset session current packet and direction
session_set0_cur_pkt(sess, NULL);
@@ -103,25 +95,25 @@ static void signal_handler(int signo)
{
if (signo == SIGINT)
{
STELLAR_LOG_DEBUG("recv SIGINT, exit !!!");
STELLAR_LOG_STATE("SIGINT received, exit !!!");
ATOMIC_SET(&stellar_ctx_ptr->need_exit, 1);
}
if (signo == SIGQUIT)
{
STELLAR_LOG_DEBUG("recv SIGQUIT, exit !!!");
STELLAR_LOG_STATE("SIGQUIT received, exit !!!");
ATOMIC_SET(&stellar_ctx_ptr->need_exit, 1);
}
if (signo == SIGTERM)
{
STELLAR_LOG_DEBUG("recv SIGTERM, exit !!!");
STELLAR_LOG_STATE("SIGTERM received, exit !!!");
ATOMIC_SET(&stellar_ctx_ptr->need_exit, 1);
}
if (signo == SIGHUP)
{
STELLAR_LOG_DEBUG("recv SIGHUP, reload log config file !!!");
STELLAR_LOG_STATE("SIGHUP received, reload log level !!!");
log_reload_level(log_config_file);
}
}
@@ -187,7 +179,7 @@ static void *main_loop(void *arg)
ATOMIC_SET(&threads_ctx->is_runing, 1);
thread_set_name("stellar", thd_idx);
STELLAR_LOG_DEBUG("worker thread %d runing\n", thd_idx);
STELLAR_LOG_DEBUG("worker thread %d runing", thd_idx);
while (ATOMIC_READ(&threads_ctx->need_exit) == 0)
{
@@ -238,7 +230,7 @@ static void *main_loop(void *arg)
}
ATOMIC_SET(&threads_ctx->is_runing, 0);
STELLAR_LOG_DEBUG("worker thread %d exit\n", thd_idx);
STELLAR_LOG_DEBUG("worker thread %d stop", thd_idx);
return NULL;
}
@@ -248,7 +240,7 @@ static int thread_context_init(struct stellar_context *ctx)
struct system_config *sys_cfg = &ctx->config.sys_cfg;
struct session_manager_config *sess_mgr_cfg = &ctx->config.sess_mgr_cfg;
for (uint16_t i = 0; i < sys_cfg->nr_threads; i++)
for (uint8_t i = 0; i < sys_cfg->nr_threads; i++)
{
struct thread_context *threads_ctx = &ctx->threads_ctx[i];
threads_ctx->index = i;
@@ -270,19 +262,20 @@ static void thread_context_free(struct stellar_context *ctx)
{
struct system_config *sys_cfg = &ctx->config.sys_cfg;
for (uint16_t i = 0; i < sys_cfg->nr_threads; i++)
for (uint8_t i = 0; i < sys_cfg->nr_threads; i++)
{
struct thread_context *threads_ctx = &ctx->threads_ctx[i];
if (ATOMIC_READ(&threads_ctx->is_runing) == 0)
{
STELLAR_LOG_STATE("wait worker thread %d free context", i);
session_manager_destroy(threads_ctx->sess_mgr);
}
}
}
static int thread_create(struct thread_context threads_ctx[], uint16_t nr_threads)
static int thread_create(struct thread_context threads_ctx[], uint8_t nr_threads)
{
for (uint16_t i = 0; i < nr_threads; i++)
for (uint8_t i = 0; i < nr_threads; i++)
{
struct thread_context *ctx = &threads_ctx[i];
if (pthread_create(&ctx->tid, NULL, main_loop, (void *)ctx) < 0)
@@ -295,14 +288,15 @@ static int thread_create(struct thread_context threads_ctx[], uint16_t nr_thread
return 0;
}
static void thread_destroy(struct thread_context threads_ctx[], uint16_t nr_threads)
static void thread_destroy(struct thread_context threads_ctx[], uint8_t nr_threads)
{
for (uint16_t i = 0; i < nr_threads; i++)
for (uint8_t i = 0; i < nr_threads; i++)
{
struct thread_context *ctx = &threads_ctx[i];
ATOMIC_SET(&ctx->need_exit, 1);
while (ATOMIC_READ(&ctx->is_runing) == 1)
{
STELLAR_LOG_STATE("wait worker thread %d stop", i);
sleep(1);
}
}
@@ -350,7 +344,7 @@ struct packet_io *packet_io_create(struct system_config *sys_cfg)
int opt = 1;
cpu_set_t coremask;
CPU_ZERO(&coremask);
for (int i = 0; i < sys_cfg->nr_threads; i++)
for (uint8_t i = 0; i < sys_cfg->nr_threads; i++)
{
CPU_SET(sys_cfg->cpu_mask[i], &coremask);
}
@@ -397,8 +391,14 @@ int main(int argc, char **argv)
memset(stellar_ctx_ptr, 0, sizeof(struct stellar_context));
timestamp_update();
signal(SIGINT, signal_handler);
signal(SIGQUIT, signal_handler);
signal(SIGTERM, signal_handler);
signal(SIGHUP, signal_handler);
if (log_init(log_config_file) != 0)
{
STELLAR_LOG_ERROR("unable to init log");
return -1;
}
@@ -406,19 +406,21 @@ int main(int argc, char **argv)
if (config_load(&stellar_ctx_ptr->config, stellar_config_file) != 0)
{
STELLAR_LOG_ERROR("unable to load config file");
return -1;
}
config_dump(&stellar_ctx_ptr->config);
struct system_config *sys_cfg = &stellar_ctx_ptr->config.sys_cfg;
uint16_t nr_threads = sys_cfg->nr_threads;
uint8_t nr_threads = sys_cfg->nr_threads;
// TODO init plugin
if (id_generator_init(sys_cfg->device_base, sys_cfg->device_offset) != 0)
{
STELLAR_LOG_ERROR("unable to init id generator");
return -1;
}
signal(SIGINT, signal_handler);
signal(SIGQUIT, signal_handler);
signal(SIGTERM, signal_handler);
signal(SIGHUP, signal_handler);
// TODO load plugin
stellar_ctx_ptr->pkt_io = packet_io_create(sys_cfg);
if (stellar_ctx_ptr->pkt_io == NULL)
@@ -454,6 +456,8 @@ error_out:
// TODO free plugin
id_generator_free();
log_free();
return 0;