Avoid deadlocks when closing follower

follower/follower.go

@@ -40,16 +40,17 @@ type Config struct {
 }
 
 type Follower struct {
-	once     sync.Once
-	file     *os.File
-	filename string
-	lines    chan Line
-	err      error
-	config   Config
-	reader   *bufio.Reader
-	watcher  *fsnotify.Watcher
-	offset   int64
-	closeCh  chan struct{}
+	once      sync.Once
+	file      *os.File
+	filename  string
+	lines     chan Line
+	err       error
+	config    Config
+	reader    *bufio.Reader
+	watcher   *fsnotify.Watcher
+	offset    int64
+	closeCh   chan struct{}
+	closeOnce sync.Once
 }
 
 func New(filename string, config Config) (*Follower, error) {
@@ -79,7 +80,7 @@ func (t *Follower) Err() error {
 }
 
 func (t *Follower) Close() {
-	t.closeCh <- struct{}{}
+	t.closeOnce.Do(func() { close(t.closeCh) })
 }
 
 func (t *Follower) run() {
@@ -149,7 +150,10 @@ func (t *Follower) follow() error {
 				break
 			}
 
-			t.sendLine(s, discarded)
+			if !t.sendLine(s, discarded) {
+				t.watcher.Remove(t.filename)
+				return nil
+			}
 		}
 
 		// we're now at EOF, so wait for changes
@@ -243,7 +247,16 @@ func (t *Follower) follow() error {
 
 func (t *Follower) rewatch() error {
 	t.watcher.Remove(t.filename)
-	if err := t.reopen(); err != nil {
+
+	// After a rename the new file may not exist yet, retry for up to 1 minute
+	var err error
+	for i := 0; i < 20; i++ {
+		if err = t.reopen(); err == nil || !os.IsNotExist(err) {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	if err != nil {
 		return err
 	}
 
@@ -278,8 +291,13 @@ func (t *Follower) close(err error) {
 	close(t.lines)
 }
 
-func (t *Follower) sendLine(l []byte, d int) {
-	t.lines <- Line{l[:len(l)-1], d}
+func (t *Follower) sendLine(l []byte, d int) bool {
+	select {
+	case t.lines <- Line{l[:len(l)-1], d}:
+		return true
+	case <-t.closeCh:
+		return false
+	}
 }
 
 func (t *Follower) watchFileEvents(eventChan chan fsnotify.Event, errChan chan error) {

follower/follower_test.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"path"
 	"runtime"
-	"sync"
 	"testing"
 	"time"
 
@@ -218,6 +217,54 @@ func TestSymlink(t *testing.T) {
 	assertFollowedLines(t, f, testLines[1])
 }
 
+// Regression test for: Close() deadlocks when follow() is blocked in
+// the inner read loop (ReadBytes waiting for data). The closeCh select
+// is only in the outer loop, so Close() blocks forever on the unbuffered
+// closeCh send.
+func TestCloseDeadlockNoConsumer(t *testing.T) {
+	file := testFile(t, "TestCloseDeadlockNoConsumer")
+	f, err := New(file.Name(), Config{
+		Reopen: true,
+		Offset: 0,
+		Whence: io.SeekEnd,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Write a partial line (no trailing newline). The follower reads it via
+	// ReadBytes and blocks waiting for the newline delimiter. There is no
+	// consumer on f.Lines(), so sendLine would block too, but ReadBytes
+	// blocks first (it must complete before sendLine can be reached).
+	if _, err := file.WriteString("partial line"); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(100 * time.Millisecond)
+
+	// Close() must return within the deadline. Before the fix it deadlocks.
+	done := make(chan struct{})
+	go func() {
+		f.Close()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Close() deadlocked: follow() was blocked in the inner read loop")
+	}
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify the lines channel is closed (confirms run() called close()).
+	select {
+	case _, ok := <-f.Lines():
+		if ok {
+			t.Fatal("Lines channel should be closed after Close()")
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatal("Lines channel was not closed after Close()")
+	}
+}
+
 func testFile(t *testing.T, name string) *os.File {
 	// open in append mode since most loggers will be doing such
 	file, err := os.OpenFile(path.Join(tmpDir, name), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0666)
@@ -267,25 +314,20 @@ func writeLines(file *os.File, lines []string) error {
 func assertFollowedLines(t *testing.T, f *Follower, lines []string) {
 	assert := assert.New(t)
 
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	go func() {
-		defer wg.Done()
-
-		i := 0
-		for line := range f.Lines() {
-			assert.Equal(lines[i], line.String())
-			i++
-			if i == len(lines) {
-				return
-			}
+	i := 0
+	for line := range f.Lines() {
+		assert.Equal(lines[i], line.String())
+		i++
+		if i == len(lines) {
+			return
 		}
-	}()
+	}
 
+	// Lines() closed before all expected lines arrived. Err() is only safe to
+	// read once Lines() is closed, since the follower goroutine sets the error
+	// right before closing the channel.
 	if err := f.Err(); err != nil {
 		t.Fatal(err)
 	}
-
-	wg.Wait()
+	t.Fatalf("lines channel closed early after %d of %d lines", i, len(lines))
 }