From cd534af43085f89fcbefc5a427a304d470796cc1 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Fri, 18 Jan 2019 12:57:15 -0600
Subject: [PATCH] prevent deadlock in replication logic on reopening a store

Depending on where in the replicate() loop you are when a
store is closed or reassigned, it's possible for it to deadlock.
The deadlock would be that replicate has just successfully read an
entry from your PrimaryTranslateStore.Reader, when a new
PrimaryTranslateStore event happens. Then handlePrimaryTranslateStore
grabs the mutex, signals that the replication handler should
close, and waits for the replication handler to close. Meanwhile,
the replicate() loop tries to grab the mutex... and deadlocks.

Solution: Make the replicate() loop part that needs the mutex
a goroutine that signals on a channel, so we can put it in a select
along with checking for the replicationClosing signal (or the
context terminating). If one of those happens, replicate()
terminates, allowing monitorReplication() to return, which
causes the anonymous function which called it to call
repWG.Done(), allowing handlePrimaryTranslateStore to continue
and eventually release the mutex. At some later point, appendEntry
succeeds or fails, dumps its result status in a buffered
channel, and exits, and the buffered channel is garbage collected.

This is way simpler than it sounds, but it took me a while
to figure out how simple it was.
---
 translate.go | 35 +++++++++++++++++++++++++++--------
 1 file changed, 27 insertions(+), 8 deletions(-)

diff --git a/translate.go b/translate.go
index b86041e3d..db129cd8f 100644
--- a/translate.go
+++ b/translate.go
@@ -372,7 +372,6 @@ func (s *TranslateFile) monitorReplication() {
 		if err := s.replicate(ctx); err != nil {
 			s.logger.Printf("pilosa: replication error: %s", err)
 		}
-
 		select {
 		case <-ctx.Done():
 			return
@@ -412,22 +411,42 @@ func (s *TranslateFile) replicate(ctx context.Context) error {
 	// Wrap in bufferred I/O so it implements io.ByteReader.
 	bufr := bufio.NewReader(rc)
 
+	// we need a way to make an asynchronous routine hand us back an error,
+	// but we might not still be there to get it. so we have a buffer.
+	chErr := make(chan error, 1)
+
 	// Continually read new entries from primary and append to local store.
 	for {
 		// Read next available entry.
 		var entry LogEntry
-		if _, err := entry.ReadFrom(bufr); err == io.EOF {
+		if _, err = entry.ReadFrom(bufr); err == io.EOF {
 			return nil
 		} else if err != nil {
 			return err
 		}
-		s.mu.Lock()
-		// Write to local store.
-		if err := s.appendEntry(&entry); err != nil {
-			s.mu.Unlock()
-			return err
+		// note: we should never end up spawning two of this goroutine
+		// at once. either we end up reading the error from chErr below,
+		// and this loop continues, or we don't, and the whole function
+		// returns. if the function returns, we can write that single
+		// error to the empty channel with a buffer of 1, the goroutine
+		// terminates, and chErr becomes garbage-collectable.
+		go func() {
+			s.mu.Lock()
+			defer s.mu.Unlock()
+			// Write to local store.
+			err = s.appendEntry(&entry)
+			chErr <- err
+		}()
+		select {
+		case err = <-chErr:
+			if err != nil {
+				return err
+			}
+		case <-s.replicationClosing:
+			return nil
+		case <-ctx.Done():
+			return nil
 		}
-		s.mu.Unlock()
 	}
 }