Split src2dst / dst2src traffic

4 files changed, 31 insertions, 21 deletions

diff --git a/example/Makefile.am b/example/Makefile.am
index 1a0570b1f..1a2bf453d 100644
--- a/example/Makefile.am
+++ b/example/Makefile.am
@@ -3,7 +3,7 @@ bin_PROGRAMS = ndpiReader
 AM_CPPFLAGS = -I$(top_srcdir)/src/include @PCAP_INC@
 AM_CFLAGS = @PTHREAD_CFLAGS@ # --coverage
 
-LDADD = $(top_builddir)/src/lib/libndpi.la @JSON_C_LIB@ @PTHREAD_LIBS@ @PCAP_LIB@ @DL_LIB@
+LDADD = $(top_builddir)/src/lib/libndpi.la @JSON_C_LIB@ @PTHREAD_LIBS@ @PCAP_LIB@ @DL_LIB@ -lm
 AM_LDFLAGS = -static @DL_LIB@
 
 ndpiReader_SOURCES = ndpiReader.c ndpi_util.c ndpi_util.h
diff --git a/example/ndpiReader.c b/example/ndpiReader.c
index 6aed8ec88..6346dca01 100644
--- a/example/ndpiReader.c
+++ b/example/ndpiReader.c
@@ -42,6 +42,7 @@
 #include <pthread.h>
 #include <sys/socket.h>
 #include <assert.h>
+#include <math.h>
 #include "../config.h"
 #include "ndpi_api.h"
 #include "uthash.h"
@@ -662,8 +663,10 @@ static void printFlow(u_int16_t thread_id, struct ndpi_flow_info *flow) {
 	      flow->detected_protocol.app_protocol,
 	      ndpi_get_proto_name(ndpi_thread_info[thread_id].workflow->ndpi_struct, flow->detected_protocol.app_protocol));
 
-    fprintf(out, "[%u pkts/%llu bytes]",
-	    flow->packets, (long long unsigned int) flow->bytes);
+    fprintf(out, "[%u pkts/%llu bytes ", flow->src2dst_packets, (long long unsigned int) flow->src2dst_bytes);
+    fprintf(out, "%s %u pkts/%llu bytes]",
+	    (flow->dst2src_packets > 0) ? "<->" : "->",
+	    flow->dst2src_packets, (long long unsigned int) flow->dst2src_bytes);
 
     if(flow->host_server_name[0] != '\0') fprintf(out, "[Host: %s]", flow->host_server_name);
     if(flow->info[0] != '\0') fprintf(out, "[%s]", flow->info);
@@ -702,8 +705,8 @@ static void printFlow(u_int16_t thread_id, struct ndpi_flow_info *flow) {
 			     json_object_new_string(ndpi_get_proto_name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
 									flow->detected_protocol.app_protocol)));
 
-    json_object_object_add(jObj,"packets",json_object_new_int(flow->packets));
-    json_object_object_add(jObj,"bytes",json_object_new_int(flow->bytes));
+    json_object_object_add(jObj,"packets",json_object_new_int(flow->src2dst_packets + flow->dst2src_packets));
+    json_object_object_add(jObj,"bytes",json_object_new_int(flow->src2dst_bytes + flow->dst2src_bytes));
 
     if(flow->host_server_name[0] != '\0')
       json_object_object_add(jObj,"host.server.name",json_object_new_string(flow->host_server_name));
@@ -796,8 +799,8 @@ static void node_proto_guess_walker(const void *node, ndpi_VISIT which, int dept
     }
 
     process_ndpi_collected_info(ndpi_thread_info[thread_id].workflow, flow);
-    ndpi_thread_info[thread_id].workflow->stats.protocol_counter[flow->detected_protocol.app_protocol]       += flow->packets;
-    ndpi_thread_info[thread_id].workflow->stats.protocol_counter_bytes[flow->detected_protocol.app_protocol] += flow->bytes;
+    ndpi_thread_info[thread_id].workflow->stats.protocol_counter[flow->detected_protocol.app_protocol]       += flow->src2dst_packets + flow->dst2src_packets;
+    ndpi_thread_info[thread_id].workflow->stats.protocol_counter_bytes[flow->detected_protocol.app_protocol] += flow->src2dst_bytes + flow->dst2src_bytes;
     ndpi_thread_info[thread_id].workflow->stats.protocol_flows[flow->detected_protocol.app_protocol]++;
   }
 }
@@ -898,17 +901,20 @@ void updateTopIpAddress(const char *addr, int count, struct info_pair top[], int
 }
 
 /* *********************************************** */
-static void updatePortStats(struct port_stats **stats, u_int32_t port, const char *addr, u_int32_t num_pkts, u_int32_t num_bytes) {
+
+static void updatePortStats(struct port_stats **stats, u_int32_t port,
+			    const char *addr, u_int32_t num_pkts,
+			    u_int32_t num_bytes) {
   struct port_stats *s;
-  int count=0;
+  int count = 0;
   
   HASH_FIND_INT(*stats, &port, s);
   if(s == NULL) {
     s = (struct port_stats*)malloc(sizeof(struct port_stats));
     if(!s) return;
 
-    s->port = port, s->num_pkts = num_pkts, s->num_bytes = num_bytes;
-    s->num_addr = 1, s->cumulative_addr = 1;
+    s->port = port, s->num_pkts = num_pkts, s->num_bytes = num_bytes,
+      s->num_addr = 1, s->cumulative_addr = 1;
 
     memset(s->top_ip_addrs, 0, MAX_NUM_IP_ADDRESS*sizeof(struct info_pair));
     updateTopIpAddress(addr, 1, s->top_ip_addrs, MAX_NUM_IP_ADDRESS);
@@ -958,15 +964,14 @@ static void deletePortsStats(struct port_stats *stats) {
 static void port_stats_walker(const void *node, ndpi_VISIT which, int depth, void *user_data) {
   struct ndpi_flow_info *flow = *(struct ndpi_flow_info **) node;
   u_int16_t sport, dport;
-  char saddr[48];
-  char daddr[48];
+  char saddr[48], daddr[48];
 
   sport = ntohs(flow->lower_port), dport = ntohs(flow->upper_port);
   strncpy(saddr, flow->lower_name, sizeof(saddr));
   strncpy(daddr, flow->upper_name, sizeof(daddr));
 
-  updatePortStats(&srcStats, sport, saddr, flow->packets, flow->bytes);
-  updatePortStats(&dstStats, dport, daddr, flow->packets, flow->bytes);
+  updatePortStats(&srcStats, sport, saddr, flow->src2dst_packets, flow->src2dst_bytes);
+  if(flow->dst2src_packets > 0) updatePortStats(&dstStats, dport, daddr, flow->dst2src_packets, flow->dst2src_bytes);
 }
 
 /* *********************************************** */
@@ -1257,7 +1262,7 @@ static int getTopStats(struct top_stats **topStats, struct port_stats *stats, u_
 
     s->port = sp->port;
     s->num_pkts = sp->num_pkts; 
-    s->prcnt_pkt = (sp->num_pkts*100.0)/total_packet_count; 
+    s->prcnt_pkt = (sp->num_pkts*100.0)/total_packet_count;
     s->num_addr = sp->num_addr;
 
     qsort(&sp->top_ip_addrs[0], MAX_NUM_IP_ADDRESS, sizeof(struct info_pair), info_pair_cmp);
diff --git a/example/ndpi_util.c b/example/ndpi_util.c
index e90400b8f..ef4353e81 100644
--- a/example/ndpi_util.c
+++ b/example/ndpi_util.c
@@ -511,7 +511,12 @@ static struct ndpi_proto packet_processing(struct ndpi_workflow * workflow,
     workflow->stats.total_wire_bytes += rawsize + 24 /* CRC etc */,
       workflow->stats.total_ip_bytes += rawsize;
     ndpi_flow = flow->ndpi_flow;
-    flow->packets++, flow->bytes += rawsize;
+
+    if(src_to_dst_direction)
+      flow->src2dst_packets++, flow->src2dst_bytes += rawsize;
+    else
+      flow->dst2src_packets++, flow->dst2src_bytes += rawsize;
+    
     flow->last_seen = time;
   } else { // flow is NULL
     workflow->stats.total_discarded_bytes++;
@@ -526,8 +531,8 @@ static struct ndpi_proto packet_processing(struct ndpi_workflow * workflow,
 							  ipsize, time, src, dst);
 
   if((flow->detected_protocol.app_protocol != NDPI_PROTOCOL_UNKNOWN)
-     || ((proto == IPPROTO_UDP) && (flow->packets > 8))
-     || ((proto == IPPROTO_TCP) && (flow->packets > 10))) {
+     || ((proto == IPPROTO_UDP) && ((flow->src2dst_packets + flow->dst2src_packets) > 8))
+     || ((proto == IPPROTO_TCP) && ((flow->src2dst_packets + flow->dst2src_packets) > 10))) {
     /* New protocol detected or give up */
     flow->detection_completed = 1;
 
diff --git a/example/ndpi_util.h b/example/ndpi_util.h
index d7fdd9928..f1807e56a 100644
--- a/example/ndpi_util.h
+++ b/example/ndpi_util.h
@@ -57,8 +57,8 @@ typedef struct ndpi_flow_info {
   char lower_name[48], upper_name[48];
   u_int8_t ip_version;
   u_int64_t last_seen;
-  u_int64_t bytes;
-  u_int32_t packets;
+  u_int64_t src2dst_bytes, dst2src_bytes;
+  u_int32_t src2dst_packets, dst2src_packets;
   
   // result only, not used for flow identification
   ndpi_protocol detected_protocol;