diff --git a/ci/travis.sh b/ci/travis.sh index 98b79a2..19307fe 100644 --- a/ci/travis.sh +++ b/ci/travis.sh @@ -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 diff --git a/conf/CMakeLists.txt b/conf/CMakeLists.txt index b27f5b5..326ef77 100644 --- a/conf/CMakeLists.txt +++ b/conf/CMakeLists.txt @@ -1 +1,2 @@ -install(FILES stellar.toml DESTINATION conf COMPONENT Profile) \ No newline at end of file +install(FILES stellar.toml DESTINATION conf COMPONENT Profile) +install(FILES log.toml DESTINATION conf COMPONENT Profile) \ No newline at end of file diff --git a/conf/log.toml b/conf/log.toml new file mode 100644 index 0000000..c49b2ce --- /dev/null +++ b/conf/log.toml @@ -0,0 +1,4 @@ +[log] +output = file # stderr, file +file = stellar.log +level = DEBUG # TRACE, DEBUG, INFO, WARN, ERROR, FATAL diff --git a/conf/stellar.toml b/conf/stellar.toml index a1527c7..7970769 100644 --- a/conf/stellar.toml +++ b/conf/stellar.toml @@ -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] diff --git a/src/config/config.cpp b/src/config/config.cpp index 2450434..9fdfa7b 100644 --- a/src/config/config.cpp +++ b/src/config/config.cpp @@ -304,7 +304,7 @@ void config_dump(struct config *cfg) CONFIG_LOG_DEBUG("system.nr_threads : %d", cfg->sys_cfg.nr_threads); for (uint16_t i = 0; i < cfg->sys_cfg.nr_threads; i++) { - CONFIG_LOG_DEBUG("system.cpu_mask[%d] : %d", i, cfg->sys_cfg.cpu_mask[i]); + CONFIG_LOG_DEBUG("system.cpu_mask[%d] : %d", i, cfg->sys_cfg.cpu_mask[i]); } CONFIG_LOG_DEBUG("session_manager.max_tcp_session_num : %ld", cfg->sess_mgr_cfg.max_tcp_session_num); diff --git a/src/log/CMakeLists.txt b/src/log/CMakeLists.txt index 791be8f..89a3b20 100644 --- a/src/log/CMakeLists.txt +++ b/src/log/CMakeLists.txt @@ -4,4 +4,6 @@ add_library(log log.cpp) target_include_directories(log PUBLIC ${CMAKE_CURRENT_LIST_DIR}) -target_link_libraries(log) \ No newline at end of file +target_link_libraries(log toml) + +add_subdirectory(test) \ No newline at end of file diff --git a/src/log/log.cpp b/src/log/log.cpp index 54c7224..ee9cef5 100644 --- a/src/log/log.cpp +++ b/src/log/log.cpp @@ -1,3 +1,307 @@ -#include "log.h" +#include +#include +#include +#include +#include +#include +#include +#include -// TODO \ No newline at end of file +#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); + } +} \ No newline at end of file diff --git a/src/log/log.h b/src/log/log.h index c96e86c..80cd55b 100644 --- a/src/log/log.h +++ b/src/log/log.h @@ -6,15 +6,46 @@ extern "C" { #endif -#include +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 } diff --git a/src/log/test/CMakeLists.txt b/src/log/test/CMakeLists.txt new file mode 100644 index 0000000..95fe6c2 --- /dev/null +++ b/src/log/test/CMakeLists.txt @@ -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/) \ No newline at end of file diff --git a/src/log/test/conf/log_file.toml b/src/log/test/conf/log_file.toml new file mode 100644 index 0000000..f423a38 --- /dev/null +++ b/src/log/test/conf/log_file.toml @@ -0,0 +1,4 @@ +[log] +output = file # stderr, file +file = stellar.log +level = DEBUG # TRACE, DEBUG, INFO, WARN, ERROR, FATAL diff --git a/src/log/test/conf/log_stderr.toml b/src/log/test/conf/log_stderr.toml new file mode 100644 index 0000000..dda2a3d --- /dev/null +++ b/src/log/test/conf/log_stderr.toml @@ -0,0 +1,4 @@ +[log] +output = stderr # stderr, file +file = stellar.log +level = DEBUG # TRACE, DEBUG, INFO, WARN, ERROR, FATAL diff --git a/src/log/test/gtest_log.cpp b/src/log/test/gtest_log.cpp new file mode 100644 index 0000000..9d8536f --- /dev/null +++ b/src/log/test/gtest_log.cpp @@ -0,0 +1,106 @@ +#include + +#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(); +} diff --git a/src/packet/packet.h b/src/packet/packet.h index 187e4d8..58b5705 100644 --- a/src/packet/packet.h +++ b/src/packet/packet.h @@ -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 { diff --git a/src/session/session_manager.h b/src/session/session_manager.h index 7f2ea23..181a94b 100644 --- a/src/session/session_manager.h +++ b/src/session/session_manager.h @@ -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 { diff --git a/src/stellar/CMakeLists.txt b/src/stellar/CMakeLists.txt index 9c79b00..caad286 100644 --- a/src/stellar/CMakeLists.txt +++ b/src/stellar/CMakeLists.txt @@ -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) \ No newline at end of file diff --git a/src/stellar/logo.h b/src/stellar/logo.h new file mode 100644 index 0000000..ba910a0 --- /dev/null +++ b/src/stellar/logo.h @@ -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 diff --git a/src/stellar/stellar.cpp b/src/stellar/stellar.cpp index 42c4287..7dacbcf 100644 --- a/src/stellar/stellar.cpp +++ b/src/stellar/stellar.cpp @@ -4,28 +4,40 @@ #include #include #include -#include #include +#include +#include +#include +#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; } - packet_parse(&pkt, data, len); - __packet_plugin_dispatch_example(&pkt); + 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); - sess = session_manager_update_session(sess_mgr, &pkt); - __session_plugin_dispatch_example(sess); + 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); - send_packet(data, len); + sess = session_manager_update_session(sess_mgr, &pkt); + __session_plugin_dispatch_example(sess); - sess = session_manager_get_evicted_session(sess_mgr); - __session_plugin_dispatch_example(sess); + 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 \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; } \ No newline at end of file diff --git a/vendor/CMakeLists.txt b/vendor/CMakeLists.txt index 393d864..c995111 100644 --- a/vendor/CMakeLists.txt +++ b/vendor/CMakeLists.txt @@ -22,4 +22,12 @@ set_property(TARGET gtest PROPERTY INTERFACE_LINK_LIBRARIES pthread) 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) \ No newline at end of file +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}) \ No newline at end of file