OMPUB-1508 For tunnel traffic, asymmetric traffic, and traffic matching the "no intercept" policy, only a session ID index flow table is created

ci/travis.sh

@@ -33,8 +33,8 @@ env | sort
 : "${COMPILER_IS_GNUCXX:=OFF}"
 
 # Install dependency from YUM
-yum install -y mrzcpd numactl-devel zlib-devel librdkafka-devel-1.2.2.1218b3c-1.el8.x86_64 librdkafka-1.2.2.1218b3c-1.el8.x86_64 systemd-devel
+yum install -y mrzcpd-corei7-4.* numactl-devel zlib-devel librdkafka-devel-1.2.2.1218b3c-1.el8.x86_64 librdkafka-1.2.2.1218b3c-1.el8.x86_64 systemd-devel
-yum install -y libcjson-devel libmaatframe-devel libMESA_field_stat2-devel libfieldstat3-devel libfieldstat4-devel libMESA_handle_logger-devel libelua-devel
+yum install -y libcjson-devel libmaatframe-4.1.39.* libmaatframe-devel-4.1.39.* libMESA_field_stat2-devel libfieldstat3-devel-3.0.* libfieldstat4-devel-4.5.* libMESA_handle_logger-devel libelua-devel
 yum install -y libMESA_htable-devel libMESA_prof_load-devel libwiredcfg-devel libWiredLB-devel sapp-devel libbreakpad_mini-devel
 yum install -y libasan
 yum install -y numactl-libs # required by mrzcpd

common/include/tfe_session_table.h

@@ -21,6 +21,7 @@ struct session_node
     uint64_t session_id;             /* first key */
     struct tuple4 session_addr; /* second key */
 
+    uint8_t is_session_id_only_key;
     void *val_data;
     fn_free_cb *val_freecb;
 
@@ -39,7 +40,7 @@ uint64_t session_table_count(struct session_table *table);
 // val_data     : shallow copy (malloc by user, free by val_freecb)
 // return  0 : suceess
 // return -1 : key exists
-int session_table_insert(struct session_table *table, uint64_t session_id, const struct tuple4 *session_addr, void *val_data, const fn_free_cb *val_freecb);
+int session_table_insert(struct session_table *table, uint8_t is_session_id_only_key, uint64_t session_id, const struct tuple4 *session_addr, void *val_data, const fn_free_cb *val_freecb);
 
 // return  0 : success
 // return -1 : key not exists

common/src/tfe_packet_io.cpp

@@ -1095,6 +1095,7 @@ static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser
     uint16_t size = 0;
     uint8_t is_passthrough = 0;
     uint8_t hit_no_intercept = 0;
+    uint8_t is_session_id_only_key = 0;
     uint16_t out_size = 0;
     char stream_traceid[24] = {0};
     char reason_no_intercept_param[] = "Hit No Intercept Policy";
@@ -1284,6 +1285,10 @@ passthrough:
         tuple4_reverse(&inner_tuple4, &s_ctx->c2s_info.tuple4);
     }
 
+    // is_passthrough为1时，只通过session id创建流表，避免四元组相同的情况下，uthash频繁扩展导致崩溃
+    if (is_passthrough)
+        is_session_id_only_key = 1;
+
     // c2s
     sids_copy(&s_ctx->c2s_info.sids, &parser->seq_sids);
     route_ctx_copy(&s_ctx->c2s_info.route_ctx, &parser->seq_route_ctx);
@@ -1293,7 +1298,7 @@ passthrough:
     route_ctx_copy(&s_ctx->s2c_info.route_ctx, &parser->ack_route_ctx);
 
     TFE_LOG_INFO(logger, "%s: session %lu %s active first, hit rule %lu", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->session_addr, rule_id);
-    session_table_insert(thread->session_table, s_ctx->session_id, &(s_ctx->c2s_info.tuple4), s_ctx, session_value_free_cb);
+    session_table_insert(thread->session_table, is_session_id_only_key, s_ctx->session_id, &(s_ctx->c2s_info.tuple4), s_ctx, session_value_free_cb);
     ATOMIC_INC(&(packet_io_fs->session_num));
     if (parser->seq_header)
         FREE(&parser->seq_header);
@@ -1490,9 +1495,11 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
         throughput_metrics_inc(&packet_io_fs->raw_bypass, 1, raw_len);
 
         if (memcmp(&inner_addr, &s_ctx->c2s_info.tuple4, sizeof(struct tuple4)) == 0) {
+            s_ctx->c2s_info.is_e2i_dir = meta.is_e2i_dir;
             throughput_metrics_inc(&s_ctx->c2s_info.rx, 1, raw_len);
         }
         else {
+            s_ctx->s2c_info.is_e2i_dir = meta.is_e2i_dir;
             throughput_metrics_inc(&s_ctx->s2c_info.rx, 1, raw_len);
         }
         

common/src/tfe_session_table.cpp

@@ -32,7 +32,10 @@ void session_table_destory(struct session_table *table)
         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->is_session_id_only_key)
+            {
+                HASH_DELETE(hh2, table->root_by_addr, node);
+            }
 
             if (node->val_freecb && node->val_data)
             {
@@ -57,7 +60,10 @@ void session_table_reset(struct session_table *table)
         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->is_session_id_only_key)
+            {
+                HASH_DELETE(hh2, table->root_by_addr, node);
+            }
 
             if (node->val_freecb && node->val_data)
             {
@@ -86,7 +92,7 @@ uint64_t session_table_count(struct session_table *table)
 
 // session_addr : deep copy
 // val_data     : shallow copy (malloc by user, free by val_freecb)
-int session_table_insert(struct session_table *table, uint64_t session_id, const struct tuple4 *session_addr, void *val_data, const fn_free_cb *val_freecb)
+int session_table_insert(struct session_table *table, uint8_t is_session_id_only_key, uint64_t session_id, const struct tuple4 *session_addr, void *val_data, const fn_free_cb *val_freecb)
 {
     struct session_node *temp = NULL;
     HASH_FIND(hh1, table->root_by_id, &session_id, sizeof(session_id), temp);
@@ -99,13 +105,17 @@ int session_table_insert(struct session_table *table, uint64_t session_id, const
     temp = (struct session_node *)calloc(1, sizeof(struct session_node));
     assert(temp);
 
+    temp->is_session_id_only_key = is_session_id_only_key;
     temp->session_id = session_id;
     memcpy(&temp->session_addr, session_addr, sizeof(struct tuple4));
     temp->val_data = val_data;
     temp->val_freecb = val_freecb;
 
     HASH_ADD(hh1, table->root_by_id, session_id, sizeof(temp->session_id), temp);
-    HASH_ADD(hh2, table->root_by_addr, session_addr, sizeof(temp->session_addr), temp);
+    if (!is_session_id_only_key)
+    {
+        HASH_ADD(hh2, table->root_by_addr, session_addr, sizeof(temp->session_addr), temp);
+    }
 
     TFE_LOG_DEBUG(g_packet_io_logger, "%s: insert: key %lu success", LOG_TAG_STABLE, session_id);
     table->session_node_count++;
@@ -124,7 +134,10 @@ int session_table_delete_by_id(struct session_table *table, uint64_t session_id)
     }
 
     HASH_DELETE(hh1, table->root_by_id, temp);
-    HASH_DELETE(hh2, table->root_by_addr, temp);
+    if (!temp->is_session_id_only_key)
+    {
+        HASH_DELETE(hh2, table->root_by_addr, temp);
+    }
 
     if (temp->val_freecb && temp->val_data)
     {

common/test/test_session_table.cpp

@@ -36,12 +36,12 @@ TEST(STREAM_TABLE, INSERT)
 	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, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == -1);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == -1);
 	EXPECT_TRUE(session_table_count(table) == 1);
 
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == -1);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == -1);
 	EXPECT_TRUE(session_table_count(table) == 2);
 
 	// TEST Destory
@@ -61,8 +61,8 @@ TEST(STREAM_TABLE, SEARCH_BY_ID)
 	INIT_ADDR_V4(addr3, "1.1.1.1", 1111, "2.2.2.2", 2222);
 
 	// TEST Insert
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == 0);
 	EXPECT_TRUE(session_table_count(table) == 2);
 
 	// TEST Search By Session ID
@@ -93,8 +93,8 @@ TEST(STREAM_TABLE, SEARCH_BY_ADDR)
 	INIT_ADDR_V4(addr3, "1.1.1.1", 1111, "2.2.2.2", 2222);
 
 	// TEST Insert
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == 0);
 	EXPECT_TRUE(session_table_count(table) == 2);
 
 	// TEST Search By Session Addr
@@ -128,8 +128,8 @@ TEST(STREAM_TABLE, SEARCH_BY_REVERSE_ADDR)
 	tuple4_reverse(&addr2, &addr2_reverse);
 
 	// TEST Insert
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == 0);
 	EXPECT_TRUE(session_table_count(table) == 2);
 
 	// TEST Search By Session Reverse Addr
@@ -161,8 +161,8 @@ TEST(STREAM_TABLE, DELETE_BY_ID)
 	tuple4_reverse(&addr2, &addr2_reverse);
 
 	// TEST Insert
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == 0);
 	EXPECT_TRUE(session_table_count(table) == 2);
 
 	// TEST Delete By Session ID
@@ -198,8 +198,8 @@ TEST(STREAM_TABLE, DELETE_BY_ADDR)
 	tuple4_reverse(&addr2, &addr2_reverse);
 
 	// TEST Insert
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == 0);
 	EXPECT_TRUE(session_table_count(table) == 2);
 
 	// TEST Delete By Session Addr
@@ -235,8 +235,8 @@ TEST(STREAM_TABLE, DELETE_BY_REVERSE_ADDR)
 	tuple4_reverse(&addr2, &addr2_reverse);
 
 	// TEST Insert
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 0, 2, &addr2, val_world, free) == 0);
 	EXPECT_TRUE(session_table_count(table) == 2);
 
 	// TEST Delete By Session Reverse Addr
@@ -268,8 +268,8 @@ TEST(STREAM_TABLE, RESET)
 	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, 0, 1, &addr1, val_hello, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 1, &addr1, val_hello, free) == -1);
+	EXPECT_TRUE(session_table_insert(table, 0, 1, &addr1, val_hello, free) == -1);
 	EXPECT_TRUE(session_table_count(table) == 1);
 
 	// TEST Reset
@@ -278,8 +278,8 @@ TEST(STREAM_TABLE, RESET)
 	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, 0, 2, &addr2, val_world, free) == 0);
-	EXPECT_TRUE(session_table_insert(table, 2, &addr2, val_world, free) == -1);
+	EXPECT_TRUE(session_table_insert(table, 0, 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);
@@ -288,6 +288,129 @@ TEST(STREAM_TABLE, RESET)
 	session_table_destory(table);
 }
 
+TEST(STREAM_TABLE_SESSION_ID_ONLY_KEY, INSERT)
+{
+	// 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, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 1, 1, &addr1, val_hello, free) == -1);
+	EXPECT_TRUE(session_table_count(table) == 1);
+
+	EXPECT_TRUE(session_table_insert(table, 1, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 1, 2, &addr2, val_world, free) == -1);
+	EXPECT_TRUE(session_table_count(table) == 2);
+
+	// TEST Destory
+	session_table_destory(table);
+}
+
+TEST(STREAM_TABLE_SESSION_ID_ONLY_KEY, SEARCH_BY_ID)
+{
+	// 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);
+	INIT_ADDR_V4(addr3, "1.1.1.1", 1111, "2.2.2.2", 2222);
+
+	// TEST Insert
+	EXPECT_TRUE(session_table_insert(table, 1, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 1, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_count(table) == 2);
+
+	// TEST Search By Session ID
+	struct session_node *node = NULL;
+	node = session_table_search_by_id(table, 1);
+	EXPECT_TRUE(node != nullptr);
+	EXPECT_STREQ((const char *)node->val_data, "HELLO");
+	node = session_table_search_by_id(table, 2);
+	EXPECT_TRUE(node != nullptr);
+	EXPECT_STREQ((const char *)node->val_data, "WORLD");
+	node = session_table_search_by_id(table, 3);
+	EXPECT_TRUE(node == nullptr);
+
+	// TEST Destory
+	session_table_destory(table);
+}
+
+TEST(STREAM_TABLE_SESSION_ID_ONLY_KEY, DELETE_BY_ID)
+{
+	// 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);
+	struct tuple4 addr1_reverse;
+	struct tuple4 addr2_reverse;
+	tuple4_reverse(&addr1, &addr1_reverse);
+	tuple4_reverse(&addr2, &addr2_reverse);
+
+	// TEST Insert
+	EXPECT_TRUE(session_table_insert(table, 1, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 1, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_count(table) == 2);
+
+	// TEST Delete By Session ID
+	EXPECT_TRUE(session_table_delete_by_id(table, 1) == 0);
+	EXPECT_TRUE(session_table_search_by_id(table, 1) == NULL);
+	EXPECT_TRUE(session_table_search_by_addr(table, &addr1) == NULL);
+	EXPECT_TRUE(session_table_search_by_addr(table, &addr1_reverse) == NULL);
+	EXPECT_TRUE(session_table_count(table) == 1);
+
+	EXPECT_TRUE(session_table_delete_by_id(table, 2) == 0);
+	EXPECT_TRUE(session_table_search_by_id(table, 2) == NULL);
+	EXPECT_TRUE(session_table_search_by_addr(table, &addr2) == NULL);
+	EXPECT_TRUE(session_table_search_by_addr(table, &addr2_reverse) == NULL);
+	EXPECT_TRUE(session_table_count(table) == 0);
+
+	// TEST Destory
+	session_table_destory(table);
+}
+
+TEST(STREAM_TABLE_SESSION_ID_ONLY_KEY, 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, 1, &addr1, val_hello, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 1, 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_count(table) == 0);
+
+	EXPECT_TRUE(session_table_insert(table, 1, 2, &addr2, val_world, free) == 0);
+	EXPECT_TRUE(session_table_insert(table, 1, 2, &addr2, val_world, free) == -1);
+	EXPECT_TRUE(session_table_search_by_id(table, 2) != nullptr);
+	EXPECT_TRUE(session_table_count(table) == 1);
+
+	// TEST Destory
+	session_table_destory(table);
+}
+
 int main(int argc, char **argv)
 {
 	::testing::InitGoogleTest(&argc, argv);

vendor/CMakeLists.txt

@@ -122,8 +122,8 @@ set_property(TARGET gmock PROPERTY INTERFACE_INCLUDE_DIRECTORIES ${INSTALL_DIR}/
 
 set(MESA_FRAMEWORK_LIB_DIR /opt/MESA/lib)
 set(MESA_FRAMEWORK_INCLUDE_DIR /opt/MESA/include)
-set(MRZCPD_LIB_DIR /opt/mrzcpd/lib)
+set(MRZCPD_LIB_DIR /opt/tsg/mrzcpd/corei7/lib)
-set(MRZCPD_INCLUDE_DIR /opt/mrzcpd/include)
+set(MRZCPD_INCLUDE_DIR /opt/tsg/mrzcpd/corei7/include)
 
 add_library(MESA_handle_logger SHARED IMPORTED GLOBAL)
 set_property(TARGET MESA_handle_logger PROPERTY IMPORTED_LOCATION ${MESA_FRAMEWORK_LIB_DIR}/libMESA_handle_logger.so)