Fix input/output deadlock condition

If the rate of input and output are very high, the message queue size
can trend towards zero. In this case, it's possible that the Input will
decide to evict from the queue, but by the time the eviction occurs the
output has drained the queue to zero, causing the eviction dequeue to
block forever. In this case, Input is now blocked trying to evict, and
Output is blocked waiting for Input.

Fix the deadlock by distinguishing between eviction of old messages and
dropping of new messages, and making those operations non-blocking.

Add a test which verifies the fix under these conditions.

Author: ironcladlou

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,
 	})
 

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))
 	}
 }

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)
+}