Refactor concurrency in govarnishlog.

cmd/govarnishlog/govarnishlog.go

@@ -38,6 +38,7 @@ import (
 	"os/signal"
 	"runtime"
 	"runtime/pprof"
+	"sync"
 	"syscall"
 	"time"
 
@@ -64,6 +65,7 @@ var (
 	totIdle      = time.Duration(0)
 	tRdrStart    time.Time
 	seen         = uint64(0)
+	writes       = uint64(0)
 	eol          = uint64(0)
 	printTxFunc  = printTxTerse
 	printRecFunc = printRecTerse
@@ -87,13 +89,22 @@ func newIdle(seen uint64, seenLast uint64, t time.Time,
 }
 
 func txReader(q *log.Query, txChan chan []log.Tx, statusChan chan log.Status,
-	sigChan chan os.Signal) {
+	stopChan chan struct{}, wg *sync.WaitGroup) {
 
+	defer wg.Done()
 	seenLast := uint64(0)
 	idle := maxIdle
 	tRdrStart = time.Now()
 	tLast := tRdrStart
 	for {
+		select {
+		case <-stopChan:
+			close(txChan)
+			return
+		default:
+			break
+		}
+
 		txGrp, status := q.NextTxGroup()
 		if status == log.EOL {
 			eol++
@@ -167,6 +178,14 @@ func print(txGrp []log.Tx) {
 	out.Flush()
 }
 
+func txWriter(txChan chan []log.Tx, wg *sync.WaitGroup) {
+	defer wg.Done()
+	for txGrp := range txChan {
+		print(txGrp)
+		writes++
+	}
+}
+
 func main() {
 	flag.Parse()
 
@@ -203,6 +222,7 @@ func main() {
 	statusChan := make(chan log.Status)
 	signalChan := make(chan os.Signal, 1)
 	signal.Notify(signalChan, syscall.SIGTERM, syscall.SIGINT)
+	stopChan := make(chan struct{})
 
 	if *cpuprof != "" {
 		f, err := os.Create(*cpuprof)
@@ -217,25 +237,22 @@ func main() {
 		defer pprof.StopCPUProfile()
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(2)
 	start := time.Now()
-	go txReader(q, txChan, statusChan, signalChan)
-
-	n := uint64(0)
+	go txReader(q, txChan, statusChan, stopChan, &wg)
+	go txWriter(txChan, &wg)
 
-LOOP:
-	for {
-		select {
-		case status := <-statusChan:
-			fmt.Fprintln(os.Stderr, "status:", status)
-			break LOOP
-		case sig := <-signalChan:
-			fmt.Fprintln(os.Stderr, "received signal:", sig)
-			break LOOP
-		case txGrp := <-txChan:
-			print(txGrp)
-			n++
-		}
+	select {
+	case status := <-statusChan:
+		fmt.Fprintln(os.Stderr, "status:", status)
+		break
+	case sig := <-signalChan:
+		fmt.Fprintln(os.Stderr, "received signal:", sig)
+		stopChan <- struct{}{}
 	}
+	wg.Wait()
+
 	stop := time.Now()
 	out.Flush()
 
@@ -256,7 +273,7 @@ LOOP:
 		fmt.Fprintf(os.Stderr, "%v wall clock time\n",
 			stop.Sub(start))
 		fmt.Fprintf(os.Stderr, "%d tx grps read\n", seen)
-		fmt.Fprintf(os.Stderr, "%d tx grps written\n", n)
+		fmt.Fprintf(os.Stderr, "%d tx grps written\n", writes)
 		fmt.Fprintf(os.Stderr, "%d r/w channel high watermark\n",
 			chanHi)
 		fmt.Fprintf(os.Stderr, "%d times reader at eol\n", eol)