执行resetall指令：Sapp重启后发送resetall状态通知SCE清空流表

common/include/session_table.h

@@ -32,6 +32,7 @@ struct session_table;
 
 struct session_table *session_table_create();
 void session_table_destory(struct session_table *table);
+void session_table_reset(struct session_table *table);
 uint64_t session_table_count(struct session_table *table);
 
 // session_addr : deep copy

common/src/session_table.cpp

@@ -47,6 +47,30 @@ void session_table_destory(struct session_table *table)
     }
 }
 
+void session_table_reset(struct session_table *table)
+{
+    if (table)
+    {
+        struct session_node *temp = NULL;
+        struct session_node *node = NULL;
+        HASH_ITER(hh1, table->root_by_id, node, temp)
+        {
+            HASH_DELETE(hh1, table->root_by_id, node);
+            HASH_DELETE(hh2, table->root_by_addr, node);
+
+            if (node->val_freecb && node->val_data)
+            {
+                node->val_freecb(node->val_data);
+            }
+
+            free(node);
+            node = NULL;
+
+            table->session_node_count--;
+        }
+    }
+}
+
 uint64_t session_table_count(struct session_table *table)
 {
     if (table)

common/test/gtest_session_table.cpp

@@ -235,6 +235,38 @@ TEST(STREAM_TABLE, DELETE_BY_REVERSE_ADDR)
 	session_table_destory(table);
 }
 
+TEST(STREAM_TABLE, RESET)
+{
+	// TEST Create
+	struct session_table *table = session_table_create();
+	EXPECT_TRUE(table != nullptr);
+
+	char *val_hello = strdup("HELLO");
+	char *val_world = strdup("WORLD");
+	INIT_ADDR_V4(addr1, "1.2.3.4", 1234, "4.3.2.1", 4321);
+	INIT_ADDR_V6(addr2, "2:3:4::5", 2345, "5:4:3::2", 5342);
+
+	// TEST Insert
+	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == -1);
+	EXPECT_TRUE(session_table_count(table) == 1);
+
+	// TEST Reset
+	session_table_reset(table);
+	EXPECT_TRUE(session_table_search_by_id(table, 1) == nullptr);
+	EXPECT_TRUE(session_table_search_by_addr(table, &addr1) == nullptr);
+	EXPECT_TRUE(session_table_count(table) == 0);
+
+	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == -1);
+	EXPECT_TRUE(session_table_search_by_id(table, 2) != nullptr);
+	EXPECT_TRUE(session_table_search_by_addr(table, &addr2) != nullptr);
+	EXPECT_TRUE(session_table_count(table) == 1);
+
+	// TEST Destory
+	session_table_destory(table);
+}
+
 int main(int argc, char **argv)
 {
 	::testing::InitGoogleTest(&argc, argv);

platform/include/sce.h

@@ -48,6 +48,9 @@ struct thread_ctx
     struct packet_io *ref_io;
     struct global_metrics *ref_metrics;
     struct policy_enforcer *ref_enforcer;
+    struct sce_ctx *ref_sce_ctx;
+
+    int session_table_need_reset;
 };
 
 /******************************************************************************

platform/src/main.cpp

@@ -28,7 +28,11 @@ static void *worker_thread_cycle(void *arg)
             LOG_INFO("%s: worker thread %d recv %03d packets from endpoint", LOG_TAG_SCE, thread_ctx->thread_index, n_packet_recv);
         }
 
-        // TODO reset session_table
+        if (__atomic_fetch_add(&thread_ctx->session_table_need_reset, 0, __ATOMIC_RELAXED) > 0)
+        {
+            session_table_reset(thread_ctx->session_table);
+            __atomic_fetch_and(&thread_ctx->session_table_need_reset, 0, __ATOMIC_RELAXED);
+        }
     }
 
     LOG_ERROR("%s: worker thread %d exiting", LOG_TAG_SCE, thread_ctx->thread_index);
@@ -53,6 +57,8 @@ int main(int argc, char **argv)
         ctx->work_threads[i].ref_io = ctx->io;
         ctx->work_threads[i].ref_metrics = ctx->metrics;
         ctx->work_threads[i].ref_enforcer = ctx->enforcer;
+        ctx->work_threads[i].ref_sce_ctx = ctx;
+        ctx->work_threads[i].session_table_need_reset = 0;
     }
 
     for (int i = 0; i < ctx->nr_worker_threads; i++)

platform/src/packet_io.cpp

@@ -1186,12 +1186,17 @@ static int handle_session_resetall(struct metadata *meta, struct ctrl_pkt_parser
 {
     struct thread_ctx *thread = (struct thread_ctx *)ctx;
     struct global_metrics *g_metrics = thread->ref_metrics;
+    struct sce_ctx *sce_ctx = thread->ref_sce_ctx;
 
     LOG_ERROR("%s: session %lu notification clears all session tables !!!", LOG_TAG_PKTIO, meta->session_id);
 
     __atomic_fetch_and(&g_metrics->session_nums, 0, __ATOMIC_RELAXED);
 
-    // TODO reset all session_table
+    for (int i = 0; i < sce_ctx->nr_worker_threads; i++)
+    {
+        struct thread_ctx *thread_ctx = &sce_ctx->work_threads[i];
+        __atomic_fetch_add(&thread_ctx->session_table_need_reset, 1, __ATOMIC_RELAXED);
+    }
 
     return 0;
 }