Remove stale Pids from lastP

1 files changed, 31 insertions, 11 deletions

diff --git a/gstat/main.go b/gstat/main.go
index 98d89fb..6d61e8b 100644
--- a/gstat/main.go
+++ b/gstat/main.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"container/list"
 	"fmt"
 	"github.com/buetow/gstat/diskstats"
 	"github.com/buetow/gstat/process"
@@ -16,6 +17,8 @@ type twoP struct {
 }
 type processMap map[string]twoP
 
+var interval time.Duration
+
 func timedGather(timerChan <-chan bool, dRxChan chan<- diskstats.Diskstats, pRxChan chan<- process.Process) {
 	for {
 		switch <-timerChan {
@@ -42,15 +45,32 @@ func receiveD(dRxChan <-chan diskstats.Diskstats) {
 	}
 }
 
-func compareP(lastP processMap) {
-	for id, val := range lastP {
-		first := val.first.Count["syscr"]
-		second := val.second.Count["syscr"]
-		diff := first - second
-		if diff < 0 {
-			diff = -diff
+func compareP(lastP *processMap) {
+	removeItems := list.New()
+
+	for id, val := range *lastP {
+		nowTimestamp := int32(time.Now().Unix())
+		if val.first.Timestamp+int32(interval)*2 < nowTimestamp {
+			// Schedule remove obsolete pids from lastP
+			removeItems.PushFront(val.first.Id)
+
+		} else {
+			// Compare
+			first := val.first.Count["syscr"]
+			second := val.second.Count["syscr"]
+			diff := first - second
+			if diff < 0 {
+				diff = -diff
+			}
+			fmt.Printf("%d %s\n", diff, id)
 		}
-		fmt.Printf("%d %s\n", diff, id)
+	}
+
+	// Rremove obsolete pids from lastP
+	for e := removeItems.Front(); e != nil; e = e.Next() {
+		id := e.Value.(string)
+		fmt.Println("STALE: " + id)
+		delete(*lastP, id)
 	}
 }
 
@@ -58,11 +78,10 @@ func receiveP(pRxChan <-chan process.Process) {
 	lastP := make(processMap)
 	flag := false
 
-	// TODO: Cleanup lastP for stale PIDs
 	for p := range pRxChan {
 		if p.Last {
 			if flag {
-				compareP(lastP)
+				compareP(&lastP)
 			}
 			flag = !flag
 		} else {
@@ -83,6 +102,7 @@ func main() {
 	timerChan := make(chan bool)
 	dRxChan := make(chan diskstats.Diskstats)
 	pRxChan := make(chan process.Process)
+	interval = 2
 
 	go timedGather(timerChan, dRxChan, pRxChan)
 	go receiveD(dRxChan)
@@ -101,6 +121,6 @@ func main() {
 
 	for {
 		timerChan <- true
-		time.Sleep(time.Second * 2)
+		time.Sleep(time.Second * interval)
 	}
 }