Stop threads after packets finish.

Author: caesar0301

.gitignore

@@ -1,4 +1,6 @@
 *.a
 *.dblite
 lib/*
-*.txt
+*.txt
+*.o
+*~

CHANGES

@@ -0,0 +1,6 @@
+TODO List
+------------
+
+1. Add tcpdump filter to packet preprocessing method.
+
+2. Optimize flow reordering methods.

COPYING

@@ -1,7 +1,7 @@
 SConscript(['./flow/SConscript',
-			'./http/SConscript',
-			'./packet/SConscript',
-			'./tcp/SConscript'])
+            './http/SConscript',
+            './packet/SConscript',
+            './tcp/SConscript'])
 
 env = Environment(CCFLAGS='-w')
 # Library paths
@@ -14,24 +14,24 @@ cpp_path=['../include']
 # Flag debug to decide if NFM libs are used to compile program
 debug = ARGUMENTS.get('debug', 0)
 if int(debug):
-	# NFM library names
-	nfm_libs = ['nfm', 'nfm_framework', 'nfm_error', 'nfm_packet', 'nfm_rules', 'nfm_platform', 'nfe', 'nfp']
-	libs += nfm_libs
-	# NFM library path
-	lib_path.append('/opt/netronome/lib')
-	# NFM header file path
-	cpp_path.append('/opt/netronome/nfm/include')
+    # NFM library names
+    nfm_libs = ['nfm', 'nfm_framework', 'nfm_error', 'nfm_packet', 'nfm_rules', 'nfm_platform', 'nfe', 'nfp']
+    libs += nfm_libs
+    # NFM library path
+    lib_path.append('/opt/netronome/lib')
+    # NFM header file path
+    cpp_path.append('/opt/netronome/nfm/include')
 
 # Compile the programs
-env.Program(target = '../bin/weblogger', 
-			source = Glob('*.c'),
-			LIBPATH = lib_path,
-			LIBS = libs,
-			CPPPATH = cpp_path)
+env.Program(target = '../bin/http-sniffer', 
+            source = Glob('*.c'),
+            LIBPATH = lib_path,
+            LIBS = libs,
+            CPPPATH = cpp_path)
 
 env.Program(target = '../bin/tracedump', 
-			source = './tracedump/tracedump.c',
-			LIBS = libs,
-			LIBPATH = lib_path,
-			CPPPATH = cpp_path)
-		
+            source = './tracedump/tracedump.c',
+            LIBS = libs,
+            LIBPATH = lib_path,
+            CPPPATH = cpp_path)
+        

INSTALL

@@ -154,6 +154,7 @@ capture_main(const char* interface, void (*pkt_handler)(void*))
 	struct pcap_pkthdr pkthdr;
 	packet_t *packet = NULL;
 
+	printf("Online mode ...\n");
 	cap = pcap_open_live(interface, 65535, 0, 1000, errbuf);
 	if( cap == NULL){
 		printf("%s\n",errbuf);
@@ -170,6 +171,10 @@ capture_main(const char* interface, void (*pkt_handler)(void*))
 		}
 		pkt_handler(packet);
 	}
+
+	if( cap != NULL){
+		pcap_close(cap);
+	}
 	return 0;
 }
 
@@ -183,22 +188,36 @@ capture_offline(const char* filename, void (*pkt_handler)(void*))
 	pcap_t *cap = NULL;
 	struct pcap_pkthdr pkthdr;
 	packet_t *packet = NULL;
+	extern int GP_CAP_FIN;
 
+	// Open handler
+	printf("Offline mode ...\n");
 	cap = pcap_open_offline(filename, errbuf);
 	if( cap == NULL){
 		printf("%s\n",errbuf);
 		exit(1);
 	}
+
+	// Run the loop
 	while(1){
 		raw = pcap_next(cap, &pkthdr);
 		if( raw == NULL){
-			continue;
+			// No more packets?
+			GP_CAP_FIN = 1;
+			break;
 		}
 		packet = packet_preprocess(raw, &pkthdr);
 		if( NULL == packet ){
 			continue;
 		}
+		// Handle the packet
 		pkt_handler(packet);
 	}
+
+	// Close the handler
+	if( cap != NULL){
+		printf("Close handler!\n");
+		pcap_close(cap);
+	}
 	return 0;
 }

TODOS

@@ -834,7 +834,8 @@ flow_dump_file_json(const flow_t *flow, const char* file)
 	struct tm *timeinfo = NULL;
 	time( &raw_time );
 	timeinfo = localtime( &raw_time );
-	strftime(time_buf, sizeof(time_buf), "%Y%m%d %H:%M:%S", timeinfo);
+	// ISO 8601 time format
+	strftime(time_buf, sizeof(time_buf), "%Y-%m-%dT%H:%M:%S", timeinfo);
 	json_object_object_add(new_flow, "t_r", json_object_new_string(time_buf));
 	/*Convert IP addr */
 	char *saddr = malloc(sizeof("aaa.bbb.ccc.ddd"));

src/SConscript

@@ -244,12 +244,12 @@ flow_hash_scnt(void)
 }
 
 /*
- * Close a flow forcedly if the delta time is lower than timeslot.
+ * Close a flow forcedly if the delta time is lower than timeout.
  * Then add the flow to flow queue.
  * Return the number of flows deleted forcedly.
  */
 int 
-flow_scrubber(const int timeslot)
+flow_scrubber(const int timeout)
 {
 	int i = 0;
 	unsigned long delta = 0;
@@ -268,7 +268,7 @@ flow_scrubber(const int timeslot)
 			gettimeofday(&tv, &tz);
 			delta = abs(tv.tv_sec - flow->last_action_sec);
 
-			if (delta > timeslot){
+			if (delta > timeout){
 				num++;
 				flow->close = FORCED_CLOSE;	// Close flow forcedly.
 				flow_queue_enq(flow_hash_delete(flow));

src/capture.c

@@ -8,6 +8,8 @@
 #include "packet.h"
 #include "flow.h"
 
+extern int GP_CAP_FIN;
+
 /* Fetch a packet from packet queue and add it to any flow. */
 int 
 process_packet_queue(void)
@@ -16,45 +18,34 @@ process_packet_queue(void)
 	while(1)
 	{
 		pkt = packet_queue_deq();
-		if (pkt != NULL)
-		{
+		if (pkt != NULL){
 			flow_hash_add_packet(pkt);
 			continue;
+		} else if ( GP_CAP_FIN == 1 ){
+			// job done!
+			break;
 		}
 	}
 	pthread_exit("Packet processing finished.\n");
 	return 0;
 }
 
-/* Scrub flow hash table to forcely close dead flows.*/
-void 
-scrubbing_flow_htbl(void)
-{
-	int num = 0;
-	while(1){
-		sleep(1);
-		num = flow_scrubber(60);	/* seconds */
-	}
-	pthread_exit("Flow processing finished.\n");
-	return 0;
-}
-
-char*
-gen_filename(void){
-	char file_name_buf[64];
-	memset(file_name_buf, 0, sizeof(file_name_buf));
+// char*
+// gen_filename(void){
+// 	char file_name_buf[64];
+// 	memset(file_name_buf, 0, sizeof(file_name_buf));
 
-	time_t raw_time;
-	struct tm *timeinfo = NULL;
-	char time_buf[20];
-	memset(time_buf, 0, sizeof(time_buf));
-	time( &raw_time );
-	timeinfo = localtime( &raw_time );
-	strftime(time_buf, sizeof(time_buf), "%Y%m%d%H", timeinfo);
-	strcat(file_name_buf, time_buf);
-	strcat(file_name_buf, ".txt");
-	return file_name_buf;
-}
+// 	time_t raw_time;
+// 	struct tm *timeinfo = NULL;
+// 	char time_buf[20];
+// 	memset(time_buf, 0, sizeof(time_buf));
+// 	time( &raw_time );
+// 	timeinfo = localtime( &raw_time );
+// 	strftime(time_buf, sizeof(time_buf), "%Y%m%d%H", timeinfo);
+// 	strcat(file_name_buf, time_buf);
+// 	strcat(file_name_buf, ".txt");
+// 	return file_name_buf;
+// }
 
 /* Fetch a flow from flow queue and process it */
 int 
@@ -68,6 +59,7 @@ process_flow_queue(const char* dump_file)
 		if (dump_file != NULL){
 			file_name = dump_file;
 		}
+
 		flow = flow_queue_deq();
 		if(flow != NULL){
 			flow_extract_http(flow);
@@ -90,24 +82,49 @@ process_flow_queue(const char* dump_file)
 			flow_print(flow);
 			flow_free(flow);
 			continue;
+		} else if ( GP_CAP_FIN == 1 ) {
+			// job done!
+			break;
 		}
 	}
 	pthread_exit("Flow processing finished.\n");
 	return 0;
 }
 
+
+/* Scrub flow hash table to forcely close dead flows.*/
+void 
+scrubbing_flow_htbl(void)
+{
+	int num = 0;
+	while(1){
+		sleep(1);
+		if ( GP_CAP_FIN == 0 ){
+			num = flow_scrubber(60*10);	// seconds, flow timeout
+		} else {
+			num = flow_scrubber(-1); // all flows
+			break;
+		}
+	}
+	pthread_exit(NULL);
+	return 0;
+}
+
+
 /* for debugging */
 void 
 debugging_print(void)
 {
 	while(1)
 	{
-		sleep(3);
-		printf("...................................\n");
-		packet_queue_print();
-		flow_hash_print();
-		flow_queue_print();
+		if ( GP_CAP_FIN == 1 ) break;
+		else {
+			//packet_queue_print();
+			flow_hash_print();
+			//flow_queue_print();
+			sleep(1);
+		}
 	}
-	pthread_exit("Flow processing finished.\n");
+	pthread_exit(NULL);
 	return 0;
 }

src/flow/flow.c

@@ -18,6 +18,7 @@ print_usage(const char* pro_name)
 	printf("    or %s -f tracefile [-o dumpfile]\n", pro_name);
 }
 
+int GP_CAP_FIN = 0;
 int main(int argc, char *argv[]){
 	char* interface = NULL;
 	char* dumpfile = NULL;
@@ -61,18 +62,20 @@ int main(int argc, char *argv[]){
 	void *thread_result;
 	packet_queue_init();	/* Initialize packet queue */
 	flow_init();			/* Initialize flow queue and hashtable */
+	// Start backend jobs defined in jobs.c
 	pthread_create(&j_pkt_q, NULL, (void*)process_packet_queue, NULL);
-	pthread_create(&j_debug_p, NULL, (void*)debugging_print, NULL);
-	pthread_create(&j_flow_q, NULL, (void*)process_flow_queue, dumpfile);
 	pthread_create(&j_scrb_htbl, NULL, (void*)scrubbing_flow_htbl, NULL);
-	
+	pthread_create(&j_flow_q, NULL, (void*)process_flow_queue, dumpfile);
+	// For debugging
+	pthread_create(&j_debug_p, NULL, (void*)debugging_print, NULL);
+	// Start main capture in live or offline mode
 	if (interface != NULL){
 		capture_main(interface, packet_queue_enq);
 	}else	//tracefile != NULL
 	{
 		capture_offline(tracefile, packet_queue_enq);
 	}
-
+	// Wait the threads to end
 	pthread_join(j_pkt_q, &thread_result);
 	pthread_join(j_debug_p, &thread_result);
 	pthread_join(j_flow_q, &thread_result);