Update IPFIXLIB to fix some annoying bugs relating to generation of bandwith...

Update IPFIXLIB to fix some annoying bugs relating to generation of bandwith stats, and increase performance in bandwidth calculation, havard.mork@gmail.com

git-svn-id: file:///home/svn/mapi/branches/haavardm-mapi@46 8d5bb341-7cf1-0310-8cf6-ba355fef3186

ipfixlib/engine.c

@@ -395,7 +395,8 @@ static u_int getMSDistance(unsigned long long new, unsigned long long old) {
 
 static void calculateBitrate(u_int divisor, u_int64_t stamp, u_int64_t *lastupdate, 
 			     u_int32_t *rate_best_max, u_int32_t *rate_best_min,
-			     u_int32_t *averager, u_int len) {
+			     u_int32_t *averager, u_int16_t *avgpos, u_int len,
+			     u_int force) {
   /* 'divisor' is the number of milliseconds that one slot consists of. */
   /*    i.e. for a total 1sec interval, there are 10x 100ms-intervals, which */
   /*    implies value of 100 expected */
@@ -409,11 +410,10 @@ static void calculateBitrate(u_int divisor, u_int64_t stamp, u_int64_t *lastupda
 
   u_int i;
   u_int numSlotsAdvance = getMSDistance(stamp,*lastupdate)/divisor;
-  const u_int lastslot = BITRATE_AVERAGER_SLOTS-1;
   u_int32_t mult = 100/divisor;
   
   // First: If advancing, then sum up and check for record first.
-  if(numSlotsAdvance > 0) {
+  if(numSlotsAdvance > 0 || force) {
     u_int32_t sum = 0;
     for(i=0; i<BITRATE_AVERAGER_SLOTS; i++) {
       sum += averager[i];
@@ -428,26 +428,43 @@ static void calculateBitrate(u_int divisor, u_int64_t stamp, u_int64_t *lastupda
     sum *= mult;
     if(sum < *rate_best_min)
       *rate_best_min = sum;
-    else if(sum!=MAX_UINT32 && sum > *rate_best_max)
+    if(sum!=MAX_UINT32 && sum > *rate_best_max)
       *rate_best_max = sum;
 
+    /* If 'force', then all we wanted was to update stats */
+    if(force)
+      return;
+
     *lastupdate = stamp;
   }
 
   if(numSlotsAdvance >= BITRATE_AVERAGER_SLOTS) {
     // More time than in the entire averager-block has passed. Invalidate all.
     memset(averager, '\0', sizeof(u_int32_t)*BITRATE_AVERAGER_SLOTS);
+    *rate_best_min = 0; // it's definitively a minimum
+    *avgpos = 0;
   }
   else if(numSlotsAdvance > 0) {
+
+    while(numSlotsAdvance > 0) {
+      *avgpos++;
+      if(*avgpos >= BITRATE_AVERAGER_SLOTS)
+	*avgpos = 0;
+
+      averager[*avgpos] = 0;
+      numSlotsAdvance--;
+    }    
+
     // Advance a given number of slots
-    memmove(averager,averager + numSlotsAdvance, sizeof(u_int32_t)*(BITRATE_AVERAGER_SLOTS-numSlotsAdvance));
-    memset(averager + (BITRATE_AVERAGER_SLOTS-numSlotsAdvance), '\0', numSlotsAdvance*sizeof(u_int32_t));
+    //memmove(averager,averager + numSlotsAdvance, sizeof(u_int32_t)*(BITRATE_AVERAGER_SLOTS-numSlotsAdvance));
+    //memset(averager + (BITRATE_AVERAGER_SLOTS-numSlotsAdvance), '\0', numSlotsAdvance*sizeof(u_int32_t));
   }
-  averager[lastslot] += len;
+  averager[*avgpos] += len;
 } 
 
 static void updateBitrateCalculation(np_ctxt_t *npctxt, HashBucket *bkt, u_int len, 
-				     unsigned long long stamp, u_int direction) {
+				     unsigned long long stamp, u_int direction,
+				     u_int force) {
   // Calculate average bitrate of a period of time
   // This is done by continuously updating the bkt->bitrateAverager1sec array
   // and summing this at the end of each time slot (which is .1 secs)
@@ -460,33 +477,39 @@ static void updateBitrateCalculation(np_ctxt_t *npctxt, HashBucket *bkt, u_int l
     if(direction==0)
       calculateBitrate(100, stamp, &bkt->src2dstBitrateLastUpdate1sec,
 		       &bkt->src2dstRateMax1sec, &bkt->src2dstRateMin1sec,
-		       bkt->src2dstBitrateAverager1sec,len);
+		       bkt->src2dstBitrateAverager1sec,&bkt->src2dstBitrateAverager1secPos,
+		       len, force);
     else
       calculateBitrate(100, stamp, &bkt->dst2srcBitrateLastUpdate1sec,
 		       &bkt->dst2srcRateMax1sec, &bkt->dst2srcRateMin1sec,
-		       bkt->dst2srcBitrateAverager1sec,len);
+		       bkt->dst2srcBitrateAverager1sec,&bkt->dst2srcBitrateAverager1secPos,
+		       len, force);
   }
 
   if((npctxt->bitrateCalcEnabled & BITRATECALC_100MS) != 0) {
     if(direction==0)
       calculateBitrate(10, stamp, &bkt->src2dstBitrateLastUpdate100ms,
 		       &bkt->src2dstRateMax100ms, &bkt->src2dstRateMin100ms,
-		       bkt->src2dstBitrateAverager100ms,len);
+		       bkt->src2dstBitrateAverager100ms,&bkt->src2dstBitrateAverager100msPos,
+		       len, force);
     else
       calculateBitrate(10, stamp, &bkt->dst2srcBitrateLastUpdate100ms,
 		       &bkt->dst2srcRateMax100ms, &bkt->dst2srcRateMin100ms,
-		       bkt->dst2srcBitrateAverager100ms,len);
+		       bkt->dst2srcBitrateAverager100ms,&bkt->dst2srcBitrateAverager100msPos,
+		       len, force);
   }
 
   if((npctxt->bitrateCalcEnabled & BITRATECALC_10MS) != 0) {
     if(direction==0)
       calculateBitrate(1, stamp, &bkt->src2dstBitrateLastUpdate10ms,
 		       &bkt->src2dstRateMax10ms,&bkt->src2dstRateMin10ms,
-		       bkt->src2dstBitrateAverager10ms,len);
+		       bkt->src2dstBitrateAverager10ms,&bkt->src2dstBitrateAverager10msPos,
+		       len, force);
     else
       calculateBitrate(1, stamp, &bkt->dst2srcBitrateLastUpdate10ms,
 		       &bkt->dst2srcRateMax10ms, &bkt->dst2srcRateMin10ms,
-		       bkt->dst2srcBitrateAverager10ms,len);
+		       bkt->dst2srcBitrateAverager10ms,&bkt->dst2srcBitrateAverager10msPos,
+		       len, force);
   }
 
 }
@@ -637,7 +660,7 @@ void addPktToHash(np_ctxt_t *npctxt,
 	bkt->src2dstTcpFlags |= flags; /* Do not move this line before updateTcpFlags(...) */
 	updatePacketSizeStats(npctxt,bkt,len,0);
 	if(npctxt->bitrateCalcEnabled!=0) {	
-	  updateBitrateCalculation(npctxt,bkt,len, stamp,0);
+	  updateBitrateCalculation(npctxt,bkt,len, stamp,0,0);
 	}
 
 	bkt->src2dstOctetDeltaCount = bkt->src2dstOctetDeltaCount + len;
@@ -662,7 +685,7 @@ void addPktToHash(np_ctxt_t *npctxt,
 	bkt->dst2srcTcpFlags |= flags; /* Do not move this line before updateTcpFlags(...) */
 	updatePacketSizeStats(npctxt,bkt,len,1);
 	if(npctxt->bitrateCalcEnabled!=0) {	
-	  updateBitrateCalculation(npctxt,bkt,len, stamp,1);
+	  updateBitrateCalculation(npctxt,bkt,len, stamp,1,0);
 	}
 
 	bkt->dst2srcOctetDeltaCount = bkt->dst2srcOctetDeltaCount + len;
@@ -767,11 +790,19 @@ void addPktToHash(np_ctxt_t *npctxt,
   bkt->dst2srcRateMax10ms  = 0;
   bkt->dst2srcRateMin10ms  = MAX_UINT32;
 
+  bkt->src2dstBitrateAverager1secPos = 0;
+  bkt->dst2srcBitrateAverager1secPos = 0;
+  bkt->src2dstBitrateAverager100msPos = 0;
+  bkt->dst2srcBitrateAverager100msPos = 0;
+  bkt->src2dstBitrateAverager10msPos = 0;
+  bkt->dst2srcBitrateAverager10msPos = 0;
+
+
   bkt->src2dstMinPktSize = bkt->dst2srcMinPktSize = 0xFFFF;
   bkt->src2dstMaxPktSize = bkt->dst2srcMaxPktSize = 0x0000;
   updatePacketSizeStats(npctxt,bkt,len,0);
   if(npctxt->bitrateCalcEnabled!=0) {	
-    updateBitrateCalculation(npctxt,bkt,len,stamp,0);
+    updateBitrateCalculation(npctxt,bkt,len,stamp,0,0);
   }
 
   if(src.ipVersion == 4 && proto == IPPROTO_TCP)
@@ -1006,6 +1037,12 @@ void walkHash(np_ctxt_t *npctxt, int flushHash) {
 #endif
 
       myNextBucket = myBucket->next;
+
+      // Flush bitrate calculation before proceeding
+      if(npctxt->bitrateCalcEnabled!=0) {	
+	updateBitrateCalculation(npctxt,myBucket,0,0,0,1);
+      }
+
       queueBucketToExport(npctxt, myBucket);
       numBucketsExported++;
       if(numBucketsExported >= npctxt->minNumFlowsPerPacket) {

ipfixlib/nprobe_bucket.h

@@ -138,6 +138,9 @@ typedef struct hashBucket {
   u_int32_t src2dstRateMin100ms, dst2srcRateMin100ms;
   u_int32_t src2dstRateMin10ms,  dst2srcRateMin10ms;
 
+  u_int16_t src2dstBitrateAverager1secPos,  dst2srcBitrateAverager1secPos;
+  u_int16_t src2dstBitrateAverager100msPos, dst2srcBitrateAverager100msPos;
+  u_int16_t src2dstBitrateAverager10msPos,  dst2srcBitrateAverager10msPos;