refactor: packet manager and session manager add on_thread_init/on_thread_exit entry

4 files changed, 75 insertions, 49 deletions

diff --git a/infra/session_manager/session_manager.c b/infra/session_manager/session_manager.c
index 4b2aed9..a6b7929 100644
--- a/infra/session_manager/session_manager.c
+++ b/infra/session_manager/session_manager.c
@@ -30,7 +30,6 @@ struct session_manager_schema
 
 struct session_manager
 {
-    uint16_t thread_num;
     struct session_manager_config *cfg;
     struct session_manager_schema *schema;
     struct session_manager_runtime *runtime[MAX_THREAD_NUM];
@@ -301,30 +300,8 @@ error_out:
 
 void session_manager_free(struct session_manager *sess_mgr)
 {
-    struct session_manager_stat *stat = NULL;
-    struct session_manager_runtime *sess_mgr_rt = NULL;
-
     if (sess_mgr)
     {
-        for (int i = 0; i < sess_mgr->thread_num; i++)
-        {
-            sess_mgr_rt = sess_mgr->runtime[i];
-            if (sess_mgr_rt == NULL)
-            {
-                continue;
-            }
-
-            stat = session_manager_runtime_get_stat(sess_mgr_rt);
-            while (stat->tcp_sess_used || stat->udp_sess_used)
-            {
-                clean_session(sess_mgr_rt, UINT64_MAX);
-            }
-
-            SESSION_MANAGER_LOG_INFO("runtime: %p, idx: %d, will be cleaned", sess_mgr_rt, i);
-            session_manager_runtime_print_stat(sess_mgr_rt);
-            session_manager_runtime_free(sess_mgr->runtime[i]);
-        }
-
         session_manager_schema_free(sess_mgr->schema);
         session_manager_config_free(sess_mgr->cfg);
         free(sess_mgr);
@@ -333,20 +310,6 @@ void session_manager_free(struct session_manager *sess_mgr)
 
 struct session_manager *session_manager_new(struct stellar_module_manager *mod_mgr, struct packet_manager *pkt_mgr, struct mq_schema *mq_schema, const char *toml_file)
 {
-
-    uint64_t thread_num;
-    uint64_t instance_id;
-    uint64_t now_ms = clock_get_real_time_ms();
-
-    if (load_toml_integer_config(toml_file, "instance.id", (uint64_t *)&instance_id, 0, 4095))
-    {
-        return NULL;
-    }
-    if (load_toml_integer_config(toml_file, "packet_io.thread_num", (uint64_t *)&thread_num, 0, MAX_THREAD_NUM))
-    {
-        return NULL;
-    }
-
     struct session_manager *sess_mgr = calloc(1, sizeof(struct session_manager));
     if (sess_mgr == NULL)
     {
@@ -368,18 +331,6 @@ struct session_manager *session_manager_new(struct stellar_module_manager *mod_m
         goto error_out;
     }
 
-    sess_mgr->thread_num = (uint16_t)thread_num;
-    for (int i = 0; i < sess_mgr->thread_num; i++)
-    {
-        sess_mgr->cfg->session_id_seed = instance_id << 8 | i;
-        sess_mgr->runtime[i] = session_manager_runtime_new(sess_mgr->cfg, now_ms);
-        if (sess_mgr->runtime[i] == NULL)
-        {
-            SESSION_MANAGER_LOG_ERROR("failed to create session_manager_runtime");
-            goto error_out;
-        }
-    }
-
     stellar_module_manager_polling_subscribe(mod_mgr, on_polling, sess_mgr);
 
     return sess_mgr;
@@ -429,6 +380,49 @@ int session_manager_subscribe_tcp_stream(struct session_manager *sess_mgr, on_tc
     return mq_schema_subscribe(sess_mgr->schema->mq, sess_mgr->schema->topic_id_tcp_stream, (on_msg_cb_func *)(void *)cb, args);
 }
 
+int session_manager_init(struct session_manager *sess_mgr, uint16_t thread_id)
+{
+    assert(sess_mgr);
+    assert(thread_id < sess_mgr->cfg->thread_num);
+    uint64_t now_ms = clock_get_real_time_ms();
+
+    sess_mgr->cfg->session_id_seed = sess_mgr->cfg->instance_id << 8 | thread_id;
+    struct session_manager_runtime *sess_mgr_rt = session_manager_runtime_new(sess_mgr->cfg, now_ms);
+    if (sess_mgr_rt == NULL)
+    {
+        SESSION_MANAGER_LOG_ERROR("failed to create session_manager_runtime");
+        return -1;
+    }
+    else
+    {
+        sess_mgr->runtime[thread_id] = sess_mgr_rt;
+        return 0;
+    }
+}
+
+void session_manager_clean(struct session_manager *sess_mgr, uint16_t thread_id)
+{
+    assert(sess_mgr);
+    assert(thread_id < sess_mgr->cfg->thread_num);
+
+    struct session_manager_runtime *sess_mgr_rt = sess_mgr->runtime[thread_id];
+    if (sess_mgr_rt == NULL)
+    {
+        return;
+    }
+
+    struct session_manager_stat *stat = session_manager_runtime_get_stat(sess_mgr_rt);
+    while (stat->tcp_sess_used || stat->udp_sess_used)
+    {
+        clean_session(sess_mgr_rt, UINT64_MAX);
+    }
+
+    SESSION_MANAGER_LOG_INFO("runtime: %p, idx: %d, will be cleaned", sess_mgr_rt, thread_id);
+    session_manager_runtime_print_stat(sess_mgr_rt);
+    session_manager_runtime_free(sess_mgr_rt);
+    sess_mgr_rt = NULL;
+}
+
 /******************************************************************************
  * session manager module
  ******************************************************************************/
@@ -483,4 +477,30 @@ void session_manager_on_exit(struct stellar_module_manager *mod_mgr __attribute_
         stellar_module_free(mod);
         SESSION_MANAGER_LOG_FATAL("session_manager exited");
     }
+}
+
+struct stellar_module *session_manager_on_thread_init(struct stellar_module_manager *mod_mgr, int thread_id, struct stellar_module *mod)
+{
+    struct session_manager *sess_mgr = stellar_module_get_ctx(mod);
+    assert(sess_mgr);
+    assert(thread_id < sess_mgr->cfg->thread_num);
+
+    if (session_manager_init(sess_mgr, thread_id) != 0)
+    {
+        SESSION_MANAGER_LOG_ERROR("failed to int session_manager_init");
+        return NULL;
+    }
+    else
+    {
+        return mod;
+    }
+}
+
+void session_manager_on_thread_exit(struct stellar_module_manager *mod_mgr __attribute__((unused)), int thread_id, struct stellar_module *mod)
+{
+    struct session_manager *sess_mgr = stellar_module_get_ctx(mod);
+    assert(sess_mgr);
+    assert(thread_id < sess_mgr->cfg->thread_num);
+
+    session_manager_clean(sess_mgr, thread_id);
 }
\ No newline at end of file
diff --git a/infra/session_manager/session_manager_runtime.c b/infra/session_manager/session_manager_runtime.c
index b2aa250..2434c19 100644
--- a/infra/session_manager/session_manager_runtime.c
+++ b/infra/session_manager/session_manager_runtime.c
@@ -481,6 +481,8 @@ struct session_manager_config *session_manager_config_new(const char *toml_file)
      }
 
      int ret = 0;
+     ret += load_toml_integer_config(toml_file, "instance.id", (uint64_t *)&sess_mgr_cfg->instance_id, 0, 4095);
+     ret += load_toml_integer_config(toml_file, "packet_io.thread_num", (uint64_t *)&sess_mgr_cfg->thread_num, 0, MAX_THREAD_NUM);
      ret += load_toml_integer_config(toml_file, "session_manager.tcp_session_max", (uint64_t *)&sess_mgr_cfg->tcp_session_max, EVICTE_SESSION_BURST * 2, UINT64_MAX);
      ret += load_toml_integer_config(toml_file, "session_manager.udp_session_max", (uint64_t *)&sess_mgr_cfg->udp_session_max, EVICTE_SESSION_BURST * 2, UINT64_MAX);
 
diff --git a/infra/session_manager/session_manager_runtime.h b/infra/session_manager/session_manager_runtime.h
index c72c959..4178e4c 100644
--- a/infra/session_manager/session_manager_runtime.h
+++ b/infra/session_manager/session_manager_runtime.h
@@ -15,6 +15,8 @@ extern "C"
 
 struct session_manager_config
 {
+    uint64_t instance_id;
+    uint16_t thread_num;
     uint64_t session_id_seed;
     uint64_t tcp_session_max;
     uint64_t udp_session_max;
diff --git a/infra/session_manager/test/default_config.h b/infra/session_manager/test/default_config.h
index f8ce5fd..f6d8af6 100644
--- a/infra/session_manager/test/default_config.h
+++ b/infra/session_manager/test/default_config.h
@@ -10,6 +10,8 @@ extern "C"
 #include "session_manager_runtime.h"
 
 static struct session_manager_config sess_mgr_cfg = {
+    .instance_id = 1,
+    .thread_num = 1,
     .session_id_seed = 0xFFFFF,
     .tcp_session_max = 256,
     .udp_session_max = 256,