✨ feat(udp dup stream): 支持udp重复流识别，key为四元组

11 files changed, 56 insertions, 19 deletions

diff --git a/bin/etc/sapp.toml b/bin/etc/sapp.toml
index 7b88469..19a843e 100644
--- a/bin/etc/sapp.toml
+++ b/bin/etc/sapp.toml
@@ -111,6 +111,7 @@
 
 [DUPLICATE_PKT]
 [dup_pkt.traffic.original]
+    kickout_udp_stream_enabled=1
     original_ipv4_tcp_enabled=1
     original_ipv4_udp_enabled=0
 ### note, can't distinguish between duplicate traffic and application retransmit traffic for IPv6 packets,
@@ -122,7 +123,7 @@
 [dup_pkt.parameters]
     bloom_capacity=1000000
     bloom_error_rate=0.00001
-    bloom_timeout=10
+    bloom_timeout=30
 
 
 [STREAM]
diff --git a/include/private/duplicate_pkt_distinguish.h b/include/private/duplicate_pkt_distinguish.h
index 9b768f5..9e9ceb0 100644
--- a/include/private/duplicate_pkt_distinguish.h
+++ b/include/private/duplicate_pkt_distinguish.h
@@ -14,12 +14,12 @@
 struct sapp_dup_key_ip4_tuple{
 	unsigned int sip;
 	unsigned int dip;
-};
+}__attribute__((packed, aligned(1)));
 
 struct sapp_dup_key_ip6_tuple{
 	struct in6_addr sip;
 	struct in6_addr dip;
-};
+}__attribute__((packed, aligned(1)));
 
 struct __sapp_dup_pkt_key_l4{
 	unsigned int tcp_seq;
diff --git a/include/private/sapp_global_val.h b/include/private/sapp_global_val.h
index f2b9564..8235ddc 100644
--- a/include/private/sapp_global_val.h
+++ b/include/private/sapp_global_val.h
@@ -201,7 +201,7 @@ typedef struct{
 
 
 typedef struct{
-	//int dup_pkt_distinguish_enable;
+	int kickout_udp_stream_enabled;
 	int dup_pkt_distinguish_all_inject;
 	int dup_pkt_distinguish_ipv4_tcp;
 	int dup_pkt_distinguish_ipv4_udp;
diff --git a/include/private/sapp_pkt_stat.h b/include/private/sapp_pkt_stat.h
index bc6e137..40f6b4b 100644
--- a/include/private/sapp_pkt_stat.h
+++ b/include/private/sapp_pkt_stat.h
@@ -189,6 +189,8 @@ typedef enum __sapp_sys_stat_type{
 	SAPP_STAT_UDP_OFFLOAD_PKTS,
 	SAPP_STAT_UDP_OFFLOAD_BYTES,	
 
+	SAPP_STAT_UDP_DUP_KICK_STREAM,
+
 	
 	/*************  memory stat  *******************/
 
diff --git a/src/config/config_parse.cpp b/src/config/config_parse.cpp
index a37e212..eb0849c 100644
--- a/src/config/config_parse.cpp
+++ b/src/config/config_parse.cpp
@@ -1726,6 +1726,7 @@ int sapp_parse_config(void)
 	
 
 	/**  duplicate packet config **/
+	tomlc99_wrap_load_int_def(ABBR_CFG_FILE_MAIN_ENTRY, (char *)"dup_pkt.traffic.original", (char *)"kickout_udp_stream_enabled", &pconfig->packet_io.dup_pkt_para.kickout_udp_stream_enabled, 0);
 	tomlc99_wrap_load_int_def(ABBR_CFG_FILE_MAIN_ENTRY, (char *)"dup_pkt.traffic.original", (char *)"original_ipv4_tcp_enabled", &pconfig->packet_io.dup_pkt_para.dup_pkt_distinguish_ipv4_tcp, 0);
 	tomlc99_wrap_load_int_def(ABBR_CFG_FILE_MAIN_ENTRY, (char *)"dup_pkt.traffic.original", (char *)"original_ipv4_udp_enabled", &pconfig->packet_io.dup_pkt_para.dup_pkt_distinguish_ipv4_udp, 0);
 	tomlc99_wrap_load_int_def(ABBR_CFG_FILE_MAIN_ENTRY, (char *)"dup_pkt.traffic.inject", (char *)"inject_all_enabled", &pconfig->packet_io.dup_pkt_para.dup_pkt_distinguish_all_inject, 0);
diff --git a/src/dealpkt/deal_udp.c b/src/dealpkt/deal_udp.c
index 1771e2f..36bc4b8 100644
--- a/src/dealpkt/deal_udp.c
+++ b/src/dealpkt/deal_udp.c
@@ -4,6 +4,7 @@
 20130530 :����ǿ��udp����̭���ܣ�
 */
 #include "sapp_api.h"
+#include "sapp_pkt_stat.h"
 #include "sapp_private_api.h"
 #include "sapp_declaration.h"
 
@@ -638,8 +639,16 @@ int dealipv4udppkt(struct streamindex *pindex, const struct mesa_ip4_hdr * this_
 		a_index =(struct streamindex * ) findstreamindex (pindex, raw_pkt);
 		if (unlikely(!a_index)){
 			//a_index = udp_add_new_stream (ptmp, udph);
-			a_index = udp_add_new_stream(pindex, udph, raw_pkt);
-		}
+            if(sapp_global_val->config.packet_io.dup_pkt_para.kickout_udp_stream_enabled)
+            {
+                if (sapp_dup_stream_search(pstream) == 1)
+                {
+					local_sys_stat->count[SAPP_STAT_UDP_DUP_KICK_STREAM]++;
+                    return PASS;
+                }
+            }
+            a_index = udp_add_new_stream(pindex, udph, raw_pkt);
+        }
 		//add by lqy ֻ�е�����δ����ʱ����lru
 		else 
 		{
@@ -886,7 +895,15 @@ int dealipv6udppkt(struct streamindex *pindex,const struct mesa_ip6_hdr *a_packe
 		a_index =(struct streamindex *)findstreamindex (pindex, raw_pkt);
 		if(unlikely(!a_index))
 		{
-			a_index = udp_add_new_stream(pindex, udph, raw_pkt);
+            if (sapp_global_val->config.packet_io.dup_pkt_para.kickout_udp_stream_enabled)
+            {
+                if (sapp_dup_stream_search(pstream) == 1)
+                {
+                    local_sys_stat->count[SAPP_STAT_UDP_DUP_KICK_STREAM]++;
+                    return PASS;
+                }
+            }
+            a_index = udp_add_new_stream(pindex, udph, raw_pkt);
 		}
 		else 
 		{
diff --git a/src/dealpkt/duplicate_pkt_distinguish.c b/src/dealpkt/duplicate_pkt_distinguish.c
index f1e142b..1f1ac87 100644
--- a/src/dealpkt/duplicate_pkt_distinguish.c
+++ b/src/dealpkt/duplicate_pkt_distinguish.c
@@ -102,7 +102,7 @@ static inline int sapp_set_dup_pkt_key(enum addr_type_t addr_type, enum stream_t
     }
     if (addr_type == ADDR_TYPE_IPV6)
     {
-        sapp_dup_pkt_key_v4_t *key_v6 = (sapp_dup_pkt_key_v4_t *)key;
+        sapp_dup_pkt_key_v6_t *key_v6 = (sapp_dup_pkt_key_v6_t *)key;
         const struct mesa_ip6_hdr *ip6hdr = (const struct mesa_ip6_hdr *)iphdr;
         memcpy(&key_v6->ip_addr.sip, &ip6hdr->ip6_src, sizeof(struct in6_addr));
         memcpy(&key_v6->ip_addr.dip, &ip6hdr->ip6_dst, sizeof(struct in6_addr));
@@ -163,7 +163,8 @@ int sapp_dup_pkt_init(void)
 	int i;
 	const sapp_dup_pkt_t *p_dup_conf = &sapp_global_val->config.packet_io.dup_pkt_para;
 	
-	if((0 == sapp_global_val->config.packet_io.dup_pkt_para.dup_pkt_distinguish_all_inject)
+	if( (0 == sapp_global_val->config.packet_io.dup_pkt_para.kickout_udp_stream_enabled)
+	  && (0 == sapp_global_val->config.packet_io.dup_pkt_para.dup_pkt_distinguish_all_inject)
 	  && (0 == sapp_global_val->config.packet_io.dup_pkt_para.dup_pkt_distinguish_ipv4_tcp)
 	  && (0 == sapp_global_val->config.packet_io.dup_pkt_para.dup_pkt_distinguish_ipv4_udp)){
 		return 0;
diff --git a/src/dealpkt/stream_manage.c b/src/dealpkt/stream_manage.c
index 5848c65..b6a85ea 100644
--- a/src/dealpkt/stream_manage.c
+++ b/src/dealpkt/stream_manage.c
@@ -338,6 +338,10 @@ void streamaddlist(struct streamindex *pindex,struct stream_list *plist)
 				pindex_to_del->stream.stream_public.ptcpdetail->createtime,
 				pindex_to_del->stream.stream_public.ptcpdetail->lastmtime,
 				g_CurrentTime);
+		if(pindex_to_del->stream.stream_public.type == STREAM_TYPE_UDP && sapp_global_val->config.packet_io.dup_pkt_para.kickout_udp_stream_enabled)
+		{
+			sapp_dup_stream_add(&pindex_to_del->stream.stream_public);
+		}
 		del_stream(pindex_to_del, timeout_reason);
 	}
 	
diff --git a/src/extensions/sapp_metrics.cpp b/src/extensions/sapp_metrics.cpp
index b21b9be..f9a8a02 100644
--- a/src/extensions/sapp_metrics.cpp
+++ b/src/extensions/sapp_metrics.cpp
@@ -369,6 +369,8 @@ static int sapp_fs2_init(sapp_global_t *global_paramters)
 	p_fs2_rt->fs_id_count_array[SAPP_STAT_UDP_OFFLOAD_STREAM] = FS_register(fs2_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "Udp_Offload_STM");	
 	p_fs2_rt->fs_id_count_array[SAPP_STAT_UDP_OFFLOAD_PKTS] = FS_register(fs2_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "Udp_Offload_Pkt");
 	p_fs2_rt->fs_id_length_array[SAPP_STAT_UDP_OFFLOAD_BYTES] = FS_register(fs2_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "Udp_Offload_Bit");
+
+	p_fs2_rt->fs_id_count_array[SAPP_STAT_UDP_DUP_KICK_STREAM] = FS_register(fs2_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "Udp_DUP_STM");
 	
 	return 0;
 }
@@ -609,6 +611,9 @@ static int sapp_fs3_init(sapp_global_t *global_paramters)
 								NULL, 0);	
 	fs3_rt->metrics_ids[1][SAPP_STAT_UDP_OFFLOAD_BYTES] = fieldstat_register(fs3_handle, FIELD_TYPE_COUNTER, "Udp_Offload_Bit", 
 								NULL, 0);	
+								
+	fs3_rt->metrics_ids[0][SAPP_STAT_UDP_DUP_KICK_STREAM] = fieldstat_register(fs3_handle, FIELD_TYPE_COUNTER, "Udp_DUP_STM", 
+								NULL, 0);									
 
 	return 0;
 }
diff --git a/src/packet_io/packet_io_status.cpp b/src/packet_io/packet_io_status.cpp
index aa1aa6c..bae92e4 100644
--- a/src/packet_io/packet_io_status.cpp
+++ b/src/packet_io/packet_io_status.cpp
@@ -126,6 +126,8 @@ static void packet_io_update_metrics_fs2(void)
 	sapp_fs2_update_count(SAPP_STAT_UDP_OFFLOAD_STREAM,  cur_count[SAPP_STAT_UDP_OFFLOAD_STREAM]);
 	sapp_fs2_update_count(SAPP_STAT_UDP_OFFLOAD_PKTS,  cur_count[SAPP_STAT_UDP_OFFLOAD_PKTS]);
 	sapp_fs2_update_length(SAPP_STAT_UDP_OFFLOAD_BYTES,  cur_length[SAPP_STAT_UDP_OFFLOAD_BYTES]);
+
+	sapp_fs2_update_count(SAPP_STAT_UDP_DUP_KICK_STREAM,  cur_count[SAPP_STAT_UDP_DUP_KICK_STREAM]);
 }
 
 static void packet_io_update_metrics_fs3(void)
@@ -223,6 +225,9 @@ static void packet_io_update_metrics_fs3(void)
 	sapp_fs3_metric_incrby(0, SAPP_STAT_UDP_OFFLOAD_STREAM,cur_count[SAPP_STAT_UDP_OFFLOAD_STREAM]-history_count[SAPP_STAT_UDP_OFFLOAD_STREAM]);
 	sapp_fs3_metric_incrby(0, SAPP_STAT_UDP_OFFLOAD_PKTS,cur_count[SAPP_STAT_UDP_OFFLOAD_PKTS]-history_count[SAPP_STAT_UDP_OFFLOAD_PKTS]);
 	sapp_fs3_metric_incrby(0, SAPP_STAT_UDP_OFFLOAD_BYTES,cur_count[SAPP_STAT_UDP_OFFLOAD_BYTES]-history_count[SAPP_STAT_UDP_OFFLOAD_BYTES]);
+
+
+	sapp_fs3_metric_incrby(0, SAPP_STAT_UDP_DUP_KICK_STREAM,cur_count[SAPP_STAT_UDP_DUP_KICK_STREAM]-history_count[SAPP_STAT_UDP_DUP_KICK_STREAM]);
 	
 	return;
 }
diff --git a/src/support/dablooms/src/dablooms.c b/src/support/dablooms/src/dablooms.c
index 6bc9326..7c7dd0a 100755
--- a/src/support/dablooms/src/dablooms.c
+++ b/src/support/dablooms/src/dablooms.c
@@ -524,7 +524,7 @@ error_out:
 }
 
 int expiry_dablooms_element_count_get(struct expiry_dablooms_handle *handle, uint64_t *count){
-    if(handle->cur_bloom == NULL){
+    if(handle == NULL || handle->cur_bloom == NULL){
         return EXPIRY_DABLOOMS_ERRNO_BLOOM_NULL;
     }
     *count = handle->cur_bloom_inc_id;
@@ -532,7 +532,7 @@ int expiry_dablooms_element_count_get(struct expiry_dablooms_handle *handle, uin
 }
 
 static int bloom_expired_check(struct expiry_dablooms_handle *handle, time_t cur_time){
-    if(handle->cur_bloom == NULL){
+    if(handle == NULL || handle->cur_bloom == NULL){
         return EXPIRY_DABLOOMS_ERRNO_BLOOM_NULL;
     }
     time_t delta_time = cur_time - handle->cur_bloom_start;
@@ -559,15 +559,16 @@ static int bloom_expired_check(struct expiry_dablooms_handle *handle, time_t cur
 }
 
 int expiry_dablooms_add(struct expiry_dablooms_handle *handle, const char *key, size_t len,  time_t cur_time){
+    if(key==NULL || len ==0 || handle == NULL)
+    {
+        return -1;
+    }
     int ret = bloom_expired_check(handle, cur_time);
     if(ret < 0)
     {
         return ret;
     }
-    if(key==NULL || len ==0)
-    {
-        return -1;
-    }
+
     scaling_bloom_add(handle->cur_bloom, key, len, handle->cur_bloom_inc_id);
     handle->cur_bloom_inc_id++;
     time_t delta_time = cur_time - handle->cur_bloom_start;
@@ -589,15 +590,15 @@ int expiry_dablooms_add(struct expiry_dablooms_handle *handle, const char *key,
 }
 
 int expiry_dablooms_search(struct expiry_dablooms_handle *handle, const char *key, size_t len,  time_t cur_time){
+    if(key==NULL || len ==0 || handle == NULL)
+    {
+        return -1;
+    }
     int ret = bloom_expired_check(handle, cur_time);
     if(ret < 0)
     {
         return ret;
     }
-    if(key == NULL || len == 0)
-    {
-        return 0;
-    }
     int bloom_hit = scaling_bloom_check(handle->cur_bloom, key, len);
     return bloom_hit;
 }