🧪 test(test_app_sapp): 增加timed测试dropme逻辑

6 files changed, 86 insertions, 61 deletions

diff --git a/src/config/cmd_args.c b/src/config/cmd_args.c
index ede0122..76de6fa 100644
--- a/src/config/cmd_args.c
+++ b/src/config/cmd_args.c
@@ -33,7 +33,6 @@ static char *popen_result_store(FILE *fp)
 	int chr;
 	char *exec_result;
 	int exec_result_size = 0;
-	char line_buf[PATH_MAX];
 
 	exec_result = (char *)sapp_mem_calloc(SAPP_MEM_FIX_GLOBAL_VAL, MEM_STAT_GLOBAL_THREAD_ID, 1);
 	
@@ -86,11 +85,9 @@ static int get_dumpfile_list_array(const char *opt_arg)
 	tobe_free_result = exec_result;
 	sapp_global_val->cla.dumpfie_list_array = NULL;
 	
-	while(ptr_cont = strtok_r(exec_result, delim, &save_ptr)){
+	while((ptr_cont = strtok_r(exec_result, delim, &save_ptr))){
 		sapp_global_val->cla.dumpfie_list_array = (char **)sapp_mem_realloc(SAPP_MEM_FIX_GLOBAL_VAL, MEM_STAT_GLOBAL_THREAD_ID,sapp_global_val->cla.dumpfie_list_array, sizeof(void *) * (line_num + 2));//��ǰҪ�洢��ָ���ټ���EOF
 		sapp_global_val->cla.dumpfie_list_array[line_num] = sapp_strdup(ptr_cont);;
-		
-		//printf("#### %d: result_content:%s\n", line_num, sapp_global_val->cla.dumpfie_list_array[line_num]);
 		line_num++;
 		exec_result = NULL;
 	}
@@ -374,7 +371,6 @@ void sapp_update_main_config_file(const char *main_entry_cfg)
 
 void sapp_update_config_root_dir(const char *cfg_root_dir)
 {
-	char sapp_cfg_filename[PATH_MAX];
 	sapp_config_file_link_t *pconfig;
 
 	pconfig = &sapp_global_val->config.cfg_file_path;
@@ -435,19 +431,15 @@ void sapp_update_data_root_dir(const char *data_root_dir)
 
 static void sapp_close_stdout(void)
 {
-#if 0
-	int null_fd = open("/dev/null", O_CREAT | O_RDWR); /* ����sappˢ����Ϣ */
-	if(null_fd > 0){
-		dup2(null_fd, STDOUT_FILENO);
-	}
-#else
 	if(stdout){
 		fclose(stdout);
 		stdout = fopen("/dev/null", "w");
 	}
-#endif	
 }
 
+
+extern int sapp_config_check(void);
+
 int sapp_parse_cmd_args(int argc, char *argv[])
 {
 	int ret = 0;
diff --git a/src/dealpkt/plug_support.c b/src/dealpkt/plug_support.c
index 03a2ab8..6e11550 100644
--- a/src/dealpkt/plug_support.c
+++ b/src/dealpkt/plug_support.c
@@ -333,8 +333,8 @@ int get_rawpkt_opt_from_streaminfo(const struct streaminfo *pstream, int type, v
 /*Convert tuple4 to string. Format: 10.0.0.1, 1234->10.0.0.2,5678*/
 const char *printaddr (const struct layer_addr *paddrinfo,int threadindex)
 {
-	static char maxbuf[MAX_THREAD_NUM][128];
-	char *buf=maxbuf[threadindex];
+	static const char maxbuf[MAX_THREAD_NUM][128];
+	char *buf=(char*)maxbuf[threadindex];
 	char ip_str[INET6_ADDRSTRLEN];
 	struct stream_tuple4_v4 *paddr;
 	struct stream_tuple4_v6 *paddr6;
@@ -889,7 +889,7 @@ static int sapp_get_vxlan_info_from_streaminfo(const struct streaminfo *pstream,
 
 	memset(vxinfo, 0, sizeof(struct vxlan_info));
 
-	if(ADDR_TYPE_MAC_IN_MAC == mim_stream->addr.addrtype)
+    if (ADDR_TYPE_MAC_IN_MAC == mim_stream->addr.addrtype)
     {
         mim_addr = mim_stream->addr.mimac;
         sapp_mac_addr_to_long(mim_addr->outer_src_mac, &smac_integer_type);
@@ -899,7 +899,7 @@ static int sapp_get_vxlan_info_from_streaminfo(const struct streaminfo *pstream,
         mac_addr = mim_stream->addr.mac;
         sapp_mac_addr_to_long(mac_addr->src_addr.h_source, &smac_integer_type);
     }
-	vxinfo->encap_type = (smac_integer_type & GDEV_SMAC_MASK_ENCAP_TYPE) >> 16;
+    vxinfo->encap_type = (smac_integer_type & GDEV_SMAC_MASK_ENCAP_TYPE) >> 16;
 	vxinfo->entrance_id = (smac_integer_type & GDEV_SMAC_MASK_ENTRANCE_ID) >> 10;
 	vxinfo->dev_id = (smac_integer_type & GDEV_SMAC_MASK_DEV_ID) >> 4;
 	vxinfo->link_id = (smac_integer_type & GDEV_SMAC_MASK_LINK_ID ) >> 1;
@@ -959,7 +959,7 @@ static void *MESA_get_stream_plug_pme_from_platform_entry(const StreamFunInfo *p
 /*
 	���ݲ����ں���������, ��ȡ�����˽�д洢�ռ�ָ��pme.
 */
-static int MESA_get_stream_plug_pme(const struct streaminfo *pstream, void *opt_val, int *opt_val_len)
+static int MESA_get_stream_plug_pme(const struct streaminfo *pstream, void *opt_val, const int *opt_val_len)
 {
 	int ret = 0;
 	struct mso_plug_pme *in_arg;
@@ -1317,7 +1317,7 @@ int MESA_get_stream_opt(const struct streaminfo *pstream, enum MESA_stream_opt o
 			pstream_pr = (const struct streaminfo_private *)pstream;
 			unsigned long long *out_args = (unsigned long long *)opt_val;
 			unsigned long long dev_id = *out_args;
-			if(dev_id < 0 || dev_id > 4095){
+			if(dev_id > 4095){
 				sapp_runtime_log(RLOG_LV_INFO, "%s,MESA_get_stream_opt() MSO_GLOBAL_STREAM_ID error:invalid dev_id:%llu!\n", printaddr(&pstream->addr, pstream->threadnum), dev_id);
 				return -1;
 			}
diff --git a/src/packet_io/under_ddos.cpp b/src/packet_io/under_ddos.cpp
index d6aff8f..c57ac68 100644
--- a/src/packet_io/under_ddos.cpp
+++ b/src/packet_io/under_ddos.cpp
@@ -226,6 +226,15 @@ void packet_io_under_ddos_run(void)
 	}
 }
 
+void packet_io_under_ddos_limit_arg_free(cpu_limit_handle h, void *user_arg)
+{
+	if(user_arg)
+	{
+		free(user_arg);
+	}
+	return;
+}
+
 int packet_io_under_ddos_init(void)
 {
 	sapp_under_ddos_config_t *p_ddos_cfg = &sapp_global_val->config.packet_io.under_ddos_config;
@@ -262,6 +271,7 @@ int packet_io_under_ddos_init(void)
 	cpu_limit_set_opt(ud_handle, CL_OPT_FACTOR_GET_FUN, (void *)&sapp_get_create_stream_rate_cb, sizeof(void *));	
 	under_sapp_user_args_t *ud_user_arg = (under_sapp_user_args_t *)calloc(1, sizeof(under_sapp_user_args_t));
 	cpu_limit_set_opt(ud_handle, CL_OPT_USER_ARG, ud_user_arg, sizeof(void *));	
+	cpu_limit_set_opt(ud_handle, CL_OPT_DESTROY_CALLBCAK_FUN, (void *)packet_io_under_ddos_limit_arg_free, sizeof(void *));	
 	cpu_limit_set_opt(ud_handle, CL_OPT_DEBUG_LOG_FILE_NAME, (void *)sapp_global_val->config.data_file_path.data_under_ddos_stat_log_absolute, strlen(sapp_global_val->config.data_file_path.data_under_ddos_stat_log_absolute));		
 
 	sapp_global_val->individual_fixed.under_ddos_handle = ud_handle;
diff --git a/src/sapp_dev/sapp_plug.c b/src/sapp_dev/sapp_plug.c
index 46e997c..b85e1af 100644
--- a/src/sapp_dev/sapp_plug.c
+++ b/src/sapp_dev/sapp_plug.c
@@ -152,10 +152,11 @@ static void *sapp_monitor_thread(void *args)
 
 	sapp_runtime_log(RLOG_LV_FATAL, "sapp_monitor_thread  begin libsapp_destroy_env!\n");
 	libsapp_destroy_env();
-
+	pthread_exit(NULL);
 	return NULL;
 }
 
+extern void sapp_update_main_config_file(const char *main_entry_cfg);
 
 int libsapp_setup_env(int argc, char *argv[])
 {
@@ -196,11 +197,6 @@ int libsapp_setup_env(int argc, char *argv[])
 	printf("\033[33m[Warning]This sapp is a temp version for solve confounded bug!\033[0m\n");
 #endif
 	
-#if COMPAT_PAPP_FOR_BENCHMARK
-	printf("\033[33m[Warning]This sapp is a emasculate version for compare with papp, in other word, it's papp!\033[0m\n");
-	sleep(1);
-#endif
-
 	sapp_init_breakpad_mini();
 	signal_take_over();
 	
@@ -238,7 +234,7 @@ static void wait_for_all_io_threads(void)
 	sapp_timer_destroy(sapp_global_val->individual_fixed.sapp_standalone_timer);
 }
 
-
+extern void packet_io_clean_thread_context(int thread_seq);
 
 void libsapp_destroy_env_per_thread(int tseq)
 {
@@ -256,7 +252,8 @@ void libsapp_destroy_env_per_thread(int tseq)
 }
 
 
-
+extern void packet_io_stop(void);
+extern void packet_io_exit(void);
 
 void libsapp_destroy_env(void)
 {
diff --git a/src/support/cpu_limit/cpu_limit.c b/src/support/cpu_limit/cpu_limit.c
index d1a3538..8ffd13d 100644
--- a/src/support/cpu_limit/cpu_limit.c
+++ b/src/support/cpu_limit/cpu_limit.c
@@ -365,14 +365,14 @@ static void cl_debug_log(cpu_limit_inner_t *h)
 			thread_bypass_state++;
 		}
 		if(CL_LIMIT_STATE_PENDING == this_stat->op_state){
-			fprintf(fp, "%9d %7s %9.2f %15ld %18ld %15s\n", tid, 
+			fprintf(fp, "%9d %7s %9.2f %15lld %18lld %15s\n", tid, 
 					cl_op_state_ntop(this_stat->op_state), 
 					this_stat->realtime_res_val, 
 					this_stat->debug_factor_work_per_second + this_stat->debug_factor_bypass_per_second,
 					this_stat->debug_factor_bypass_per_second,
 					"infinity"/*无限制*/);
 		}else{
-			fprintf(fp, "%9d %7s %9.2f %15ld %18ld %15.2f\n", tid, 
+			fprintf(fp, "%9d %7s %9.2f %15lld %18lld %15.2f\n", tid, 
 					cl_op_state_ntop(this_stat->op_state), 
 					this_stat->realtime_res_val, 
 					this_stat->debug_factor_work_per_second + this_stat->debug_factor_bypass_per_second,
@@ -424,6 +424,7 @@ static inline long long _cl_get_current_time_in_ms(void)
 	return tms;
 }
 
+extern int sapp_usleep(int usec);
 
 /* 后台实时统计分析线程, 不占用sapp IO线程的时间片 */
 static void *cl_bg_thread(void *_arg_h)
@@ -552,21 +553,24 @@ int cpu_limit_get_opt(cpu_limit_handle _arg_h, cpu_limit_opt opt, void *value, i
 
 	if(NULL == _arg_h || NULL == value || NULL == value_len){
 		return -1;
-	}	
-
-	switch(opt){
-		case CL_OPT_GLOBAL_BYPASS_STATE:
-		{
-			if(h->running_state != CL_STATE_RUNNING){
-				return -1;
-			}
-			int *out_value = (int *)value;
-			*out_value = h->global_bypass_state;
-		}
-		break;
 	}
 
-	return ret;
+    switch (opt)
+    {
+    case CL_OPT_GLOBAL_BYPASS_STATE: {
+        if (h->running_state != CL_STATE_RUNNING)
+        {
+            return -1;
+        }
+        int *out_value = (int *)value;
+        *out_value = h->global_bypass_state;
+    }
+    break;
+    default:
+        break;
+    }
+
+    return ret;
 }
 
 cpu_limit_handle cpu_limit_create(void)
@@ -641,12 +645,22 @@ void cpu_limit_destroy(cpu_limit_handle _arg_h)
 	
 	h->running_state = CL_STATE_TO_BE_DESTORY;
 	pthread_join(h->thread_id, NULL);
-	
-	if(h->destroy_cb_fun){
+	int i;
+    for (i = 0; i < h->thread_count; i++)
+    {
+        free(h->mthread_stat[i].recent_smooth_res_value);
+        h->mthread_stat[i].recent_smooth_index = 0;
+    }
+	if(h->debug_log_filename)
+	{
+		free(h->debug_log_filename);
+	}
+
+    if(h->destroy_cb_fun){
 		h->destroy_cb_fun(_arg_h, h->user_arg);
 	}
 
-	memset(h, 0xFE, sizeof(cpu_limit_inner_t));
+	//memset(h, 0xFE, sizeof(cpu_limit_inner_t));
 	
 	free(h);
 	
diff --git a/test/test_app_sapp.c b/test/test_app_sapp.c
index ed417cc..7b6e35a 100644
--- a/test/test_app_sapp.c
+++ b/test/test_app_sapp.c
@@ -2,6 +2,7 @@
 #include "sapp_api.h"
 #include "sapp_private_api.h"
 #include "gdev_keepalive.h"
+#include "stream_base.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -498,23 +499,34 @@ char test_set_stream_timed(struct streaminfo *pstream,  void **pme, int thread_s
 {
 	int ret;
 	int tout_val;
-	
-	if(pstream->pktstate== OP_STATE_PENDING){
-			tout_val = 1;
-			ret = MESA_set_stream_opt(pstream, MSO_STREAM_TIMER, &tout_val, sizeof(int));
-			if(ret < 0){
-				DPRINT("stream:%p, MESA_set_stream_opt error:\n", pstream);
-				return APP_STATE_DROPME;
-			}			
-			DPRINT("stream: %p %s set timer at %ld!\n", pstream, printaddr(&pstream->addr, thread_seq), time(NULL));
-	}
-	if(pstream->pktstate == OP_STATE_TIMED){
-		DPRINT("stream timeup: %p %s at %ld!\n", pstream, printaddr(&pstream->addr, thread_seq), time(NULL));
-	}
-	if(pstream->pktstate == OP_STATE_CLOSE){
-		DPRINT("stream close: %p %s at %ld!\n", pstream, printaddr(&pstream->addr, thread_seq), time(NULL));
-	}
-	return APP_STATE_GIVEME;	
+	static int timed_cnt=0;
+
+    if (pstream->pktstate == OP_STATE_PENDING)
+    {
+        tout_val = 1;
+        ret = MESA_set_stream_opt(pstream, MSO_STREAM_TIMER, &tout_val, sizeof(int));
+        if (ret < 0)
+        {
+            DPRINT("stream:%p, MESA_set_stream_opt error:\n", pstream);
+            return APP_STATE_DROPME;
+        }
+        DPRINT("stream: %p %s set timer at %ld!\n", pstream, printaddr(&pstream->addr, thread_seq), time(NULL));
+    }
+    if (pstream->pktstate == OP_STATE_TIMED)
+    {
+		timed_cnt+=1;
+        DPRINT("stream timed callback: %p %s, cnt=%d at %ld!\n", pstream, printaddr(&pstream->addr, thread_seq), timed_cnt, time(NULL));
+		if(timed_cnt >= 5)
+		{
+        	DPRINT("stream timed callback: %p %s, cnt=%d >= 5, dropme!\n", pstream, printaddr(&pstream->addr, thread_seq), timed_cnt);
+			return APP_STATE_DROPME;	
+		}
+    }
+    if (pstream->pktstate == OP_STATE_CLOSE)
+    {
+        DPRINT("stream close: %p %s at %ld!\n", pstream, printaddr(&pstream->addr, thread_seq), time(NULL));
+    }
+    return APP_STATE_GIVEME;	
 }
 
 char test_sapp_get_platform_opt(struct streaminfo *pstream,  void **pme, int thread_seq,void *a_packet)