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

receive packets from mrzcpd; log output supports stderr or file

Browse Source
This commit is contained in:
luwenpeng
2024-01-30 18:07:08 +08:00
parent 7d7cc8e90c
commit 6d1f352bc9
18 changed files with 810 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
ci/travis.sh
View File

@@ -32,6 +32,11 @@ env | sort
: "${NO_RTTI:=OFF}"
: "${COMPILER_IS_GNUCXX:=OFF}"
# Install dependency from YUM
yum install -y mrzcpd
yum install -y numactl-libs # required by mrzcpd
yum install -y libibverbs # required by mrzcpd
if [ $ASAN_OPTION ] && [ -f "/opt/rh/devtoolset-7/enable" ]; then
source /opt/rh/devtoolset-7/enable
fi

1
conf/CMakeLists.txt
View File

@@ -1 +1,2 @@
install(FILES stellar.toml DESTINATION conf COMPONENT Profile)
install(FILES log.toml DESTINATION conf COMPONENT Profile)

4
conf/log.toml Normal file
View File

@@ -0,0 +1,4 @@
[log]
output = file # stderr, file
file = stellar.log
level = DEBUG # TRACE, DEBUG, INFO, WARN, ERROR, FATAL

4
conf/stellar.toml
View File

@@ -1,6 +1,6 @@
[system]
app_symbol = "stellar"
dev_symbol = "eth0"
app_symbol = stellar
dev_symbol = nf_0_fw
nr_threads = 1
cpu_mask = [5, 6, 7, 8, 9, 10, 11, 12]

4
src/log/CMakeLists.txt
View File

@@ -4,4 +4,6 @@
add_library(log log.cpp)
target_include_directories(log PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_link_libraries(log)
target_link_libraries(log toml)
add_subdirectory(test)

308
src/log/log.cpp
View File

@@ -1,3 +1,307 @@
#include "log.h"
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdarg.h>
#include <time.h>
#include <stdlib.h>
// TODO
#include "log.h"
#include "toml.h"
enum log_output
{
LOG_OUTPUT_STDERR,
LOG_OUTPUT_FILE,
};
struct log_config
{
enum log_output output;
enum log_level level;
char log_file[256];
};
struct log_context
{
struct log_config config;
int log_fd;
int log_file_reopen_day;
};
struct log_context g_log_context;
struct log_context *g_log_ctx = &g_log_context;
static unsigned char weekday_str[7][4] = {"Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"};
static unsigned char month_str[12][4] = {"Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"};
static unsigned char level_str[7][6] = {"TRACE", "DEBUG", "INFO", "WARN", "ERROR", "FATAL", "STATE"};
static inline void local_time(struct tm *local)
{
time_t t;
time(&t);
localtime_r(&t, local);
}
static inline enum log_level check_level(const char *level)
{
if (level == NULL)
{
return LOG_NONE;
}
if (strcasecmp(level, "TRACE") == 0)
{
return LOG_TRACE;
}
else if (strcasecmp(level, "DEBUG") == 0)
{
return LOG_DEBUG;
}
else if (strcasecmp(level, "INFO") == 0)
{
return LOG_INFO;
}
else if (strcasecmp(level, "WARN") == 0)
{
return LOG_WARN;
}
else if (strcasecmp(level, "ERROR") == 0)
{
return LOG_ERROR;
}
else if (strcasecmp(level, "FATAL") == 0)
{
return LOG_FATAL;
}
else if (strcasecmp(level, "STATE") == 0)
{
return LOG_STATE;
}
else
{
return LOG_NONE;
}
}
// return 0: success
// return -1: failed
static int parse_config(struct log_config *config, const char *cfg_file)
{
int ret = -1;
FILE *fp = NULL;
char errbuf[200];
const char *ptr;
toml_table_t *log_section = NULL;
toml_table_t *conf_table = NULL;
fp = fopen(cfg_file, "r");
if (fp == NULL)
{
fprintf(stderr, "open config file %s failed, %s\n", cfg_file, strerror(errno));
goto error_out;
}
conf_table = toml_parse_file(fp, errbuf, sizeof(errbuf));
if (conf_table == NULL)
{
fprintf(stderr, "parse config file %s failed, %s\n", cfg_file, errbuf);
goto error_out;
}
log_section = toml_table_in(conf_table, "log");
if (log_section == NULL)
{
fprintf(stderr, "config file %s missing log section\n", cfg_file);
goto error_out;
}
// output
ptr = toml_raw_in(log_section, "output");
if (ptr == NULL)
{
fprintf(stderr, "config file %s missing log.output\n", cfg_file);
goto error_out;
}
if (strcasecmp(ptr, "stderr") == 0)
{
config->output = LOG_OUTPUT_STDERR;
}
else if (strcasecmp(ptr, "file") == 0)
{
config->output = LOG_OUTPUT_FILE;
}
else
{
fprintf(stderr, "config file %s invalid log.output\n", cfg_file);
goto error_out;
}
// file
if (config->output == LOG_OUTPUT_FILE)
{
ptr = toml_raw_in(log_section, "file");
if (ptr == NULL)
{
fprintf(stderr, "config file %s missing log.file\n", cfg_file);
goto error_out;
}
strncpy(config->log_file, ptr, sizeof(config->log_file) - 1);
}
// level
ptr = toml_raw_in(log_section, "level");
if (ptr == NULL)
{
fprintf(stderr, "config file %s missing log.level\n", cfg_file);
goto error_out;
}
config->level = check_level(ptr);
if (config->level == LOG_NONE)
{
fprintf(stderr, "config file %s invalid log.level\n", cfg_file);
goto error_out;
}
ret = 0;
error_out:
if (conf_table)
{
toml_free(conf_table);
}
if (fp)
{
fclose(fp);
}
return ret;
}
// return 0: success
// return -1: failed
static int log_reopen()
{
int new_fd;
int old_fd;
struct tm local;
char buff[512] = {0};
local_time(&local);
snprintf(buff, sizeof(buff), "%s.%d-%02d-%02d", g_log_ctx->config.log_file, local.tm_year + 1900, local.tm_mon + 1, local.tm_mday);
new_fd = open(buff, O_WRONLY | O_APPEND | O_CREAT, 0644);
if (new_fd == -1)
{
fprintf(stderr, "open() \"%s\" failed, %s", buff, strerror(errno));
return -1;
}
g_log_ctx->log_file_reopen_day = local.tm_mday;
old_fd = g_log_ctx->log_fd;
g_log_ctx->log_fd = new_fd;
if (old_fd > 0)
{
close(old_fd);
}
return 0;
}
/******************************************************************************
* Public API
******************************************************************************/
// return 0: success
// return -1: failed
int log_init(const char *config_file)
{
memset(g_log_ctx, 0, sizeof(struct log_context));
if (parse_config(&g_log_ctx->config, config_file) != 0)
{
return -1;
}
if (g_log_context.config.output == LOG_OUTPUT_FILE)
{
if (log_reopen() != 0)
{
return -1;
}
}
return 0;
}
void log_free()
{
if (g_log_ctx->config.output == LOG_OUTPUT_FILE)
{
if (g_log_ctx->log_fd > 0)
{
close(g_log_ctx->log_fd);
g_log_ctx->log_fd = -1;
}
}
}
void log_reload_level(const char *config_file)
{
struct log_config config;
if (parse_config(&config, config_file) != 0)
{
return;
}
g_log_context.config.level = config.level;
}
void log_print(enum log_level level, const char *module, const char *fmt, ...)
{
if (level < g_log_ctx->config.level)
{
return;
}
int nwrite;
char buf[2048] = {0};
char *p = buf;
char *end = buf + sizeof(buf);
va_list args;
struct tm local;
local_time(&local);
// add time
p += snprintf(p, end - p, "%s %s %d %02d:%02d:%02d %d ",
weekday_str[local.tm_wday], month_str[local.tm_mon], local.tm_mday, local.tm_hour, local.tm_min, local.tm_sec, local.tm_year + 1900);
// add level
p += snprintf(p, end - p, "%s ", level_str[level]);
// add module
p += snprintf(p, end - p, "(%s), ", module);
// add content
va_start(args, fmt);
p += vsnprintf(p, end - p, fmt, args);
va_end(args);
// add end of line
p += snprintf(p, end - p, "\n");
if (g_log_ctx->config.output == LOG_OUTPUT_STDERR)
{
fprintf(stderr, "%s", buf);
return;
}
else
{
if (g_log_ctx->log_file_reopen_day != local.tm_mday)
{
log_reopen();
}
do
{
nwrite = write(g_log_ctx->log_fd, buf, p - buf);
} while (nwrite == -1 && errno == EINTR);
}
}

39
src/log/log.h
View File

@@ -6,15 +6,46 @@ extern "C"
{
#endif
#include <stdio.h>
enum log_level
{
LOG_NONE = -1,
LOG_TRACE = 0,
LOG_DEBUG = 1,
LOG_INFO = 2,
LOG_WARN = 3,
LOG_ERROR = 4,
LOG_FATAL = 5,
LOG_STATE = 6,
};
#define LOG_TRACE(module, format, ...) \
log_print(LOG_TRACE, module, format, ##__VA_ARGS__);
#define LOG_DEBUG(module, format, ...) \
fprintf(stderr, "DEBUG (%s), " format "\n", module, ##__VA_ARGS__);
log_print(LOG_DEBUG, module, format, ##__VA_ARGS__);
#define LOG_INFO(module, format, ...) \
log_print(LOG_INFO, module, format, ##__VA_ARGS__);
#define LOG_WARN(module, format, ...) \
log_print(LOG_WARN, module, format, ##__VA_ARGS__);
#define LOG_ERROR(module, format, ...) \
fprintf(stderr, "ERROR (%s), " format "\n", module, ##__VA_ARGS__);
log_print(LOG_ERROR, module, format, ##__VA_ARGS__);
// TODO
#define LOG_FATAL(module, format, ...) \
log_print(LOG_FATAL, module, format, ##__VA_ARGS__);
#define LOG_STATE(module, format, ...) \
log_print(LOG_STATE, module, format, ##__VA_ARGS__);
// return 0: success
// return -1: failed
int log_init(const char *config_file);
void log_free();
void log_reload_level(const char *config_file);
void log_print(enum log_level level, const char *module, const char *fmt, ...);
#ifdef __cpluscplus
}

12
src/log/test/CMakeLists.txt Normal file
View File

@@ -0,0 +1,12 @@
###############################################################################
# gtest
###############################################################################
add_executable(gtest_log gtest_log.cpp)
target_link_libraries(gtest_log log gtest)
include(GoogleTest)
gtest_discover_tests(gtest_log)
file(COPY ./conf/log_file.toml DESTINATION ./conf/)
file(COPY ./conf/log_stderr.toml DESTINATION ./conf/)

4
src/log/test/conf/log_file.toml Normal file
View File

@@ -0,0 +1,4 @@
[log]
output = file # stderr, file
file = stellar.log
level = DEBUG # TRACE, DEBUG, INFO, WARN, ERROR, FATAL

4
src/log/test/conf/log_stderr.toml Normal file
View File

@@ -0,0 +1,4 @@
[log]
output = stderr # stderr, file
file = stellar.log
level = DEBUG # TRACE, DEBUG, INFO, WARN, ERROR, FATAL

106
src/log/test/gtest_log.cpp Normal file
View File

@@ -0,0 +1,106 @@
#include <gtest/gtest.h>
#include "log.h"
#if 1
TEST(LOG, STDERR)
{
char buffer[1024] = {0};
const char *config = "./conf/log_stderr.toml";
snprintf(buffer, sizeof(buffer), "sed -i 's/DEBUG/ERROR/g' %s", config);
EXPECT_TRUE(log_init(config) == 0);
LOG_TRACE("test", "test log 1");
LOG_DEBUG("test", "test log 1");
LOG_INFO("test", "test log 1");
LOG_WARN("test", "test log 1");
LOG_ERROR("test", "test log 1");
LOG_FATAL("test", "test log 1");
LOG_STATE("test", "test log 1");
system(buffer);
log_reload_level(config);
LOG_TRACE("test", "test log 2");
LOG_DEBUG("test", "test log 2");
LOG_INFO("test", "test log 2");
LOG_WARN("test", "test log 2");
LOG_ERROR("test", "test log 2");
LOG_FATAL("test", "test log 2");
LOG_STATE("test", "test log 2");
log_free();
}
#endif
#if 1
TEST(LOG, FILE)
{
char buffer[1024] = {0};
const char *config = "./conf/log_file.toml";
snprintf(buffer, sizeof(buffer), "sed -i 's/DEBUG/ERROR/g' %s", config);
EXPECT_TRUE(log_init(config) == 0);
LOG_TRACE("test", "test log 1");
LOG_DEBUG("test", "test log 1");
LOG_INFO("test", "test log 1");
LOG_WARN("test", "test log 1");
LOG_ERROR("test", "test log 1");
LOG_FATAL("test", "test log 1");
LOG_STATE("test", "test log 1");
system(buffer);
log_reload_level(config);
LOG_TRACE("test", "test log 2");
LOG_DEBUG("test", "test log 2");
LOG_INFO("test", "test log 2");
LOG_WARN("test", "test log 2");
LOG_ERROR("test", "test log 2");
LOG_FATAL("test", "test log 2");
LOG_STATE("test", "test log 2");
log_free();
}
#endif
#if 1
TEST(LOG, REOPEN)
{
char buffer1[1024] = "date \"+%Y-%m-%d\" >> .date.txt";
char buffer2[1024] = "date -s 2099/01/01";
char buffer3[1024] = "cat .date.txt | xargs date -s && rm .date.txt";
const char *config = "./conf/log_file.toml";
system(buffer1); // record current date
EXPECT_TRUE(log_init(config) == 0);
for (int i = 0; i < 3; i++)
{
LOG_TRACE("test", "test log %d", i);
LOG_DEBUG("test", "test log %d", i);
LOG_INFO("test", "test log %d", i);
LOG_WARN("test", "test log %d", i);
LOG_ERROR("test", "test log %d", i);
LOG_FATAL("test", "test log %d", i);
LOG_STATE("test", "test log %d", i);
if (i == 1)
{
system(buffer2); // set date to 2099/01/01
}
else if (i == 2)
{
system(buffer3); // recover date
}
}
log_free();
}
#endif
int main(int argc, char **argv)
{
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

11
src/packet/packet.h
View File

@@ -13,15 +13,8 @@ extern "C"
#define PACKET_MAX_LAYERS 16
#define PACKET_LOG_ERROR(format, ...) LOG_ERROR("packet", format, ##__VA_ARGS__)
#ifndef PACKET_LOG_ERROR
#define PACKET_LOG_ERROR(format, ...) \
fprintf(stderr, "ERROR (packet), " format "\n", ##__VA_ARGS__);
#endif
#define PACKET_LOG_DEBUG(format, ...) LOG_DEBUG("packet", format, ##__VA_ARGS__)
#ifndef PACKET_LOG_DEBUG
#define PACKET_LOG_DEBUG(format, ...) \
fprintf(stderr, "DEBUG (packet), " format "\n", ##__VA_ARGS__);
#endif
#define PACKET_LOG_DEBUG(format, ...) (void)
//#define PACKET_LOG_DEBUG(format, ...) LOG_DEBUG("packet", format, ##__VA_ARGS__)
enum layer_type
{

8
src/session/session_manager.h
View File

@@ -10,15 +10,7 @@ extern "C"
#include "log.h"
#define SESSION_LOG_ERROR(format, ...) LOG_ERROR("session", format, ##__VA_ARGS__)
#ifndef SESSION_LOG_ERROR
#define SESSION_LOG_ERROR(format, ...) \
fprintf(stderr, "ERROR (session), " format "\n", ##__VA_ARGS__);
#endif
#define SESSION_LOG_DEBUG(format, ...) LOG_DEBUG("session", format, ##__VA_ARGS__)
#ifndef SESSION_LOG_DEBUG
#define SESSION_LOG_DEBUG(format, ...) \
fprintf(stderr, "DEBUG (session), " format "\n", ##__VA_ARGS__);
#endif
struct session_manager_config
{

2
src/stellar/CMakeLists.txt
View File

@@ -1,4 +1,4 @@
add_executable(stellar stellar.cpp)
target_link_libraries(stellar session_manager pthread config)
target_link_libraries(stellar session_manager pthread config mrzcpd)
install(TARGETS stellar RUNTIME DESTINATION bin COMPONENT Program)

27
src/stellar/logo.h Normal file
View File

@@ -0,0 +1,27 @@
#ifndef _LOGO_H
#define _LOGO_H
#ifdef __cpluscplus
extern "C"
{
#endif
/* _ _ _
* ___ | |_ ___ | | | | __ _ _ __
* / __| | __| / _ \ | | | | / _` | | '__|
* \__ \ | |_ | __/ | | | | | (_| | | |
* |___/ \__| \___| |_| |_| \__,_| |_|
*/
static const char logo_str[] =
" _ _ _\n"
" ___ | |_ ___ | | | | __ _ _ __\n"
" / __| | __| / _ \\ | | | | / _` | | '__|\n"
" \\__ \\ | |_ | __/ | | | | | (_| | | |\n"
" |___/ \\__| \\___| |_| |_| \\__,_| |_|\n";
#ifdef __cpluscplus
}
#endif
#endif

388
src/stellar/stellar.cpp
View File

@@ -4,28 +4,40 @@
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <sys/prctl.h>
#include <signal.h>
#include <stdlib.h>
#include <sys/prctl.h>
#include <netinet/ether.h>
#include "logo.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
#define STELLAR_LOG_STATE(format, ...) LOG_STATE("stellar", format, ##__VA_ARGS__)
#define STELLAR_LOG_ERROR(format, ...) LOG_ERROR("stellar", format, ##__VA_ARGS__)
#ifndef STELLAR_LOG_ERROR
#define STELLAR_LOG_ERROR(format, ...) \
fprintf(stderr, "ERROR (stellar), " format "\n", ##__VA_ARGS__);
#endif
#define STELLAR_LOG_DEBUG(format, ...) LOG_DEBUG("stellar", format, ##__VA_ARGS__)
#ifndef STELLAR_LOG_DEBUG
#define STELLAR_LOG_DEBUG(format, ...) \
fprintf(stderr, "DEBUG (stellar), " format "\n", ##__VA_ARGS__);
#endif
#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;
struct mr_vdev *mr_dev;
struct mr_sendpath *mr_path;
};
struct thread_context
{
pthread_t tid;
@@ -40,48 +52,19 @@ struct stellar_context
uint64_t need_exit;
struct config config;
struct thread_context thread_ctx[MAX_THREAD_NUM];
} g_stellar_ctx = {
.need_exit = 0,
struct packet_io *pkt_io;
struct thread_context threads_ctx[MAX_THREAD_NUM];
};
static int recv_packet(const char **data)
{
static unsigned char packet_data[] = {
0x5c, 0x5e, 0xab, 0x2a, 0xa2, 0x00, 0x2c, 0x6b, 0xf5, 0x45, 0x88, 0x29, 0x08, 0x00, 0x45, 0x00, 0x00, 0x5c, 0x0b, 0x4d, 0x00, 0x00, 0x3b, 0x29, 0x09, 0xc8,
0xd2, 0x4d, 0x58, 0xa3, 0x3b, 0x42, 0x04, 0x32, 0x60, 0x00, 0x00, 0x00, 0x00, 0x20, 0x06, 0x40, 0x20, 0x01, 0x0d, 0xa8, 0x02, 0x00, 0x90, 0x0e, 0x02, 0x00,
0x5e, 0xfe, 0xd2, 0x4d, 0x58, 0xa3, 0x26, 0x00, 0x14, 0x0e, 0x00, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x17, 0x02, 0x10, 0x58, 0xcd, 0x4c, 0x00, 0x50,
0x81, 0x80, 0x5c, 0x76, 0x00, 0x00, 0x00, 0x00, 0x80, 0x02, 0x20, 0x00, 0xf7, 0x57, 0x00, 0x00, 0x02, 0x04, 0x04, 0xc4, 0x01, 0x03, 0x03, 0x08, 0x01, 0x01,
0x04, 0x02};
struct stellar_context stellar_context;
struct stellar_context *stellar_ctx_ptr = &stellar_context;
*data = (const char *)packet_data;
return sizeof(packet_data);
}
static const char *log_config_file = "./conf/log.toml";
static const char *stellar_config_file = "./conf/stellar.toml";
static void send_packet(const char *data, uint16_t len)
{
}
static void signal_handler(int signo)
{
if (signo == SIGINT)
{
STELLAR_LOG_DEBUG("recv SIGINT, exit !!!");
ATOMIC_SET(&g_stellar_ctx.need_exit, 1);
}
if (signo == SIGQUIT)
{
STELLAR_LOG_DEBUG("recv SIGQUIT, exit !!!");
ATOMIC_SET(&g_stellar_ctx.need_exit, 1);
}
if (signo == SIGTERM)
{
STELLAR_LOG_DEBUG("recv SIGTERM, exit !!!");
ATOMIC_SET(&g_stellar_ctx.need_exit, 1);
}
}
/******************************************************************************
* example
******************************************************************************/
static void __packet_plugin_dispatch_example(const struct packet *pkt)
{
@@ -112,49 +95,150 @@ static void __session_plugin_dispatch_example(struct session *sess)
session_set_cur_dir(sess, SESSION_DIR_NONE);
}
static void *thread_cycle(void *arg)
/******************************************************************************
* util
******************************************************************************/
static void signal_handler(int signo)
{
uint16_t len = 0;
const char *data = NULL;
struct packet pkt;
struct session *sess = NULL;
struct thread_context *thread_ctx = (struct thread_context *)arg;
struct session_manager *sess_mgr = thread_ctx->sess_mgr;
char thd_name[16];
ATOMIC_SET(&thread_ctx->is_runing, 1);
snprintf(thd_name, sizeof(thd_name), "stellar:%d", thread_ctx->index);
prctl(PR_SET_NAME, (unsigned long long)thd_name, NULL, NULL, NULL);
STELLAR_LOG_DEBUG("worker thread %s runing\n", thd_name);
while (ATOMIC_READ(&thread_ctx->need_exit) == 0)
if (signo == SIGINT)
{
len = recv_packet(&data);
if (data == NULL)
STELLAR_LOG_DEBUG("recv SIGINT, exit !!!");
ATOMIC_SET(&stellar_ctx_ptr->need_exit, 1);
}
if (signo == SIGQUIT)
{
STELLAR_LOG_DEBUG("recv SIGQUIT, exit !!!");
ATOMIC_SET(&stellar_ctx_ptr->need_exit, 1);
}
if (signo == SIGTERM)
{
STELLAR_LOG_DEBUG("recv SIGTERM, exit !!!");
ATOMIC_SET(&stellar_ctx_ptr->need_exit, 1);
}
if (signo == SIGHUP)
{
STELLAR_LOG_DEBUG("recv SIGHUP, reload log config file !!!");
log_reload_level(log_config_file);
}
}
// return 0 : not keepalive packet
// return 1 : is keepalive packet
static inline int is_keepalive_packet(const char *data, int len)
{
if (data == NULL || len < (int)(sizeof(struct ethhdr)))
{
return 0;
}
struct ethhdr *eth_hdr = (struct ethhdr *)data;
if (eth_hdr->h_proto == 0xAAAA)
{
return 1;
}
else
{
return 0;
}
}
/******************************************************************************
* thread
******************************************************************************/
static inline void thread_set_name(const char *thd_symbol, uint16_t thd_idx)
{
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);
}
static void *main_loop(void *arg)
{
int n_pkt_recved;
uint16_t len = 0;
const char *data;
struct packet pkt;
struct session *sess;
marsio_buff_t *rx_buff;
marsio_buff_t *rx_buffs[RX_BURST_MAX];
struct thread_context *threads_ctx = (struct thread_context *)arg;
struct session_manager *sess_mgr = threads_ctx->sess_mgr;
struct packet_io *pkt_io = stellar_ctx_ptr->pkt_io;
uint16_t thd_idx = threads_ctx->index;
struct mr_vdev *mr_dev = pkt_io->mr_dev;
struct mr_sendpath *mr_path = pkt_io->mr_path;
struct mr_instance *mr_ins = pkt_io->mr_ins;
struct mr_vdev *vdevs[1] = {
mr_dev,
};
int min_timeout_ms = 10;
if (marsio_thread_init(mr_ins) != 0)
{
STELLAR_LOG_ERROR("unable to init marsio thread");
return NULL;
}
ATOMIC_SET(&threads_ctx->is_runing, 1);
thread_set_name("stellar", thd_idx);
STELLAR_LOG_DEBUG("worker thread %d runing\n", thd_idx);
while (ATOMIC_READ(&threads_ctx->need_exit) == 0)
{
n_pkt_recved = marsio_recv_burst(mr_dev, thd_idx, rx_buffs, RX_BURST_MAX);
if (n_pkt_recved <= 0)
{
goto poll_wait;
}
for (int i = 0; i < n_pkt_recved; i++)
{
rx_buff = rx_buffs[i];
data = marsio_buff_mtod(rx_buff);
len = marsio_buff_datalen(rx_buff);
if (is_keepalive_packet(data, len))
{
marsio_send_burst(mr_path, thd_idx, &rx_buff, 1);
continue;
}
else
{
packet_parse(&pkt, data, len);
__packet_plugin_dispatch_example(&pkt);
sess = session_manager_update_session(sess_mgr, &pkt);
__session_plugin_dispatch_example(sess);
send_packet(data, len);
sess = session_manager_get_evicted_session(sess_mgr);
__session_plugin_dispatch_example(sess);
// TODO
if (1) // action == forward
{
marsio_send_burst(mr_path, thd_idx, &rx_buff, 1);
}
else // action == drop
{
marsio_buff_free(mr_ins, &rx_buff, 1, 0, thd_idx);
}
}
}
poll_wait:
sess = session_manager_get_expired_session(sess_mgr);
__session_plugin_dispatch_example(sess);
usleep(1000); // session_manager_get_expire_interval(sess_mgr); (seconds)
marsio_poll_wait(mr_ins, vdevs, 1, thd_idx, min_timeout_ms);
}
ATOMIC_SET(&thread_ctx->is_runing, 0);
STELLAR_LOG_DEBUG("worker thread %s exit\n", thd_name);
ATOMIC_SET(&threads_ctx->is_runing, 0);
STELLAR_LOG_DEBUG("worker thread %d exit\n", thd_idx);
return NULL;
}
@@ -166,13 +250,13 @@ static int thread_context_init(struct stellar_context *ctx)
for (uint16_t i = 0; i < sys_cfg->nr_threads; i++)
{
struct thread_context *thread_ctx = &ctx->thread_ctx[i];
thread_ctx->index = i;
thread_ctx->need_exit = 0;
thread_ctx->is_runing = 0;
struct thread_context *threads_ctx = &ctx->threads_ctx[i];
threads_ctx->index = i;
threads_ctx->need_exit = 0;
threads_ctx->is_runing = 0;
thread_ctx->sess_mgr = session_manager_create(sess_mgr_cfg);
if (thread_ctx->sess_mgr == NULL)
threads_ctx->sess_mgr = session_manager_create(sess_mgr_cfg);
if (threads_ctx->sess_mgr == NULL)
{
STELLAR_LOG_ERROR("unable to create session manager");
return -1;
@@ -188,22 +272,20 @@ static void thread_context_free(struct stellar_context *ctx)
for (uint16_t i = 0; i < sys_cfg->nr_threads; i++)
{
struct thread_context *thread_ctx = &ctx->thread_ctx[i];
if (ATOMIC_READ(&thread_ctx->is_runing) == 0)
struct thread_context *threads_ctx = &ctx->threads_ctx[i];
if (ATOMIC_READ(&threads_ctx->is_runing) == 0)
{
session_manager_destroy(thread_ctx->sess_mgr);
session_manager_destroy(threads_ctx->sess_mgr);
}
}
}
static int thread_create(struct stellar_context *ctx)
static int thread_create(struct thread_context threads_ctx[], uint16_t nr_threads)
{
struct system_config *sys_cfg = &ctx->config.sys_cfg;
for (uint16_t i = 0; i < sys_cfg->nr_threads; i++)
for (uint16_t i = 0; i < nr_threads; i++)
{
struct thread_context *thread_ctx = &ctx->thread_ctx[i];
if (pthread_create(&thread_ctx->tid, NULL, thread_cycle, (void *)thread_ctx) < 0)
struct thread_context *ctx = &threads_ctx[i];
if (pthread_create(&ctx->tid, NULL, main_loop, (void *)ctx) < 0)
{
STELLAR_LOG_ERROR("unable to create worker thread, error %d: %s", errno, strerror(errno));
return -1;
@@ -213,72 +295,166 @@ static int thread_create(struct stellar_context *ctx)
return 0;
}
static void thread_destroy(struct stellar_context *ctx)
static void thread_destroy(struct thread_context threads_ctx[], uint16_t nr_threads)
{
struct system_config *sys_cfg = &ctx->config.sys_cfg;
for (uint16_t i = 0; i < sys_cfg->nr_threads; i++)
for (uint16_t i = 0; i < nr_threads; i++)
{
struct thread_context *thread_ctx = &ctx->thread_ctx[i];
ATOMIC_SET(&thread_ctx->need_exit, 1);
while (ATOMIC_READ(&thread_ctx->is_runing) == 1)
struct thread_context *ctx = &threads_ctx[i];
ATOMIC_SET(&ctx->need_exit, 1);
while (ATOMIC_READ(&ctx->is_runing) == 1)
{
sleep(1);
}
}
}
int main(int argc, char **argv)
/******************************************************************************
* packet io
******************************************************************************/
void packet_io_destroy(struct packet_io *pkt_io)
{
if (argc != 2)
if (pkt_io == NULL)
{
printf("usage: %s <config file>\n", argv[0]);
return 0;
return;
}
if (config_load(&g_stellar_ctx.config, argv[1]) != 0)
if (pkt_io->mr_path != NULL)
{
marsio_sendpath_destory(pkt_io->mr_path);
pkt_io->mr_path = NULL;
}
if (pkt_io->mr_dev != NULL)
{
marsio_close_device(pkt_io->mr_dev);
pkt_io->mr_dev = NULL;
}
if (pkt_io->mr_ins != NULL)
{
marsio_destory(pkt_io->mr_ins);
pkt_io->mr_ins = NULL;
}
free(pkt_io);
pkt_io = NULL;
}
struct packet_io *packet_io_create(struct system_config *sys_cfg)
{
struct packet_io *pkt_io = (struct packet_io *)calloc(1, sizeof(struct packet_io));
if (pkt_io == NULL)
{
STELLAR_LOG_ERROR("unable to alloc packet io");
return NULL;
}
int opt = 1;
cpu_set_t coremask;
CPU_ZERO(&coremask);
for (int i = 0; i < sys_cfg->nr_threads; i++)
{
CPU_SET(sys_cfg->cpu_mask[i], &coremask);
}
pkt_io->mr_ins = marsio_create();
if (pkt_io->mr_ins == NULL)
{
STELLAR_LOG_ERROR("unable to create marsio instance");
goto error_out;
}
marsio_option_set(pkt_io->mr_ins, MARSIO_OPT_THREAD_MASK_IN_CPUSET, &coremask, sizeof(coremask));
marsio_option_set(pkt_io->mr_ins, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt));
if (marsio_init(pkt_io->mr_ins, sys_cfg->app_symbol) != 0)
{
STELLAR_LOG_ERROR("unable to init marsio instance");
goto error_out;
}
pkt_io->mr_dev = marsio_open_device(pkt_io->mr_ins, sys_cfg->dev_symbol, sys_cfg->nr_threads, sys_cfg->nr_threads);
if (pkt_io->mr_dev == NULL)
{
STELLAR_LOG_ERROR("unable to open marsio device");
goto error_out;
}
pkt_io->mr_path = marsio_sendpath_create_by_vdev(pkt_io->mr_dev);
if (pkt_io->mr_path == NULL)
{
STELLAR_LOG_ERROR("unable to create marsio sendpath");
goto error_out;
}
return pkt_io;
error_out:
packet_io_destroy(pkt_io);
return NULL;
}
/******************************************************************************
* main
******************************************************************************/
int main(int argc, char **argv)
{
memset(stellar_ctx_ptr, 0, sizeof(struct stellar_context));
timestamp_update();
if (log_init(log_config_file) != 0)
{
return -1;
}
config_dump(&g_stellar_ctx.config);
STELLAR_LOG_STATE("Start Stellar (version: %s)\n %s", __stellar_version, logo_str);
// TODO init log
if (config_load(&stellar_ctx_ptr->config, stellar_config_file) != 0)
{
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;
// TODO init plugin
signal(SIGINT, signal_handler);
signal(SIGQUIT, signal_handler);
signal(SIGTERM, signal_handler);
signal(SIGHUP, signal_handler);
timestamp_update();
stellar_ctx_ptr->pkt_io = packet_io_create(sys_cfg);
if (stellar_ctx_ptr->pkt_io == NULL)
{
STELLAR_LOG_ERROR("unable to create packet io");
goto error_out;
}
if (thread_context_init(&g_stellar_ctx) != 0)
if (thread_context_init(stellar_ctx_ptr) != 0)
{
STELLAR_LOG_ERROR("unable to init thread context");
goto error_out;
}
if (thread_create(&g_stellar_ctx) != 0)
if (thread_create(stellar_ctx_ptr->threads_ctx, nr_threads) != 0)
{
STELLAR_LOG_ERROR("unable to create worker thread");
goto error_out;
}
while (!g_stellar_ctx.need_exit)
while (!ATOMIC_READ(&stellar_ctx_ptr->need_exit))
{
timestamp_update();
sleep(1);
}
error_out:
thread_destroy(&g_stellar_ctx);
packet_io_destroy(stellar_ctx_ptr->pkt_io);
thread_context_free(&g_stellar_ctx);
thread_destroy(stellar_ctx_ptr->threads_ctx, nr_threads);
thread_context_free(stellar_ctx_ptr);
// TODO free plugin
// TODO free log
log_free();
return 0;
}

8
vendor/CMakeLists.txt vendored
View File

@@ -23,3 +23,11 @@ add_library(gmock STATIC IMPORTED GLOBAL)
add_dependencies(gmock googletest)
set_property(TARGET gmock PROPERTY IMPORTED_LOCATION ${INSTALL_DIR}/lib/libgmock.a)
set_property(TARGET gmock PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${INSTALL_DIR}/include)
# MRZCPD
set(MRZCPD_LIB_DIR /opt/mrzcpd/lib)
set(MRZCPD_INCLUDE_DIR /opt/mrzcpd/include)
add_library(mrzcpd SHARED IMPORTED GLOBAL)
set_property(TARGET mrzcpd PROPERTY IMPORTED_LOCATION ${MRZCPD_LIB_DIR}/libmarsio.so)
set_property(TARGET mrzcpd PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${MRZCPD_INCLUDE_DIR})