load config from toml file

cmake/Package.cmake

@@ -31,6 +31,6 @@ else()
 endif()
 
 # setup %config(noreplace)
-set(CPACK_RPM_USER_FILELIST "%config(noreplace) ${CMAKE_INSTALL_PREFIX}/conf/stellar.conf")
+set(CPACK_RPM_USER_FILELIST "%config(noreplace) ${CMAKE_INSTALL_PREFIX}/conf/stellar.toml")
 
 include(CPack)

conf/CMakeLists.txt

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

conf/stellar.conf

@@ -1 +0,0 @@
-# TODO

conf/stellar.toml

@@ -0,0 +1,37 @@
+[system]
+app_symbol = "stellar"
+dev_symbol = "eth0"
+nr_threads = 1
+cpu_mask = [5, 6, 7, 8, 9, 10, 11, 12]
+
+[session_manager]
+# max session number
+max_tcp_session_num = 100
+max_udp_session_num = 100
+
+# session overload evict
+tcp_overload_evict_old_sess = 1 # 1: evict old session, 0: bypass new session
+udp_overload_evict_old_sess = 1 # 1: evict old session, 0: bypass new session
+
+# TCP timeout
+tcp_timeout_init = 5          # seconds, Range: 1-60
+tcp_timeout_handshake = 10    # seconds, Range: 1-60
+tcp_timeout_data = 3600       # seconds, Range: 1-15,999,999
+tcp_timeout_half_closed = 120 # seconds, Range: 1-604,800
+tcp_timeout_time_wait = 15    # seconds, Range: 1-600
+tcp_timeout_discard = 90      # seconds, Range: 1-15,999,999
+
+# UDP timeout
+udp_timeout_data = 10 # seconds, Range: 1-15,999,999
+
+# TCP duplicate packet filter
+tcp_dupkt_filter_enable = 1
+tcp_dupkt_filter_capacity = 1000000
+tcp_dupkt_filter_timeout = 10         # seconds, Range: 1-60
+tcp_dupkt_filter_error_rate = 0.00001
+
+# UDP eviction filter
+udp_eviction_filter_enable = 1
+udp_eviction_filter_capacity = 1000000
+udp_eviction_filter_timeout = 10         # seconds, Range: 1-60
+udp_eviction_filter_error_rate = 0.00001

src/CMakeLists.txt

@@ -5,4 +5,5 @@ add_subdirectory(packet)
 add_subdirectory(dupkt_filter)
 add_subdirectory(eviction_filter)
 add_subdirectory(session)
+add_subdirectory(config)
 add_subdirectory(stellar)

src/config/CMakeLists.txt

@@ -0,0 +1,7 @@
+###############################################################################
+# config
+###############################################################################
+
+add_library(config config.cpp)
+target_include_directories(config PUBLIC ${CMAKE_CURRENT_LIST_DIR})
+target_link_libraries(config toml session_manager)

src/config/config.cpp

@@ -0,0 +1,334 @@
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "config.h"
+#include "toml.h"
+
+// return  0: success
+// retuun -1: failed
+static int parse_system_section(struct config *cfg, toml_table_t *conf_file_handle)
+{
+    const char *ptr;
+    toml_table_t *system_table;
+    toml_array_t *mask_array;
+
+    system_table = toml_table_in(conf_file_handle, "system");
+    if (system_table == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing system section");
+        return -1;
+    }
+
+    ptr = toml_raw_in(system_table, "app_symbol");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing system.app_symbol");
+        return -1;
+    }
+    strncpy(cfg->sys_cfg.app_symbol, ptr, sizeof(cfg->sys_cfg.app_symbol) - 1);
+
+    ptr = toml_raw_in(system_table, "dev_symbol");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing system.dev_symbol");
+        return -1;
+    }
+    strncpy(cfg->sys_cfg.dev_symbol, ptr, sizeof(cfg->sys_cfg.dev_symbol) - 1);
+
+    ptr = toml_raw_in(system_table, "nr_threads");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing system.nr_threads");
+        return -1;
+    }
+    cfg->sys_cfg.nr_threads = atoi(ptr);
+
+    mask_array = toml_array_in(system_table, "cpu_mask");
+    if (mask_array == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing system.cpu_mask");
+        return -1;
+    }
+    for (uint16_t i = 0; i < cfg->sys_cfg.nr_threads; i++)
+    {
+        ptr = toml_raw_at(mask_array, i);
+        if (ptr == NULL)
+        {
+            CONFIG_LOG_ERROR("config file missing system.cpu_mask[%d]", i);
+            return -1;
+        }
+        cfg->sys_cfg.cpu_mask[i] = atoi(ptr);
+    }
+
+    return 0;
+}
+
+// return  0: success
+// retuun -1: failed
+static int parse_session_manager_section(struct config *cfg, toml_table_t *conf_file_handle)
+{
+    const char *ptr;
+    toml_table_t *sess_mgr_table;
+
+    sess_mgr_table = toml_table_in(conf_file_handle, "session_manager");
+    if (sess_mgr_table == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager section");
+        return -1;
+    }
+
+    // max session number
+    ptr = toml_raw_in(sess_mgr_table, "max_tcp_session_num");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.max_tcp_session_num");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.max_tcp_session_num = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "max_udp_session_num");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.max_udp_session_num");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.max_udp_session_num = atoll(ptr);
+
+    // session overload (1: evict old session, 0: bypass new session)
+    ptr = toml_raw_in(sess_mgr_table, "tcp_overload_evict_old_sess");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_overload_evict_old_sess");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_overload_evict_old_sess = atoi(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "udp_overload_evict_old_sess");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.udp_overload_evict_old_sess");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.udp_overload_evict_old_sess = atoi(ptr);
+
+    // TCP timeout
+    ptr = toml_raw_in(sess_mgr_table, "tcp_timeout_init");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_timeout_init");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_timeout_init = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_timeout_handshake");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_timeout_handshake");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_timeout_handshake = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_timeout_data");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_timeout_data");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_timeout_data = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_timeout_half_closed");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_timeout_half_closed");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_timeout_half_closed = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_timeout_time_wait");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_timeout_time_wait");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_timeout_time_wait = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_timeout_discard");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_timeout_discard");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_timeout_discard = atoll(ptr);
+
+    // UDP timeout
+    ptr = toml_raw_in(sess_mgr_table, "udp_timeout_data");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.udp_timeout_data");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.udp_timeout_data = atoll(ptr);
+
+    // TCP duplicate packet filter
+    ptr = toml_raw_in(sess_mgr_table, "tcp_dupkt_filter_enable");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_dupkt_filter_enable");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_dupkt_filter_enable = atoi(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_dupkt_filter_capacity");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_dupkt_filter_capacity");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_dupkt_filter_capacity = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_dupkt_filter_timeout");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_dupkt_filter_timeout");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_dupkt_filter_timeout = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "tcp_dupkt_filter_error_rate");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.tcp_dupkt_filter_error_rate");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.tcp_dupkt_filter_error_rate = atof(ptr);
+
+    // UDP eviction filter
+    ptr = toml_raw_in(sess_mgr_table, "udp_eviction_filter_enable");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.udp_eviction_filter_enable");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.udp_eviction_filter_enable = atoi(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "udp_eviction_filter_capacity");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.udp_eviction_filter_capacity");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.udp_eviction_filter_capacity = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "udp_eviction_filter_timeout");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.udp_eviction_filter_timeout");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.udp_eviction_filter_timeout = atoll(ptr);
+
+    ptr = toml_raw_in(sess_mgr_table, "udp_eviction_filter_error_rate");
+    if (ptr == NULL)
+    {
+        CONFIG_LOG_ERROR("config file missing session_manager.udp_eviction_filter_error_rate");
+        return -1;
+    }
+    cfg->sess_mgr_cfg.udp_eviction_filter_error_rate = atof(ptr);
+
+    return 0;
+}
+
+// return  0: success
+// retuun -1: failed
+int config_load(struct config *cfg, const char *cfg_file)
+{
+    int ret = -1;
+    char errbuf[200];
+    FILE *fp = NULL;
+    toml_table_t *conf_file_handle = NULL;
+
+    memset(cfg, 0, sizeof(*cfg));
+
+    fp = fopen(cfg_file, "r");
+    if (fp == NULL)
+    {
+        CONFIG_LOG_ERROR("open config file %s failed, %s", cfg_file, strerror(errno));
+        goto error_out;
+    }
+
+    conf_file_handle = toml_parse_file(fp, errbuf, sizeof(errbuf));
+    if (conf_file_handle == NULL)
+    {
+        CONFIG_LOG_ERROR("parse config file %s failed, %s", cfg_file, errbuf);
+        goto error_out;
+    }
+
+    // system config
+    if (parse_system_section(cfg, conf_file_handle) != 0)
+    {
+        goto error_out;
+    }
+
+    // session manager config
+    if (parse_session_manager_section(cfg, conf_file_handle) != 0)
+    {
+        goto error_out;
+    }
+
+    ret = 0;
+
+error_out:
+    if (conf_file_handle)
+    {
+        toml_free(conf_file_handle);
+    }
+
+    if (fp)
+    {
+        fclose(fp);
+    }
+
+    return ret;
+}
+
+void config_dump(struct config *cfg)
+{
+    if (cfg == NULL)
+    {
+        return;
+    }
+
+    CONFIG_LOG_DEBUG("system.app_symbol       : %s", cfg->sys_cfg.app_symbol);
+    CONFIG_LOG_DEBUG("system.dev_symbol       : %s", cfg->sys_cfg.dev_symbol);
+    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("session_manager.max_tcp_session_num            : %ld", cfg->sess_mgr_cfg.max_tcp_session_num);
+    CONFIG_LOG_DEBUG("session_manager.max_udp_session_num            : %ld", cfg->sess_mgr_cfg.max_udp_session_num);
+
+    CONFIG_LOG_DEBUG("session_manager.tcp_overload_evict_old_sess    : %d", cfg->sess_mgr_cfg.tcp_overload_evict_old_sess);
+    CONFIG_LOG_DEBUG("session_manager.udp_overload_evict_old_sess    : %d", cfg->sess_mgr_cfg.udp_overload_evict_old_sess);
+
+    CONFIG_LOG_DEBUG("session_manager.tcp_timeout_init               : %ld", cfg->sess_mgr_cfg.tcp_timeout_init);
+    CONFIG_LOG_DEBUG("session_manager.tcp_timeout_handshake          : %ld", cfg->sess_mgr_cfg.tcp_timeout_handshake);
+    CONFIG_LOG_DEBUG("session_manager.tcp_timeout_data               : %ld", cfg->sess_mgr_cfg.tcp_timeout_data);
+    CONFIG_LOG_DEBUG("session_manager.tcp_timeout_half_closed        : %ld", cfg->sess_mgr_cfg.tcp_timeout_half_closed);
+    CONFIG_LOG_DEBUG("session_manager.tcp_timeout_time_wait          : %ld", cfg->sess_mgr_cfg.tcp_timeout_time_wait);
+    CONFIG_LOG_DEBUG("session_manager.tcp_timeout_discard            : %ld", cfg->sess_mgr_cfg.tcp_timeout_discard);
+
+    CONFIG_LOG_DEBUG("session_manager.udp_timeout_data               : %ld", cfg->sess_mgr_cfg.udp_timeout_data);
+
+    CONFIG_LOG_DEBUG("session_manager.tcp_dupkt_filter_enable        : %d", cfg->sess_mgr_cfg.tcp_dupkt_filter_enable);
+    CONFIG_LOG_DEBUG("session_manager.tcp_dupkt_filter_capacity      : %ld", cfg->sess_mgr_cfg.tcp_dupkt_filter_capacity);
+    CONFIG_LOG_DEBUG("session_manager.tcp_dupkt_filter_timeout       : %ld", cfg->sess_mgr_cfg.tcp_dupkt_filter_timeout);
+    CONFIG_LOG_DEBUG("session_manager.tcp_dupkt_filter_error_rate    : %f", cfg->sess_mgr_cfg.tcp_dupkt_filter_error_rate);
+
+    CONFIG_LOG_DEBUG("session_manager.udp_eviction_filter_enable     : %d", cfg->sess_mgr_cfg.udp_eviction_filter_enable);
+    CONFIG_LOG_DEBUG("session_manager.udp_eviction_filter_capacity   : %ld", cfg->sess_mgr_cfg.udp_eviction_filter_capacity);
+    CONFIG_LOG_DEBUG("session_manager.udp_eviction_filter_timeout    : %ld", cfg->sess_mgr_cfg.udp_eviction_filter_timeout);
+    CONFIG_LOG_DEBUG("session_manager.udp_eviction_filter_error_rate : %f", cfg->sess_mgr_cfg.udp_eviction_filter_error_rate);
+}

src/config/config.h

@@ -0,0 +1,45 @@
+#ifndef _CONFIG_H
+#define _CONFIG_H
+
+#ifdef __cpluscplus
+extern "C"
+{
+#endif
+
+#include "session_manager.h"
+
+#define CONFIG_LOG_ERROR(format, ...) LOG_ERROR("config", format, ##__VA_ARGS__)
+#ifndef CONFIG_LOG_ERROR
+#define CONFIG_LOG_ERROR(format, ...) \
+    fprintf(stderr, "ERROR (config), " format "\n", ##__VA_ARGS__);
+#endif
+#define CONFIG_LOG_DEBUG(format, ...) LOG_DEBUG("config", format, ##__VA_ARGS__)
+#ifndef CONFIG_LOG_DEBUG
+#define CONFIG_LOG_DEBUG(format, ...) \
+    fprintf(stderr, "DEBUG (config), " format "\n", ##__VA_ARGS__);
+#endif
+
+#define MAX_THREAD_NUM 256
+
+struct system_config
+{
+    char app_symbol[64];
+    char dev_symbol[64];
+    uint16_t nr_threads;
+    uint16_t cpu_mask[MAX_THREAD_NUM];
+};
+
+struct config
+{
+    struct system_config sys_cfg;
+    struct session_manager_config sess_mgr_cfg;
+};
+
+int config_load(struct config *cfg, const char *cfg_file);
+void config_dump(struct config *cfg);
+
+#ifdef __cpluscplus
+}
+#endif
+
+#endif

src/stellar/CMakeLists.txt

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

src/stellar/stellar.cpp

@@ -7,6 +7,7 @@
 #include <sys/prctl.h>
 #include <signal.h>
 
+#include "config.h"
 #include "packet.h"
 #include "timestamp.h"
 #include "session_manager.h"
@@ -34,49 +35,15 @@ struct thread_context
     struct session_manager *sess_mgr;
 };
 
-struct stellar_ctx
+struct stellar_context
 {
     uint64_t need_exit;
-    uint16_t thread_num;
+    struct config config;
 
-    struct session_manager_config sess_mgr_cfg;
+    struct thread_context thread_ctx[MAX_THREAD_NUM];
-
-    struct thread_context thread_ctx[128];
 } g_stellar_ctx = {
     .need_exit = 0,
-    .thread_num = 1,
+};
-    .sess_mgr_cfg = {
-        // max session number
-        .max_tcp_session_num = 3,
-        .max_udp_session_num = 3,
-
-        // session overload
-        .tcp_overload_evict_old_sess = 1, // 1: evict old session, 0: bypass new session
-        .udp_overload_evict_old_sess = 1, // 1: evict old session, 0: bypass new session
-
-        // tcp timeout
-        .tcp_timeout_init = 2,
-        .tcp_timeout_handshake = 2,
-        .tcp_timeout_data = 2,
-        .tcp_timeout_half_closed = 2,
-        .tcp_timeout_time_wait = 2,
-        .tcp_timeout_discard = 2,
-
-        // udp timeout
-        .udp_timeout_data = 1,
-
-        // tcp duplicate packet filter
-        .tcp_dupkt_filter_enable = 1,
-        .tcp_dupkt_filter_capacity = 1000,
-        .tcp_dupkt_filter_timeout = 10,
-        .tcp_dupkt_filter_error_rate = 0.0001,
-
-        // udp eviction filter
-        .udp_eviction_filter_enable = 1,
-        .udp_eviction_filter_capacity = 1000,
-        .udp_eviction_filter_timeout = 10,
-        .udp_eviction_filter_error_rate = 0.0001,
-    }};
 
 static int recv_packet(const char **data)
 {
@@ -116,7 +83,7 @@ static void signal_handler(int signo)
     }
 }
 
-static void __packet_dispatch(const struct packet *pkt)
+static void __packet_plugin_dispatch_example(const struct packet *pkt)
 {
     if (pkt == NULL)
     {
@@ -128,7 +95,7 @@ static void __packet_dispatch(const struct packet *pkt)
     printf("<= packet dispatch\n\n");
 }
 
-static void __session_dispatch(struct session *sess)
+static void __session_plugin_dispatch_example(struct session *sess)
 {
     if (sess == NULL)
     {
@@ -140,42 +107,11 @@ static void __session_dispatch(struct session *sess)
     session_dump(sess);
     printf("<= session dispatch\n\n");
 
+    // after session dispatch, we should reset session current packet and direction
     session_set0_cur_pkt(sess, NULL);
     session_set_cur_dir(sess, SESSION_DIR_NONE);
 }
 
-static int thread_context_init(struct stellar_ctx *ctx)
-{
-    for (uint16_t i = 0; i < ctx->thread_num; i++)
-    {
-        struct thread_context *thread_ctx = &ctx->thread_ctx[i];
-        thread_ctx->index = i;
-        thread_ctx->need_exit = 0;
-        thread_ctx->is_runing = 0;
-
-        thread_ctx->sess_mgr = session_manager_create(&ctx->sess_mgr_cfg);
-        if (thread_ctx->sess_mgr == NULL)
-        {
-            STELLAR_LOG_ERROR("unable to create session manager");
-            return -1;
-        }
-    }
-
-    return 0;
-}
-
-static void thread_context_free(struct stellar_ctx *ctx)
-{
-    for (uint16_t i = 0; i < ctx->thread_num; i++)
-    {
-        struct thread_context *thread_ctx = &ctx->thread_ctx[i];
-        if (ATOMIC_READ(&thread_ctx->is_runing) == 0)
-        {
-            session_manager_destroy(thread_ctx->sess_mgr);
-        }
-    }
-}
-
 static void *thread_cycle(void *arg)
 {
     uint16_t len = 0;
@@ -200,19 +136,19 @@ static void *thread_cycle(void *arg)
         }
 
         packet_parse(&pkt, data, len);
-        __packet_dispatch(&pkt);
+        __packet_plugin_dispatch_example(&pkt);
 
         sess = session_manager_update_session(sess_mgr, &pkt);
-        __session_dispatch(sess);
+        __session_plugin_dispatch_example(sess);
 
         send_packet(data, len);
 
         sess = session_manager_get_evicted_session(sess_mgr);
-        __session_dispatch(sess);
+        __session_plugin_dispatch_example(sess);
 
     poll_wait:
         sess = session_manager_get_expired_session(sess_mgr);
-        __session_dispatch(sess);
+        __session_plugin_dispatch_example(sess);
 
         usleep(1000); // session_manager_get_expire_interval(sess_mgr); (seconds)
     }
@@ -223,9 +159,89 @@ static void *thread_cycle(void *arg)
     return NULL;
 }
 
+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++)
+    {
+        struct thread_context *thread_ctx = &ctx->thread_ctx[i];
+        thread_ctx->index = i;
+        thread_ctx->need_exit = 0;
+        thread_ctx->is_runing = 0;
+
+        thread_ctx->sess_mgr = session_manager_create(sess_mgr_cfg);
+        if (thread_ctx->sess_mgr == NULL)
+        {
+            STELLAR_LOG_ERROR("unable to create session manager");
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+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++)
+    {
+        struct thread_context *thread_ctx = &ctx->thread_ctx[i];
+        if (ATOMIC_READ(&thread_ctx->is_runing) == 0)
+        {
+            session_manager_destroy(thread_ctx->sess_mgr);
+        }
+    }
+}
+
+static int thread_create(struct stellar_context *ctx)
+{
+    struct system_config *sys_cfg = &ctx->config.sys_cfg;
+
+    for (uint16_t i = 0; i < sys_cfg->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)
+        {
+            STELLAR_LOG_ERROR("unable to create worker thread, error %d: %s", errno, strerror(errno));
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+static void thread_destroy(struct stellar_context *ctx)
+{
+    struct system_config *sys_cfg = &ctx->config.sys_cfg;
+
+    for (uint16_t i = 0; i < sys_cfg->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)
+        {
+            sleep(1);
+        }
+    }
+}
+
 int main(int argc, char **argv)
 {
-    // TODO parse command line
+    if (argc != 2)
+    {
+        printf("usage: %s <config file>\n", argv[0]);
+        return 0;
+    }
+
+    if (config_load(&g_stellar_ctx.config, argv[1]) != 0)
+    {
+        return -1;
+    }
+
+    config_dump(&g_stellar_ctx.config);
 
     // TODO init log
 
@@ -243,15 +259,11 @@ int main(int argc, char **argv)
         goto error_out;
     }
 
-    for (uint16_t i = 0; i < g_stellar_ctx.thread_num; i++)
+    if (thread_create(&g_stellar_ctx) != 0)
     {
-        struct thread_context *thread_ctx = &g_stellar_ctx.thread_ctx[i];
+        STELLAR_LOG_ERROR("unable to create worker thread");
-        if (pthread_create(&thread_ctx->tid, NULL, thread_cycle, (void *)thread_ctx) < 0)
-        {
-            STELLAR_LOG_ERROR("unable to create worker thread, error %d: %s", errno, strerror(errno));
         goto error_out;
     }
-    }
 
     while (!g_stellar_ctx.need_exit)
     {
@@ -259,17 +271,9 @@ int main(int argc, char **argv)
         sleep(1);
     }
 
-    for (uint16_t i = 0; i < g_stellar_ctx.thread_num; i++)
-    {
-        struct thread_context *thread_ctx = &g_stellar_ctx.thread_ctx[i];
-        ATOMIC_SET(&thread_ctx->need_exit, 1);
-        while (ATOMIC_READ(&thread_ctx->is_runing) == 1)
-        {
-            sleep(1);
-        }
-    }
-
 error_out:
+    thread_destroy(&g_stellar_ctx);
+
     thread_context_free(&g_stellar_ctx);
 
     // TODO free plugin