TSG-10254 eBPF支持二元组分流

TSG-10527 eBPF支持四元组分流
TSG-10369 KNI多线程在TAP模式下支持eBPF分流

1 files changed, 127 insertions, 57 deletions

diff --git a/entry/src/kni_entry.cpp b/entry/src/kni_entry.cpp
index 96e86d0..b0d581b 100644
--- a/entry/src/kni_entry.cpp
+++ b/entry/src/kni_entry.cpp
@@ -27,7 +27,8 @@ extern "C" {
 #ifdef __cplusplus
 }
 #endif
-#include "kni_tun.h"
+#include "kni_tap_rss.h"
+#include <linux/if_tun.h>
 #include <tsg/tsg_statistic.h>
 #include <MESA/stream_inc/stream_control.h>
 #include "kni_entry.h"
@@ -805,28 +806,34 @@ static int send_to_tfe_normal_mode(char *raw_data, uint16_t raw_len, int thread_
 	return 0;
 }
 
-static int send_to_tfe_tun_mode(char *raw_data, uint16_t raw_len, addr_type_t addr_type){
-	struct kni_tun_handle *handle = g_kni_handle->tun_handle;
-	char *dst_data = ALLOC(char, KNI_MTU);
-	add_ether_header(dst_data, raw_data, raw_len, addr_type);
-	int ret = kni_tun_write(handle, dst_data, raw_len + sizeof(struct ethhdr));
-	FREE(&dst_data);
-	if(ret < 0){
+static int send_to_tfe_tap_mode(char *raw_data, uint16_t raw_len, int thread_seq, addr_type_t addr_type, void *logger)
+{
+	int tap_fd = g_kni_handle->threads_handle[thread_seq].tap_fd;
+	char *send_buff = g_kni_handle->threads_handle[thread_seq].buff;
+
+	add_ether_header(send_buff, raw_data, raw_len, addr_type);
+	int ret = kni_tap_write_per_thread(tap_fd, send_buff, raw_len + sizeof(struct ethhdr), logger);
+	if (ret < 0)
+	{
 		return -1;
 	}
+
 	return 0;
 }
 
-static int send_to_tfe(char *raw_data, uint16_t raw_len, int thread_seq, int tfe_id, addr_type_t addr_type){
+static int send_to_tfe(char *raw_data, uint16_t raw_len, int thread_seq, int tfe_id, addr_type_t addr_type, void *logger)
+{
 	int mode = g_kni_handle->deploy_mode;
-	int ret;
-	if(mode == KNI_DEPLOY_MODE_TUN){
-		ret = send_to_tfe_tun_mode(raw_data, raw_len, addr_type);
-	}
-	else{
-		ret = send_to_tfe_normal_mode(raw_data, raw_len, thread_seq, tfe_id, addr_type);
+	switch (mode)
+	{
+	case KNI_DEPLOY_MODE_TAP:
+		return send_to_tfe_tap_mode(raw_data, raw_len, thread_seq, addr_type, logger);
+	case KNI_DEPLOY_MODE_NORMAL:
+		return send_to_tfe_normal_mode(raw_data, raw_len, thread_seq, tfe_id, addr_type);
+	default:
+		assert(0);
+		return -1;
 	}
-	return ret;
 }
 
 static void wrapped_kni_header_parse(const void *a_packet, struct pme_info *pmeinfo, struct pkt_info *pktinfo){
@@ -1062,10 +1069,6 @@ static int dabloom_add(struct pkt_info *pktinfo, int thread_seq){
 	return ret;
 }
 
-
-
-
-
 static struct session_attribute_label * kni_pull_session_attribute_results(struct streaminfo *a_stream,struct pme_info *pmeinfo)
 {
 	struct session_attribute_label *session_attribute = NULL;
@@ -1148,7 +1151,7 @@ static struct session_attribute_label * kni_pull_session_attribute_results(struc
 			}else{
 				KNI_LOG_DEBUG(logger, "share-session-attribute: destination ip location country is: %s,stream traceid = %s",session_attribute->server_location->country_full,pmeinfo->stream_traceid);			
 			}
-			if(session_attribute->server_location->province_full,pmeinfo->stream_traceid == NULL){
+			if(session_attribute->server_location->province_full == NULL){
 				KNI_LOG_DEBUG(logger, "share-session-attribute: destination ip location province is NULL,stream traceid = %s",pmeinfo->stream_traceid);
 			}else{
 				KNI_LOG_DEBUG(logger, "share-session-attribute: destination ip location province is: %s,stream traceid = %s",session_attribute->server_location->province_full,pmeinfo->stream_traceid);				
@@ -1438,7 +1441,7 @@ static int first_data_intercept(struct streaminfo *stream, struct pme_info *pmei
 	//Note: traceid2pme_add_fail, still work. no cmsg
 	traceid2pme_htable_add(pmeinfo);
 	//send to tfe
-	ret = send_to_tfe(buff, len, thread_seq, pmeinfo->tfe_id, pmeinfo->addr_type);
+	ret = send_to_tfe(buff, len, thread_seq, pmeinfo->tfe_id, pmeinfo->addr_type, logger);
 	if(ret < 0){
 		KNI_LOG_DEBUG(logger, "Intercept error: failed at send first packet to tfe%d, stream traceid = %s, stream addr = %s", 
 							pmeinfo->tfe_id, pmeinfo->stream_traceid, pmeinfo->stream_addr);
@@ -1578,7 +1581,7 @@ char next_data_intercept(struct pme_info *pmeinfo, const void *a_packet, struct
 		}
 	}
 
-	ret = send_to_tfe((char*)a_packet, len, thread_seq, pmeinfo->tfe_id, pmeinfo->addr_type);
+	ret = send_to_tfe((char*)a_packet, len, thread_seq, pmeinfo->tfe_id, pmeinfo->addr_type, logger);
 	if(ret < 0){
 		KNI_LOG_ERROR(logger, "Failed at send continue packet to tfe%d, stream traceid = %s, stream addr = %s", 
 						pmeinfo->tfe_id, pmeinfo->stream_traceid, pmeinfo->stream_addr);
@@ -1878,7 +1881,6 @@ int tuple2stream_htable_search(MESA_htable_handle handle, struct ethhdr *ether_h
 	tuple2stream_htable_value *value = NULL;
 	struct pkt_info pktinfo;
 	int reversed = 0, ret;
-	int key_size = 0;
 	char key_str[KNI_ADDR_MAX];
 	struct stream_tuple4_v6 key_v6;
 	struct stream_tuple4_v4 key_v4;
@@ -1973,21 +1975,28 @@ extern "C" char kni_polling_all_entry(const struct streaminfo *stream, void** pm
 			marsio_buff_free(g_kni_handle->marsio_handle->instance, rx_buffs, nr_recv, 0, 0);
 		}
 	}
-	//tun mode
-	else{
-		char buff[KNI_MTU];
-		int ret = kni_tun_read(g_kni_handle->tun_handle, buff, sizeof(buff));
-		if(ret < 0){
-			KNI_LOG_ERROR(logger, "Failed at read from tun");
-		}
-		else{
-			if(ret > 0){
-				struct ethhdr *ether_hdr = (struct ethhdr*)buff;
+	else if (g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_TAP)
+	{
+		int tap_fd = g_kni_handle->threads_handle[thread_seq].tap_fd;
+		char *recv_buff = g_kni_handle->threads_handle[thread_seq].buff;
+		int recv_buff_size = g_kni_handle->threads_handle[thread_seq].buff_size;
+
+		// 只支持一个 TFE, 每次最多收 100 个包
+		for (int j = 0; j < 100; j++)
+		{
+			if (kni_tap_read_per_thread(tap_fd, recv_buff, recv_buff_size, logger) > 0)
+			{
+				struct ethhdr *ether_hdr = (struct ethhdr *)recv_buff;
 				tuple2stream_htable_search(tuple2stream_htable, ether_hdr, thread_seq);
 				flag = POLLING_STATE_WORK;
 			}
+			else
+			{
+				break;
+			}
 		}
 	}
+
 	return flag;
 }
 
@@ -2433,12 +2442,26 @@ error_out:
 	return NULL;
 }
 
+extern "C" void kni_destroy()
+{
+	if (g_kni_handle != NULL)
+	{
+		for (int i = 0; i < g_kni_handle->thread_count; i++)
+		{
+			kni_tap_close_per_thread(g_kni_handle->threads_handle[i].tap_fd);
+
+			if (g_kni_handle->threads_handle[i].buff)
+			{
+				free(g_kni_handle->threads_handle[i].buff);
+				g_kni_handle->threads_handle[i].buff = NULL;
+			}
+		}
+
+		kni_tap_global_unload_rss_bpf(g_kni_handle->tap_bpf_ctx);
 
-extern "C" void kni_destroy(){
-	if(g_kni_handle != NULL){
 		FREE(&g_kni_handle);
+		g_kni_handle = NULL;
 	}
-	g_kni_handle = NULL;
 }
 
 //eliminate_type: 0:FIFO; 1:LRU
@@ -2514,7 +2537,7 @@ extern "C" int kni_init(){
 	char *kni_git_verison = (char*)KNI_GIT_VERSION;
 	const char *profile = "./etc/kni/kni.conf";
 	const char *section = "global";
-
+	char tap_name[1024] = {0};
 	//init logger
 	char log_path[KNI_PATH_MAX] = "";
 	int tfe_node_count = 1;
@@ -2551,10 +2574,20 @@ extern "C" int kni_init(){
 
 	char deploy_mode[KNI_SYMBOL_MAX];
 	ret = MESA_load_profile_string_def(profile, section, "deploy_mode", deploy_mode, sizeof(deploy_mode), "normal");
-	g_kni_handle->deploy_mode = KNI_DEPLOY_MODE_NORMAL;
-	if(strcmp(deploy_mode, "tun") == 0){
-		g_kni_handle->deploy_mode = KNI_DEPLOY_MODE_TUN;
+	if (strcmp(deploy_mode, "tap") == 0)
+	{
+		g_kni_handle->deploy_mode = KNI_DEPLOY_MODE_TAP;
+	}
+	else if (strcmp(deploy_mode, "normal") == 0)
+	{
+		g_kni_handle->deploy_mode = KNI_DEPLOY_MODE_NORMAL;
+	}
+	else
+	{
+		KNI_LOG_ERROR(local_logger, "KNI Invalid deploy_mode: %s\n", deploy_mode);
+		goto error_out;
 	}
+	
 	if(g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_NORMAL){
 		ret = MESA_load_profile_int_nodef(profile, section, "tfe_node_count", &tfe_node_count);
 		if(ret < 0){
@@ -2622,27 +2655,54 @@ extern "C" int kni_init(){
 			goto error_out;
 		}
 	}
-
-	//init tun
-	if(g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_TUN){
-		if(g_kni_handle->thread_count != 1){
-			KNI_LOG_ERROR(local_logger, "Tun mode, thread count must be 1, while it's %d", g_kni_handle->thread_count);
+	else if (g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_TAP)
+	{
+		if (tfe_node_count != 1)
+		{
+			KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, only support one tfe node.");
 			goto error_out;
 		}
-		char tun_name[KNI_SYMBOL_MAX];
-		ret = MESA_load_profile_string_nodef(profile, section, "tun_name", tun_name, sizeof(tun_name));
-		if(ret < 0){
-			KNI_LOG_ERROR(local_logger, "MESA_prof_load: tun_name not set, profile = %s, section = %s", profile, section);
+
+		if (MESA_load_profile_string_nodef(profile, "tap", "tap_name", tap_name, sizeof(tap_name)) < 0)
+		{
+			KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, tap_name is required.");
 			goto error_out;
 		}
-		KNI_LOG_ERROR(local_logger, "MESA_prof_load, [%s]:\n tun_name: %s", section, tun_name);
-		g_kni_handle->tun_handle = kni_tun_init(tun_name, KNI_TUN_MODE_NOBLOCK, local_logger);
-		if(g_kni_handle->tun_handle == NULL){
-			KNI_LOG_ERROR(local_logger, "Failed at init kni_tun");
-			goto error_out;
+
+		int tap_allow_mutilthread = 0;
+		MESA_load_profile_int_nodef(profile, "tap", "tap_allow_mutilthread", &tap_allow_mutilthread);
+		if (tap_allow_mutilthread)
+		{
+			uint32_t bpf_debug_log = 0;
+			uint32_t bpf_hash_mode = 2;
+			uint32_t bpf_queue_num = get_thread_count();
+			char bpf_obj[1024] = {0};
+			MESA_load_profile_int_nodef(profile, "tap", "bpf_debug_log", (int *)&bpf_debug_log);
+			MESA_load_profile_int_nodef(profile, "tap", "bpf_hash_mode", (int *)&bpf_hash_mode);
+			ret = MESA_load_profile_string_nodef(profile, "tap", "bpf_obj", bpf_obj, sizeof(bpf_obj));
+			if (ret < 0)
+			{
+				KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, when enable tap_allow_mutilthread, bpf_obj is required.");
+				goto error_out;
+			}
+
+			g_kni_handle->tap_bpf_ctx = kni_tap_global_load_rss_bpf(bpf_obj, bpf_queue_num, bpf_hash_mode, bpf_debug_log, local_logger);
+			if (g_kni_handle->tap_bpf_ctx == NULL)
+			{
+				goto error_out;
+			}
+		}
+		else
+		{
+			if (get_thread_count() > 1)
+			{
+				KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, when disable tap_allow_mutilthread, only support one work thread.");
+				goto error_out;
+			}
 		}
 	}
-	
+
+
 	//init_filedstat
 	fs_handle = fs_init(profile);
 	if(fs_handle == NULL){
@@ -2673,7 +2733,17 @@ extern "C" int kni_init(){
 			goto error_out;
 		}
 		g_kni_handle->threads_handle[i].tuple2stream_htable = tuple2stream_htable;
- 	}
+		if (g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_TAP)
+		{
+			g_kni_handle->threads_handle[i].tap_fd = kni_tap_open_per_thread(tap_name, IFF_TAP | IFF_NO_PI | IFF_MULTI_QUEUE, kni_tap_get_bpf_prog_fd(g_kni_handle->tap_bpf_ctx), local_logger);
+			if (g_kni_handle->threads_handle[i].tap_fd == -1)
+			{
+				goto error_out;
+			}
+			g_kni_handle->threads_handle[i].buff_size = KNI_MTU;
+			g_kni_handle->threads_handle[i].buff = ALLOC(char, g_kni_handle->threads_handle[i].buff_size);
+		}
+	}
 	//init ssl dynamic bypass htable
 	ret = ssl_dynamic_bypass_htable_init(profile, local_logger);
 	if(ret < 0){