Merge pull request #6449 from ironcladlou/logshifter-wedge-fix

Abhishek Gupta · web-flow · commit 7fd191093ac5 · 2017-07-26T11:39:03.000-07:00
Fix input/output deadlock condition
diff --git a/logshifter/core_test.go b/logshifter/core_test.go
@@ -91,7 +91,7 @@ func TestDropOnBlockedOutput(t *testing.T) {
 	// and the writer is unblocked.
 	ag.AssertStatsEqual(t, map[string]float64{
 		"input.read":   1000,
-		"input.drop":   998,
+		"input.evict":  998,
 		"output.write": 2,
 	})
 
diff --git a/logshifter/input.go b/logshifter/input.go
@@ -14,7 +14,9 @@ import (
 // be placed on the channel:
 //
 //   input.read => number of messages processed during a Read call
-//   input.drop => number of messages dropped from queue during a Read call
+//   input.queue => number of messages queued during a Read call
+//   input.drop => number of new messages dropped during a Read call
+//   input.evict => number of old messages evicted from the queue during a Read call
 //   input.read.duration => time taken to process a line from reader (micros)
 type Input struct {
 	bufferSize   int
@@ -40,17 +42,20 @@ func (input *Input) Read() *sync.WaitGroup {
 	return wg
 }
 
+func (input *Input) record(name string, val float64) {
+	if input.statsChannel != nil {
+		input.statsChannel <- Stat{name: name, value: val}
+	}
+}
+
 // Private synchronous portion of Read.
 func (input *Input) read() {
 	reader := bufio.NewReaderSize(input.reader, input.bufferSize)
 
 	for {
 		line, _, err := reader.ReadLine()
 
-		var start time.Time
-		if input.statsChannel != nil {
-			start = time.Now()
-		}
+		start := time.Now()
 
 		if err != nil {
 			break
@@ -64,24 +69,39 @@ func (input *Input) read() {
 
 		copy(cp, line)
 
-		if input.statsChannel != nil {
-			input.statsChannel <- Stat{name: "input.read", value: 1.0}
-		}
+		input.record("input.read", 1.0)
 
 		select {
 		case input.queue <- cp:
 			// queued
+			input.record("input.queue", 1.0)
 		default:
-			// evict the oldest entry to make room
-			<-input.queue
-			if input.statsChannel != nil {
-				input.statsChannel <- Stat{name: "input.drop", value: 1.0}
+			// try to evict the oldest entry to make room
+			select {
+			case <-input.queue:
+				input.record("input.evict", 1.0)
+				// try again to queue
+				select {
+				case input.queue <- cp:
+					// queued
+					input.record("input.queue", 1.0)
+				default:
+					// no room, drop it
+					input.record("input.drop", 1.0)
+				}
+			default:
+				// queue is already empty, try to queue
+				select {
+				case input.queue <- cp:
+					// queued
+					input.record("input.queue", 1.0)
+				default:
+					// no room, drop it
+					input.record("input.drop", 1.0)
+				}
 			}
-			input.queue <- cp
 		}
 
-		if input.statsChannel != nil {
-			input.statsChannel <- Stat{name: "input.read.duration", value: float64(time.Now().Sub(start).Nanoseconds()) / float64(1000)}
-		}
+		input.record("input.read.duration", float64(time.Now().Sub(start).Nanoseconds())/float64(1000))
 	}
 }
diff --git a/logshifter/input_test.go b/logshifter/input_test.go
@@ -0,0 +1,69 @@
+package main
+
+import (
+	"bytes"
+	"strings"
+	"sync"
+	"testing"
+)
+
+// TestInputBlocking ensures that Input never blocks, regardless of the queue
+// consumption rate.
+func TestInputBlocking(t *testing.T) {
+	messageCount := int64(10000)
+	messageSize := 1024
+
+	// Use a small queue to help ensure an input backup.
+	queue := make(chan []byte, 1)
+
+	// Set up a fast output sink to help ensure an input backup.
+	go func() {
+		for {
+			<-queue
+		}
+	}()
+
+	// Collect stats.
+	stats := make(map[string]float64)
+	statsCh := make(chan Stat)
+	stop := make(chan struct{})
+	statsWg := sync.WaitGroup{}
+	statsWg.Add(1)
+	go func() {
+		for {
+			select {
+			case <-stop:
+				statsWg.Done()
+				return
+			case s := <-statsCh:
+				stats[s.name] = stats[s.name] + s.value
+			}
+		}
+	}()
+
+	// Create an input Reader.
+	var data string
+	var i int64 = 0
+	for ; i < messageCount; i++ {
+		data += strings.Repeat("0", messageSize-1) + "\n"
+	}
+	buffer := bytes.NewBufferString(data)
+	t.Logf("created %d byte test input (%d lines @ %d bytes each)\n", buffer.Len(), messageCount, messageSize)
+
+	// Read until the test reader closes.
+	input := &Input{
+		reader:       buffer,
+		queue:        queue,
+		bufferSize:   messageSize,
+		statsChannel: statsCh,
+	}
+	input.read()
+
+	// If execution reaches this point, the Input hasn't wedged and the test is
+	// successful.
+
+	// Shut down the stats collector.
+	close(stop)
+	statsWg.Wait()
+	t.Logf("stats: %#v", stats)
+}
