Last unexpected service behavior

bin/bin.go

@@ -51,6 +51,7 @@ func LZRMain() {
 	incomingDone.Add(options.Workers)
     done := false
 	writing := false
+	timeoutDone := true
 
     // record to file
     go func() {
@@ -81,7 +82,15 @@ func LZRMain() {
 				var intervalLoop = options.Timeout*lzr.NumHandshakes()*2
 				go func() {
 					for {
-						time.Sleep(time.Duration(intervalLoop)*time.Second)
+						startTime := time.Now()
+			                        for time.Since(startTime) < time.Duration(intervalLoop)*time.Second {
+			                            if (ipMeta.HasUpdates()) {
+			                                startTime = time.Now()
+			                                ipMeta.ResetUpdates()
+			                                continue
+			                            }
+			                            time.Sleep(time.Duration(intervalLoop)*time.Millisecond)
+			                        }
 						if (ipMetaSize == ipMeta.Count()) {
 							fmt.Fprintln(os.Stderr,"Infinite Loop, Breaking.")
 							infiniteLoop = true
@@ -159,6 +168,7 @@ func LZRMain() {
                         timeoutIncoming <- input
                         continue
                     }
+		    timeoutDone = false
                     lzr.HandleTimeout( options, input, &ipMeta, timeoutQueue, retransmitQueue, writingQueue )
                     ipMeta.FinishProcessing( input )
 		    }
@@ -181,7 +191,17 @@ func LZRMain() {
 			//closing file
 			f.F.Flush()
 			t := time.Now()
+	       		startTime := time.Now()
+		        timeoutDone = true
+	       		// 5 second repeated check if any services have called handleTimeout
+		    	for time.Since(startTime) < time.Duration(5)*time.Second {
+				if !(timeoutDone) {
+			    		startTime = time.Now()
+			    		timeoutDone = true
+				}
+		    	}
 			elapsed := t.Sub(start)
+
 			lzr.Summarize( elapsed )
             return
        }

concurrentMap.go

@@ -39,6 +39,7 @@ type pState []*pStateShared
 type pStateShared struct {
 	items        map[string]*packet_state
 	sync.RWMutex // Read Write mutex, guards access to internal map.
+	updated		bool
 }
 
 // Creates a new concurrent map.
@@ -61,6 +62,7 @@ func (m pState) Insert(key string, p * packet_state) {
 	shard.Lock()
 
 	shard.items[key] = p
+	shard.updated = true
 	shard.Unlock()
 }
 
@@ -112,9 +114,33 @@ func (m pState) Remove(key string) {
 	shard := m.GetShard(key)
 	shard.Lock()
 	delete(shard.items, key)
+	shard.updated = true
 	shard.Unlock()
 }
 
+// HasUpdates checks if map has been updated
+func (m pState) HasUpdates() bool {
+    for i := 0; i < SHARD_COUNT; i++ {
+        shard := m[i]
+        shard.RLock()
+        if shard.updated {
+            shard.RUnlock()
+            return true
+        }
+        shard.RUnlock()
+    }
+    return false
+}
+
+// ResetUpdates resets all updates to false
+func (m pState) ResetUpdates() {
+    for i := 0; i < SHARD_COUNT; i++ {
+        shard := m[i]
+        shard.Lock()
+        shard.updated = false
+        shard.Unlock()
+    }
+}
 
 /* FOR PACKET_METADATA */
 //is Processing for goPackets