From 0dc467f289a6537ce51c87cbdadf994e712cea17 Mon Sep 17 00:00:00 2001
From: Ben Grewell <BGrewell@gmail.com>
Date: Wed, 10 Mar 2021 14:28:20 -0800
Subject: [PATCH] fix issues with reporter closing

---
 .idea/workspace.xml    |   2 +
 reporter.go            |   9 +++
 reporter_linux.go      | 128 +++++++++++++++++++++--------------------
 tests/client/client.go |   5 +-
 4 files changed, 80 insertions(+), 64 deletions(-)

diff --git a/.idea/workspace.xml b/.idea/workspace.xml
index 7aa60f7..2048bf8 100644
--- a/.idea/workspace.xml
+++ b/.idea/workspace.xml
@@ -3,7 +3,9 @@
   <component name="ChangeListManager">
     <list default="true" id="fc2840de-29dc-4fca-8e0e-a283562f60ca" name="Default Changelist" comment="">
       <change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/reporter.go" beforeDir="false" afterPath="$PROJECT_DIR$/reporter.go" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/reporter_linux.go" beforeDir="false" afterPath="$PROJECT_DIR$/reporter_linux.go" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/tests/client/client.go" beforeDir="false" afterPath="$PROJECT_DIR$/tests/client/client.go" afterDir="false" />
     </list>
     <option name="SHOW_DIALOG" value="false" />
     <option name="HIGHLIGHT_CONFLICTS" value="true" />
diff --git a/reporter.go b/reporter.go
index 43a53d1..efae6d4 100644
--- a/reporter.go
+++ b/reporter.go
@@ -1,9 +1,15 @@
 package iperf
 
+import (
+	"fmt"
+	"github.com/BGrewell/tail"
+)
+
 type Reporter struct {
 	ReportingChannel chan *StreamIntervalReport
 	LogFile          string
 	running          bool
+	tailer *tail.Tail
 }
 
 func (r *Reporter) Start() {
@@ -13,7 +19,10 @@ func (r *Reporter) Start() {
 
 func (r *Reporter) Stop() {
 	r.running = false
+	r.tailer.Stop()
+	r.tailer.Cleanup()
 	close(r.ReportingChannel)
+	fmt.Println("reporter stopped")
 }
 
 // runLogProcessor is OS specific because of differences in iperf on Windows and Linux
diff --git a/reporter_linux.go b/reporter_linux.go
index b12a515..30b845a 100644
--- a/reporter_linux.go
+++ b/reporter_linux.go
@@ -7,6 +7,7 @@ import (
 	"log"
 	"strconv"
 	"strings"
+	"time"
 )
 
 /*
@@ -44,9 +45,10 @@ Connecting to host 10.254.100.100, port 5201
 iperf Done.
 
 */
-
+//TODO: NEED TO UPDATE WINDOWS IMPLEMENTATION ALSO!!!!!
 func (r *Reporter) runLogProcessor() {
-	tailer, err := tail.TailFile(r.LogFile, tail.Config{
+	var err error
+	r.tailer, err = tail.TailFile(r.LogFile, tail.Config{
 		Follow:    true,
 		ReOpen:    true,
 		Poll:      true, // on linux we don't need to poll as the fsnotify works properly
@@ -55,69 +57,73 @@ func (r *Reporter) runLogProcessor() {
 	if err != nil {
 		log.Fatalf("failed to tail log file: %v", err)
 	}
-	for line := range tailer.Lines {
-		// TODO: For now this only cares about individual streams it ignores the sum lines
-		if len(line.Text) > 5 {
-			id := line.Text[1:4]
-			stream, err := strconv.Atoi(strings.TrimSpace(id))
-			if err != nil {
-				continue
-			}
-			fields := strings.Fields(line.Text[5:])
-			if len(fields) >= 9 {
-				if fields[0] == "local" {
+
+	for {
+		select {
+			case line := <- r.tailer.Lines:
+				if line == nil {
 					continue
 				}
-				timeFields := strings.Split(fields[0], "-")
-				start, err := strconv.ParseFloat(timeFields[0], 32)
-				if err != nil {
-					log.Printf("failed to convert start time: %s\n", err)
-				}
-				end, err := strconv.ParseFloat(timeFields[1], 32)
-				transferedStr := fmt.Sprintf("%s%s", fields[2], fields[3])
-				transferedBytes, err := conversions.StringBitRateToInt(transferedStr)
-				if err != nil {
-					log.Printf("failed to convert units: %s\n", err)
-				}
-				transferedBytes = transferedBytes / 8
-				rateStr := fmt.Sprintf("%s%s", fields[4], fields[5])
-				rate, err := conversions.StringBitRateToInt(rateStr)
-				if err != nil {
-					log.Printf("failed to convert units: %s\n", err)
-				}
-				retrans, err := strconv.Atoi(fields[6])
-				if err != nil {
-					log.Printf("failed to convert units: %s\n", err)
+				if len(line.Text) > 5 {
+					id := line.Text[1:4]
+					stream, err := strconv.Atoi(strings.TrimSpace(id))
+					if err != nil {
+						continue
+					}
+					fields := strings.Fields(line.Text[5:])
+					if len(fields) >= 9 {
+						if fields[0] == "local" {
+							continue
+						}
+						timeFields := strings.Split(fields[0], "-")
+						start, err := strconv.ParseFloat(timeFields[0], 32)
+						if err != nil {
+							log.Printf("failed to convert start time: %s\n", err)
+						}
+						end, err := strconv.ParseFloat(timeFields[1], 32)
+						transferedStr := fmt.Sprintf("%s%s", fields[2], fields[3])
+						transferedBytes, err := conversions.StringBitRateToInt(transferedStr)
+						if err != nil {
+							log.Printf("failed to convert units: %s\n", err)
+						}
+						transferedBytes = transferedBytes / 8
+						rateStr := fmt.Sprintf("%s%s", fields[4], fields[5])
+						rate, err := conversions.StringBitRateToInt(rateStr)
+						if err != nil {
+							log.Printf("failed to convert units: %s\n", err)
+						}
+						retrans, err := strconv.Atoi(fields[6])
+						if err != nil {
+							log.Printf("failed to convert units: %s\n", err)
+						}
+						cwndStr := fmt.Sprintf("%s%s", fields[7], fields[8])
+						cwnd, err := conversions.StringBitRateToInt(cwndStr)
+						if err != nil {
+							log.Printf("failed to convert units: %s\n", err)
+						}
+						cwnd = cwnd / 8
+						omitted := false
+						if len(fields) >= 10 && fields[9] == "(omitted)" {
+							omitted = true
+						}
+						report := &StreamIntervalReport{
+							Socket:           stream,
+							StartInterval:    float32(start),
+							EndInterval:      float32(end),
+							Seconds:          float32(end - start),
+							Bytes:            int(transferedBytes),
+							BitsPerSecond:    float64(rate),
+							Retransmissions:  retrans,
+							CongestionWindow: int(cwnd),
+							Omitted:          omitted,
+						}
+						r.ReportingChannel <- report
+					}
 				}
-				cwndStr := fmt.Sprintf("%s%s", fields[7], fields[8])
-				cwnd, err := conversions.StringBitRateToInt(cwndStr)
-				if err != nil {
-					log.Printf("failed to convert units: %s\n", err)
+			case <- time.After(1 * time.Second):
+				if !r.running {
+					return
 				}
-				cwnd = cwnd / 8
-				omitted := false
-				if len(fields) >= 10 && fields[9] == "(omitted)" {
-					omitted = true
-				}
-				report := &StreamIntervalReport{
-					Socket:           stream,
-					StartInterval:    float32(start),
-					EndInterval:      float32(end),
-					Seconds:          float32(end - start),
-					Bytes:            int(transferedBytes),
-					BitsPerSecond:    float64(rate),
-					Retransmissions:  retrans,
-					CongestionWindow: int(cwnd),
-					Omitted:          omitted,
-				}
-				r.ReportingChannel <- report
-			}
-		}
-
-		if !r.running {
-			fmt.Println("reporter is finished. exiting")
-			close(r.ReportingChannel)
-			return
 		}
 	}
 }
diff --git a/tests/client/client.go b/tests/client/client.go
index 494549a..20db4f5 100644
--- a/tests/client/client.go
+++ b/tests/client/client.go
@@ -10,8 +10,8 @@ func main() {
 
 	includeServer := true
 	proto := "tcp"
-	runTime := 30
-	omitSec := 10
+	runTime := 10
+	omitSec := 0
 	length := "65500"
 
 	c := iperf.NewClient("10.254.100.100")
@@ -22,7 +22,6 @@ func main() {
 	c.SetLength(length)
 	c.SetJSON(false)
 	c.SetIncludeServer(false)
-	c.SetTimeSec(20)
 	c.SetStreams(2)
 	reports := c.SetModeLive()